import ast import re import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch from uuid import UUID, uuid4 import psycopg from fastapi import FastAPI from fastapi.testclient import TestClient from app.config import settings from app.frontend.router import miner_detail_page, router as frontend_router from app.identity import get_current_user from app.inventory.exceptions import MinerNotFound from app.inventory.models import InventoryMinerDetail from app.inventory.repository import InventoryRepository from app.inventory.service import InventoryService from app.locations.models import MinerLocationDetail from app.locations.repository import LocationMapRepository from app.miner_detail.dependencies import get_miner_detail_service from app.miner_detail.models import MinerDetailAccess from app.miner_detail.router import router as detail_router from app.miner_detail.service import MinerDetailService from app.repairs.center_repository import RepairCenterRepository from app.repairs.models import RepairCenterCapabilities, RepairCenterItem NOW = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc) APP_ROOT = Path(__file__).parents[1] / "app" STATIC_ROOT = APP_ROOT / "frontend" / "static" def current_miner(**updates): values = { "id": uuid4(), "asset_id": "MOP-001", "current_mac": "aa:bb:cc:dd:ee:ff", "current_ip": "192.0.2.10", "model": "M50", "miner_sn": "MINER-SN", "lifecycle_status": "active", "runtime_status": "online", "site_id": uuid4(), "site_name": "Restricted Site", "room_id": uuid4(), "room_name": "Restricted Room", "location_position_id": uuid4(), "rack_label": "Rack A", "shelf_label": "Shelf 1", "position_label": "A-01", "position_index": 1, "observed_at": NOW, "hashrate_ths": 0.0, "power_w": 0.0, "chip_temp_max_c": 0.0, "hash_stable": None, "tags": [], "created_at": NOW - timedelta(days=10), "updated_at": NOW, "last_seen_at": NOW, "hostname": "miner-001", "control_board_sn": "CB-SN", "psu_serial": "PSU-SN", "firmware_version": "1.2.3", "notes": None, "first_seen_at": NOW - timedelta(days=9), } values.update(updates) return InventoryMinerDetail(**values) def location(): return MinerLocationDetail( site_id=uuid4(), site_name="North Site", room_id=uuid4(), room_name="Room 1", position_id=uuid4(), rack_label="Rack A", shelf_label="Shelf 1", position_label="A-01", position_index=1, ) def repair_item(*, source="modular", active=True): return RepairCenterItem( repair_id=uuid4() if source == "modular" else None, source=source, active=active, opened_at=NOW - timedelta(hours=2), repair_status="OPEN" if active else "COMPLETED", reason="CONTROL_BOARD" if source == "modular" else None, latest_context="Board diagnostics", latest_context_at=NOW, miner_id=uuid4(), lifecycle_status="in_repair" if active else "active", parts_count=1 if source == "modular" else None, parts_summary=["control board"] if source == "modular" else [], capabilities=RepairCenterCapabilities(), ) class FakeInventoryService: def __init__(self, miner=None, *, missing=False): self.miner = miner or current_miner() self.missing = missing self.calls = [] def get_current_miner(self, miner_id): self.calls.append(miner_id) if self.missing: raise MinerNotFound(miner_id) return self.miner class FakeLocationService: def __init__(self, value=None): self.value = value self.calls = [] def get_miner_location(self, miner_id): self.calls.append(miner_id) return self.value class FakeRepairService: def __init__(self, values=None): self.values = list(values or []) self.calls = [] def list_miner_repairs(self, miner_id, access): self.calls.append((miner_id, access)) return self.values class MinerDetailServiceTests(unittest.TestCase): def build(self, *, miner=None, mapped=None, repairs=None, missing=False): self.inventory = FakeInventoryService(miner, missing=missing) self.locations = FakeLocationService(mapped) self.repairs = FakeRepairService(repairs) return MinerDetailService(self.inventory, self.locations, self.repairs) def test_full_access_uses_three_domain_queries_and_preserves_zero(self): miner_id = uuid4() mapped = location() active = repair_item() service = self.build(mapped=mapped, repairs=[active]) response = service.get_overview( miner_id, MinerDetailAccess( can_view_locations=True, can_read_repairs=True, can_view_repairs=True, can_read_components=True, ), ) self.assertEqual(self.inventory.calls, [miner_id]) self.assertEqual(self.locations.calls, [miner_id]) self.assertEqual(len(self.repairs.calls), 1) self.assertEqual(response.meta.database_query_count, 3) self.assertEqual(response.data.miner.hashrate_ths, 0) self.assertEqual(response.data.miner.power_w, 0) self.assertEqual(response.data.miner.chip_temp_max_c, 0) self.assertIsNone(response.data.miner.hash_stable) self.assertEqual(response.data.location.location, mapped) self.assertEqual(response.data.repair.active_repair.repair_id, active.repair_id) self.assertEqual(response.data.operational_state, "repair") self.assertTrue(response.data.capabilities.can_view_map) self.assertTrue(response.data.capabilities.can_open_repair_center) def test_restricted_domains_are_not_queried_and_location_never_leaks(self): service = self.build(mapped=location(), repairs=[repair_item()]) response = service.get_overview(uuid4(), MinerDetailAccess()) self.assertEqual(response.meta.database_query_count, 1) self.assertEqual(self.locations.calls, []) self.assertEqual(self.repairs.calls, []) self.assertTrue(response.data.location.restricted) self.assertIsNone(response.data.location.location) self.assertTrue(response.data.repair.restricted) self.assertEqual(response.data.repair.episodes, []) miner = response.data.miner for field in ( "site_id", "site_name", "room_id", "room_name", "location_position_id", "rack_label", "shelf_label", "position_label", "position_index", ): self.assertIsNone(getattr(miner, field), field) serialized = response.model_dump_json() self.assertNotIn("Restricted Site", serialized) self.assertNotIn("Restricted Room", serialized) def test_no_repair_and_legacy_repair_are_truthfully_distinct(self): no_repair = self.build().get_overview( uuid4(), MinerDetailAccess(can_read_repairs=True) ) self.assertTrue(no_repair.data.repair.available) self.assertFalse(no_repair.data.repair.restricted) self.assertIsNone(no_repair.data.repair.active_repair) self.assertEqual(no_repair.data.repair.episodes, []) legacy = repair_item(source="legacy") legacy_response = self.build(repairs=[legacy]).get_overview( uuid4(), MinerDetailAccess(can_read_repairs=True) ) self.assertIsNone(legacy_response.data.repair.active_repair) self.assertEqual(legacy_response.data.repair.episodes[0].source, "legacy") self.assertIsNone(legacy_response.data.repair.episodes[0].repair_id) def test_unscoped_legacy_position_is_truthful_but_not_map_linkable(self): unscoped = location().model_copy( update={"site_id": None, "site_name": None, "room_id": None} ) response = self.build(mapped=unscoped).get_overview( uuid4(), MinerDetailAccess(can_view_locations=True) ) self.assertTrue(response.data.location.available) self.assertEqual(response.data.location.location.position_label, "A-01") self.assertFalse(response.data.capabilities.can_view_map) def test_factual_state_uses_no_temperature_or_age_threshold(self): cases = ( ({"runtime_status": "offline", "chip_temp_max_c": 99.0}, "offline"), ({"runtime_status": "online", "hash_stable": False}, "unstable"), ({"runtime_status": "online", "observed_at": NOW - timedelta(days=90)}, "online"), ({"runtime_status": "unknown", "chip_temp_max_c": 1000.0}, "unknown"), ({"runtime_status": "online", "lifecycle_status": "spare"}, "spare"), ({"runtime_status": "online", "lifecycle_status": "retired"}, "retired"), ) for updates, expected in cases: with self.subTest(expected=expected): response = self.build(miner=current_miner(**updates)).get_overview( uuid4(), MinerDetailAccess() ) self.assertEqual(response.data.operational_state, expected) response = self.build( miner=current_miner(hashrate_ths=float("nan")) ).get_overview(uuid4(), MinerDetailAccess()) self.assertIsNone(response.data.miner.hashrate_ths) self.assertEqual(response.data.operational_state, "online") def test_unknown_miner_propagates_domain_not_found_without_other_queries(self): service = self.build(missing=True, mapped=location(), repairs=[repair_item()]) with self.assertRaises(MinerNotFound): service.get_overview( uuid4(), MinerDetailAccess(can_view_locations=True, can_read_repairs=True), ) self.assertEqual(self.locations.calls, []) self.assertEqual(self.repairs.calls, []) class StubMinerDetailService: def __init__(self, *, missing=False): self.missing = missing self.calls = [] self.delegate = MinerDetailService( FakeInventoryService(), FakeLocationService(location()), FakeRepairService([repair_item()]), ) def get_overview(self, miner_id, access): self.calls.append((miner_id, access)) if self.missing: raise MinerNotFound(miner_id) return self.delegate.get_overview(miner_id, access) class MinerDetailRouterTests(unittest.TestCase): def setUp(self): self.service = StubMinerDetailService() self.app = FastAPI() self.app.include_router(detail_router) self.app.dependency_overrides[get_miner_detail_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_miners_view(self): path = f"/api/miners/{uuid4()}/overview" self.assertEqual(self.client.get(path).status_code, 401) self.authenticate(("locations.view", "repairs.read")) self.assertEqual(self.client.get(path).status_code, 403) def test_returns_typed_permission_aware_no_store_envelope(self): miner_id = uuid4() self.authenticate(( "miners.view", "locations.view", "repairs.read", "repairs.view", "components.read", )) response = self.client.get(f"/api/miners/{miner_id}/overview") self.assertEqual(response.status_code, 200) self.assertEqual(response.headers["cache-control"], "no-store") payload = response.json() self.assertEqual(payload["data"]["miner"]["id"], str(self.service.delegate.inventory_service.miner.id)) self.assertEqual(payload["meta"]["database_query_count"], 3) self.assertIsNone(payload["error"]) _, access = self.service.calls[-1] self.assertTrue(access.can_view_locations) self.assertTrue(access.can_read_repairs) self.assertTrue(access.can_view_repairs) self.assertTrue(access.can_read_components) def test_malformed_and_unknown_uuid_use_existing_validation_and_404(self): self.authenticate(("miners.view",)) self.assertEqual(self.client.get("/api/miners/not-a-uuid/overview").status_code, 422) self.service.missing = True response = self.client.get(f"/api/miners/{uuid4()}/overview") self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["detail"], "Miner not found") class MinerDetailRepositoryIntegrationTests(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_inventory_and_location_projections_use_latest_factual_rows(self): miner_id, site_id, room_id = uuid4(), uuid4(), uuid4() earlier = NOW - timedelta(hours=1) with self.connection.cursor() as cur: cur.execute("INSERT INTO sites (id, name) VALUES (%s, 'Detail Site')", (site_id,)) cur.execute( "INSERT INTO rooms (id, site_id, name) VALUES (%s, %s, 'Detail Room')", (room_id, site_id), ) cur.execute( "INSERT INTO miners (id, asset_tag, current_mac, current_ip, model, " "status, lifecycle_status, first_seen_at, last_seen_at) " "VALUES (%s, %s, 'aa:bb:cc:00:00:01', '192.0.2.22', 'M50', " "'online', 'active', NULL, NULL)", (miner_id, "DETAIL-" + uuid4().hex[:8]), ) cur.execute( "INSERT INTO location_positions (room_id, room_name, rack_name, " "shelf_name, position_index, position_label, miner_id) " "VALUES (%s, 'Detail Room', 'Rack A', 'Shelf 1', 2, 'A-02', %s)", (room_id, miner_id), ) cur.execute( "INSERT INTO miner_status_snapshots " "(id, miner_id, observed_at, mhs_av, power_w, chip_temp_max_c, hash_stable) " "VALUES (%s, %s, %s, 120000000, 3300, 70, true), " "(%s, %s, %s, 0, 0, 0, NULL)", (uuid4(), miner_id, earlier, uuid4(), miner_id, NOW), ) inventory = InventoryService( InventoryRepository(connection=self.connection) ).get_current_miner(miner_id) self.assertEqual(inventory.hashrate_ths, 0) self.assertEqual(inventory.power_w, 0) self.assertEqual(inventory.chip_temp_max_c, 0) self.assertIsNone(inventory.hash_stable) self.assertEqual(inventory.observed_at, NOW) placement = LocationMapRepository( connection=self.connection ).get_miner_location(miner_id) self.assertEqual(placement["site_id"], site_id) self.assertEqual(placement["room_id"], room_id) self.assertEqual(placement["rack_label"], "Rack A") self.assertIsInstance(placement["position_id"], UUID) self.assertEqual(RepairCenterRepository().list_miner_repairs(miner_id), []) class DummyCursor: def __enter__(self): return self def __exit__(self, *args): return False def execute(self, *args): return None def fetchall(self): return [] class DummyConnection: def cursor(self, **kwargs): return DummyCursor() def close(self): return None class NetworkIdentityPermissionTests(unittest.TestCase): def setUp(self): from app.main import list_miner_network_identities self.app = FastAPI() self.app.get("/api/miners/{miner_id}/network-identities")( list_miner_network_identities ) self.client = TestClient(self.app) def tearDown(self): self.client.close() def test_network_history_requires_miners_view(self): path = f"/api/miners/{uuid4()}/network-identities" self.assertEqual(self.client.get(path).status_code, 401) self.app.dependency_overrides[get_current_user] = lambda: { "id": str(uuid4()), "role": "Custom", "permissions": ["locations.view"] } self.assertEqual(self.client.get(path).status_code, 403) self.app.dependency_overrides[get_current_user] = lambda: { "id": str(uuid4()), "role": "Custom", "permissions": ["miners.view"] } with patch("app.main.get_db_connection", return_value=DummyConnection()): response = self.client.get(path) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"], []) class MinerDetailFrontendTests(unittest.TestCase): def setUp(self): self.app = FastAPI() self.app.include_router(frontend_router) self.app.dependency_overrides[get_current_user] = lambda: { "id": str(uuid4()), "username": "reader", "display_name": "Detail Reader", "role": "Read Only", "permissions": [ "miners.view", "locations.view", "repairs.read", "repairs.view" ], } self.client = TestClient(self.app) self.miner_id = uuid4() def tearDown(self): self.client.close() def test_page_requires_permission_and_renders_only_internal_alpha_shell(self): path = f"/miners/{self.miner_id}/detail" response = self.client.get(path) self.assertEqual(response.status_code, 200) for expected in ( "Miner Detail | MinerOps", 'data-view="miner_detail"', f'data-miner-id="{self.miner_id}"', "/assets/minerops-miner-detail.js", 'id="minerIdentityDetails"', 'id="minerRepairEpisodes"', 'id="minerObservations"', ): self.assertIn(expected, response.text) for legacy in ( "minerops-miner-detail-extra", "minerops-miner-detail-clean-v3", "minerops-location-management", "Manual Edit Missing Details", "Reboot", "Performance Mode", ): self.assertNotIn(legacy, response.text) self.app.dependency_overrides[get_current_user] = lambda: { "id": str(uuid4()), "role": "Custom", "permissions": [] } self.assertEqual(self.client.get(path).status_code, 403) def test_return_to_accepts_protected_local_context_and_rejects_external(self): path = f"/miners/{self.miner_id}/detail" local = "/monitoring/map?site_id=" + str(uuid4()) + "&miner_id=" + str(self.miner_id) response = self.client.get(path, params={"return_to": local}) self.assertIn("data-return-to=", response.text) self.assertIn("/monitoring/map?site_id=", response.text) response = self.client.get( path, params={"return_to": "https://example.invalid/steal"} ) self.assertNotIn("data-return-to=", response.text) self.assertNotIn("example.invalid", response.text) def test_refresh_action_has_visible_name_and_keeps_existing_behavior(self): response = self.client.get(f"/miners/{self.miner_id}/detail") self.assertRegex( response.text, r'', ) source = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8") self.assertIn('el.refresh.addEventListener("click", refreshAll);', source) refresh = re.search( r"function refreshAll\(\) \{\s*loadOverview\(\);\s*loadControls\(\);\s*\}", source, ) self.assertIsNotNone(refresh) css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8") self.assertIn(".miner-detail-actions .quiet-button { color: var(--ink); }", css) def test_live_route_precedence_and_middleware_isolation(self): from app.main import app detail_handlers = [ route.endpoint for route in app.routes if route.path == "/miners/{miner_id}/detail" and "GET" in (route.methods or set()) ] self.assertGreaterEqual(len(detail_handlers), 2) self.assertIs(detail_handlers[0], miner_detail_page) current_user = { "id": str(uuid4()), "username": "reader", "display_name": "Detail Reader", "role": "Read Only", "permissions": ["miners.view"], } app.dependency_overrides[get_current_user] = lambda: current_user try: with patch("app.identity.request_user_or_none", return_value=current_user): with TestClient(app) as client: response = client.get(f"/miners/{self.miner_id}/detail") finally: app.dependency_overrides.pop(get_current_user, None) self.assertEqual(response.status_code, 200) self.assertEqual(response.text.count('data-view="miner_detail"'), 1) self.assertIn("/assets/minerops-miner-detail.js", response.text) for legacy in ( "/assets/minerops-miner-detail-extra.js", "/assets/minerops-miner-detail-extra-v2.js", "/assets/minerops-miner-detail-clean-v3.js", "/assets/minerops-location-management.js", ): self.assertNotIn(legacy, response.text) paths = {route.path for route in app.routes} for legacy_api in ( "/api/miners/{miner_id}", "/api/miners/{miner_id}/detail-extra", "/api/miners/{miner_id}/detail-clean-v3", "/api/miners/{miner_id}/network-identities", "/api/miners/{miner_id}/status-snapshots", ): self.assertIn(legacy_api, paths) def test_frontend_is_lazy_read_only_bounded_and_xss_safe(self): source = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8") self.assertEqual( source.count( 'api("/api/miners/" + encodeURIComponent(minerId) + "/overview")' ), 1, ) self.assertEqual(source.count("Promise.all(["), 2) self.assertIn("/asset-id-history", source) self.assertIn("/network-identities", source) self.assertIn("/status-snapshots", source) self.assertIn("/events", source) self.assertIn("/parts", source) self.assertNotIn("setInterval", source) self.assertEqual(source.count('method: "POST"'), 1) self.assertIn('"/controls/" + pathAction', source) self.assertIn( 'action === "performance_mode" ? "performance-mode" : action', source ) for mutation in ( 'method: "PATCH"', 'method: "PUT"', 'method: "DELETE"', "reboot", "change-mac", "manual-details", ): self.assertNotIn(mutation, source) for unsafe in ("innerHTML", "outerHTML", "insertAdjacentHTML", "document.write"): self.assertNotIn(unsafe, source) self.assertIn("textContent", source) self.assertIn("new URLSearchParams", source) self.assertIn("encodeURIComponent", source) self.assertNotRegex( source.lower(), r"(temperature|hashrate|stale).{0,40}(threshold|>=\s*90|>\s*90)", ) main_source = (APP_ROOT / "main.py").read_text(encoding="utf-8") snapshot = re.search( r'def list_miner_status_snapshots\(.*?LIMIT 100;', main_source, re.DOTALL, ) self.assertIsNotNone(snapshot) def test_cross_navigation_preserves_context_with_durable_uuid(self): fleet = (STATIC_ROOT / "fleet.js").read_text(encoding="utf-8") repairs = (STATIC_ROOT / "repair-center.js").read_text(encoding="utf-8") for source in (fleet, repairs): self.assertIn('return_to', source) self.assertIn("window.location.pathname + window.location.search", source) self.assertIn("encodeURIComponent", source) detail = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8") self.assertIn('new URLSearchParams({ miner_id: minerIdValue })', detail) self.assertIn('params.set("site_id", location.site_id)', detail) self.assertIn('params.set("room_id", location.room_id)', detail) self.assertIn('new URLSearchParams({ scope: "all", q: data.miner.id })', detail) class MinerDetailArchitectureTests(unittest.TestCase): def test_router_and_service_preserve_dependency_boundaries(self): for relative in ("miner_detail/router.py", "miner_detail/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()) service_source = (APP_ROOT / "miner_detail" / "service.py").read_text( encoding="utf-8" ) self.assertNotIn("fastapi", service_source.lower()) def test_exactly_one_new_api_route_and_no_state_changing_page_get(self): routes = [ (route.path, route.methods) for route in detail_router.routes ] self.assertEqual(routes, [("/api/miners/{miner_id}/overview", {"GET"})]) page = next( route for route in frontend_router.routes if route.path == "/miners/{miner_id}/detail" ) self.assertEqual(page.methods, {"GET"}) if __name__ == "__main__": unittest.main()