348 lines
13 KiB
Python
348 lines
13 KiB
Python
import ast
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import psycopg
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import settings
|
|
from app.dashboard.dependencies import get_dashboard_service
|
|
from app.dashboard.models import (
|
|
DashboardAccess,
|
|
DashboardOverviewData,
|
|
DashboardOverviewResponse,
|
|
)
|
|
from app.dashboard.router import router
|
|
from app.dashboard.service import DashboardService
|
|
from app.identity import get_current_user
|
|
from app.inventory.models import (
|
|
InventoryFleetAttentionSummary,
|
|
InventoryFleetSummary,
|
|
)
|
|
from app.inventory.repository import InventoryRepository
|
|
from app.locations.models import LocationSummary
|
|
from app.locations.repository import LocationMapRepository
|
|
from app.repairs.center_repository import RepairCenterRepository
|
|
from app.repairs.models import ActiveRepairSummary
|
|
|
|
|
|
NOW = datetime(2026, 8, 11, 12, 0, tzinfo=timezone.utc)
|
|
APP_ROOT = Path(__file__).parents[1] / "app"
|
|
|
|
|
|
def fleet_summary() -> InventoryFleetSummary:
|
|
return InventoryFleetSummary(
|
|
total=12,
|
|
runtime_online=7,
|
|
runtime_offline=3,
|
|
runtime_unknown=2,
|
|
lifecycle_in_repair=2,
|
|
lifecycle_spare=1,
|
|
lifecycle_retired=1,
|
|
attention=InventoryFleetAttentionSummary(
|
|
offline=2,
|
|
unknown=1,
|
|
unstable=1,
|
|
repair=2,
|
|
),
|
|
)
|
|
|
|
|
|
class FakeInventoryService:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def summarize_fleet(self):
|
|
self.calls += 1
|
|
return fleet_summary()
|
|
|
|
|
|
class FakeRepairService:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def summarize_active_repairs(self):
|
|
self.calls += 1
|
|
return ActiveRepairSummary(
|
|
active=3,
|
|
status_counts={"OPEN": 2, "waiting_parts": 1},
|
|
)
|
|
|
|
|
|
class FakeLocationService:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def summarize_locations(self, *, include_miner_counts):
|
|
self.calls.append(include_miner_counts)
|
|
return LocationSummary(
|
|
sites=2,
|
|
rooms=4,
|
|
positions=20,
|
|
occupied_positions=12,
|
|
empty_positions=8,
|
|
mapped_miners=12 if include_miner_counts else None,
|
|
unmapped_miners=0 if include_miner_counts else None,
|
|
)
|
|
|
|
|
|
class DashboardServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.inventory = FakeInventoryService()
|
|
self.repairs = FakeRepairService()
|
|
self.locations = FakeLocationService()
|
|
self.service = DashboardService(
|
|
self.inventory,
|
|
self.repairs,
|
|
self.locations,
|
|
)
|
|
|
|
def test_composes_each_authorized_domain_once(self):
|
|
result = self.service.get_overview(
|
|
DashboardAccess(
|
|
can_view_miners=True,
|
|
can_view_repairs=True,
|
|
can_view_locations=True,
|
|
)
|
|
)
|
|
self.assertEqual(result.data.fleet.total, 12)
|
|
self.assertEqual(result.data.repairs.active, 3)
|
|
self.assertEqual(result.data.locations.mapped_miners, 12)
|
|
self.assertEqual(self.inventory.calls, 1)
|
|
self.assertEqual(self.repairs.calls, 1)
|
|
self.assertEqual(self.locations.calls, [True])
|
|
self.assertEqual(
|
|
result.meta.available_sections,
|
|
["fleet", "repairs", "locations"],
|
|
)
|
|
|
|
def test_unauthorized_domains_are_not_called_or_returned(self):
|
|
result = self.service.get_overview(DashboardAccess())
|
|
self.assertIsNone(result.data.fleet)
|
|
self.assertIsNone(result.data.repairs)
|
|
self.assertIsNone(result.data.locations)
|
|
self.assertEqual(result.meta.available_sections, [])
|
|
self.assertEqual(self.inventory.calls, 0)
|
|
self.assertEqual(self.repairs.calls, 0)
|
|
self.assertEqual(self.locations.calls, [])
|
|
|
|
def test_location_only_summary_omits_miner_placement_counts(self):
|
|
result = self.service.get_overview(
|
|
DashboardAccess(can_view_locations=True)
|
|
)
|
|
self.assertEqual(self.locations.calls, [False])
|
|
self.assertIsNone(result.data.locations.mapped_miners)
|
|
self.assertIsNone(result.data.locations.unmapped_miners)
|
|
self.assertIsNone(result.data.fleet)
|
|
|
|
|
|
class StubDashboardService:
|
|
def __init__(self):
|
|
self.access = None
|
|
|
|
def get_overview(self, access):
|
|
self.access = access
|
|
return DashboardOverviewResponse(
|
|
data=DashboardOverviewData(
|
|
fleet=fleet_summary() if access.can_view_miners else None,
|
|
repairs=(
|
|
ActiveRepairSummary(active=1, status_counts={"OPEN": 1})
|
|
if access.can_view_repairs else None
|
|
),
|
|
locations=(
|
|
LocationSummary(
|
|
sites=1,
|
|
rooms=1,
|
|
positions=2,
|
|
occupied_positions=1,
|
|
empty_positions=1,
|
|
mapped_miners=1 if access.can_view_miners else None,
|
|
unmapped_miners=11 if access.can_view_miners else None,
|
|
)
|
|
if access.can_view_locations else None
|
|
),
|
|
generated_at=NOW,
|
|
),
|
|
meta={
|
|
"available_sections": [
|
|
name
|
|
for name, allowed in (
|
|
("fleet", access.can_view_miners),
|
|
("repairs", access.can_view_repairs),
|
|
("locations", access.can_view_locations),
|
|
)
|
|
if allowed
|
|
]
|
|
},
|
|
)
|
|
|
|
|
|
class DashboardRouterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.service = StubDashboardService()
|
|
self.app = FastAPI()
|
|
self.app.include_router(router)
|
|
self.app.dependency_overrides[get_dashboard_service] = lambda: self.service
|
|
self.client = TestClient(self.app)
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
|
|
def authenticate(self, permissions):
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": str(uuid4()),
|
|
"role": "Custom",
|
|
"permissions": list(permissions),
|
|
}
|
|
|
|
def test_requires_authentication_and_dashboard_permission(self):
|
|
self.assertEqual(self.client.get("/api/dashboard/overview").status_code, 401)
|
|
self.authenticate(("miners.view",))
|
|
self.assertEqual(self.client.get("/api/dashboard/overview").status_code, 403)
|
|
|
|
def test_returns_one_typed_permission_aware_envelope(self):
|
|
self.authenticate(("dashboard.view", "miners.view", "locations.view"))
|
|
response = self.client.get("/api/dashboard/overview")
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload["data"]["fleet"]["total"], 12)
|
|
self.assertIsNone(payload["data"]["repairs"])
|
|
self.assertEqual(payload["data"]["locations"]["mapped_miners"], 1)
|
|
self.assertEqual(
|
|
payload["meta"]["available_sections"],
|
|
["fleet", "locations"],
|
|
)
|
|
self.assertEqual(response.headers["cache-control"], "no-store")
|
|
self.assertFalse(self.service.access.can_view_repairs)
|
|
|
|
def test_location_only_response_contains_no_miner_counts(self):
|
|
self.authenticate(("dashboard.view", "locations.view"))
|
|
payload = self.client.get("/api/dashboard/overview").json()
|
|
self.assertIsNone(payload["data"]["fleet"])
|
|
self.assertIsNone(payload["data"]["locations"]["mapped_miners"])
|
|
self.assertIsNone(payload["data"]["locations"]["unmapped_miners"])
|
|
|
|
|
|
class DashboardRepositoryIntegrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.connection = psycopg.connect(
|
|
host=settings.postgres_host,
|
|
port=settings.postgres_port,
|
|
dbname=settings.postgres_db,
|
|
user=settings.postgres_user,
|
|
password=settings.postgres_password,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.connection.rollback()
|
|
self.connection.close()
|
|
|
|
def test_fleet_counts_are_exact_unbounded_and_preserve_overlap(self):
|
|
repository = InventoryRepository(connection=self.connection)
|
|
before = repository.summarize_fleet()
|
|
rows = []
|
|
states = (
|
|
("offline", "in_repair"),
|
|
("online", "spare"),
|
|
("maintenance", "retired"),
|
|
("online", "active"),
|
|
("offline", "active"),
|
|
("unknown", "active"),
|
|
)
|
|
for runtime, lifecycle in states:
|
|
rows.append((uuid4(), runtime, lifecycle))
|
|
bulk = [(uuid4(), "online", "active") for _ in range(501)]
|
|
with self.connection.cursor() as cur:
|
|
cur.executemany(
|
|
"INSERT INTO miners (id, status, lifecycle_status, first_seen_at, last_seen_at) "
|
|
"VALUES (%s, %s, %s, NULL, NULL)",
|
|
[*rows, *bulk],
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO miner_status_snapshots (miner_id, hash_stable) VALUES (%s, false)",
|
|
(rows[3][0],),
|
|
)
|
|
after = repository.summarize_fleet()
|
|
self.assertEqual(after["total"] - before["total"], 507)
|
|
self.assertEqual(after["runtime_online"] - before["runtime_online"], 503)
|
|
self.assertEqual(after["runtime_offline"] - before["runtime_offline"], 2)
|
|
self.assertEqual(after["runtime_unknown"] - before["runtime_unknown"], 2)
|
|
self.assertEqual(after["lifecycle_in_repair"] - before["lifecycle_in_repair"], 1)
|
|
self.assertEqual(after["lifecycle_spare"] - before["lifecycle_spare"], 1)
|
|
self.assertEqual(after["lifecycle_retired"] - before["lifecycle_retired"], 1)
|
|
self.assertEqual(after["attention_repair"] - before["attention_repair"], 1)
|
|
self.assertEqual(after["attention_offline"] - before["attention_offline"], 1)
|
|
self.assertEqual(after["attention_unknown"] - before["attention_unknown"], 1)
|
|
self.assertEqual(after["attention_unstable"] - before["attention_unstable"], 1)
|
|
|
|
def test_location_summary_uses_current_positions_not_virtual_map_subset(self):
|
|
repository = LocationMapRepository(connection=self.connection)
|
|
before = repository.summarize_locations(include_miner_counts=True)
|
|
site_id, room_id = uuid4(), uuid4()
|
|
mapped_id, unmapped_id = uuid4(), uuid4()
|
|
with self.connection.cursor() as cur:
|
|
cur.execute("INSERT INTO sites (id, name) VALUES (%s, %s)", (site_id, str(site_id)))
|
|
cur.execute(
|
|
"INSERT INTO rooms (id, site_id, name) VALUES (%s, %s, %s)",
|
|
(room_id, site_id, str(room_id)),
|
|
)
|
|
cur.executemany(
|
|
"INSERT INTO miners (id, lifecycle_status, first_seen_at, last_seen_at) "
|
|
"VALUES (%s, 'active', NULL, NULL)",
|
|
[(mapped_id,), (unmapped_id,)],
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO location_positions (room_id, room_name, rack_name, shelf_name, "
|
|
"position_index, miner_id) VALUES (%s, 'Room', 'A', '1', 1, %s), "
|
|
"(%s, 'Room', 'A', '1', 2, NULL)",
|
|
(room_id, mapped_id, room_id),
|
|
)
|
|
after = repository.summarize_locations(include_miner_counts=True)
|
|
self.assertEqual(after["sites"] - before["sites"], 1)
|
|
self.assertEqual(after["rooms"] - before["rooms"], 1)
|
|
self.assertEqual(after["positions"] - before["positions"], 2)
|
|
self.assertEqual(after["occupied_positions"] - before["occupied_positions"], 1)
|
|
self.assertEqual(after["empty_positions"] - before["empty_positions"], 1)
|
|
self.assertEqual(after["mapped_miners"] - before["mapped_miners"], 1)
|
|
self.assertEqual(after["unmapped_miners"] - before["unmapped_miners"], 1)
|
|
|
|
def test_repair_summary_executes_without_loading_queue_items(self):
|
|
summary = RepairCenterRepository().summarize_active_repairs()
|
|
self.assertIn("active", summary)
|
|
self.assertIn("status_counts", summary)
|
|
|
|
|
|
class DashboardArchitectureTests(unittest.TestCase):
|
|
def test_router_and_service_have_no_sql_or_database_driver_imports(self):
|
|
for relative in ("dashboard/router.py", "dashboard/service.py"):
|
|
source = (APP_ROOT / relative).read_text(encoding="utf-8")
|
|
modules = []
|
|
for node in ast.walk(ast.parse(source)):
|
|
if isinstance(node, ast.Import):
|
|
modules.extend(alias.name for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
modules.append(node.module)
|
|
self.assertFalse(
|
|
any(name.startswith(("psycopg", "sqlalchemy", "asyncpg")) for name in modules)
|
|
)
|
|
for keyword in ("SELECT ", "INSERT ", "UPDATE ", "DELETE "):
|
|
self.assertNotIn(keyword, source.upper())
|
|
|
|
def test_dashboard_frontend_has_one_read_only_request_and_safe_dom_writes(self):
|
|
source = (APP_ROOT / "frontend" / "static" / "dashboard.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertEqual(source.count('fetch("/api/dashboard/overview"'), 1)
|
|
self.assertNotIn("innerHTML", source)
|
|
self.assertNotIn("setInterval", source)
|
|
self.assertNotIn('method: "POST"', source)
|
|
self.assertNotIn('method: "PATCH"', source)
|
|
self.assertNotIn('method: "DELETE"', source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|