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()