feat: add miner detail view
This commit is contained in:
1 parent
b8b741ccee
commit
19fc29c72c
22 files changed
+1882
-10
No files matched your search
@@ -0,0 +1,636 @@
|
||||
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 (
|
||||
"<title>Miner Detail | MinerOps</title>",
|
||||
'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'<button class="quiet-button" id="minerDetailRefresh" '
|
||||
r'type="button">Refresh</button>',
|
||||
)
|
||||
source = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8")
|
||||
self.assertIn(
|
||||
'el.refresh.addEventListener("click", loadOverview);',
|
||||
source,
|
||||
)
|
||||
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)
|
||||
for mutation in (
|
||||
'method: "POST"', '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()
|
||||
Reference in new issue
Block a user