401 lines
18 KiB
Python
401 lines
18 KiB
Python
import ast
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import FastAPI
|
|
import psycopg
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import settings
|
|
from app.identity import get_current_user
|
|
from app.locations.dependencies import get_location_map_service
|
|
from app.locations.repository import LocationMapRepository
|
|
from app.locations.router import router
|
|
from app.locations.service import LocationMapService
|
|
|
|
|
|
NOW = datetime(2026, 8, 10, 12, 0, tzinfo=timezone.utc)
|
|
STATIC_ROOT = Path(__file__).parents[1] / "app" / "frontend" / "static"
|
|
|
|
|
|
def map_rows():
|
|
site_a, site_b, room_a, room_b = uuid4(), uuid4(), uuid4(), uuid4()
|
|
miner_id, position_id, repair_id = uuid4(), uuid4(), uuid4()
|
|
return {
|
|
"sites": [{"id": site_a, "name": "North"}, {"id": site_b, "name": "South"}],
|
|
"rooms": [
|
|
{"id": room_a, "site_id": site_a, "site_name": "North", "name": "Room 1", "sort_order": 1},
|
|
{"id": room_b, "site_id": site_b, "site_name": "South", "name": "Room 1", "sort_order": 1},
|
|
],
|
|
"positions": [{
|
|
"location_position_id": position_id, "site_id": site_a,
|
|
"site_name": "North", "room_id": room_a, "room_name": "Room 1",
|
|
"rack_name": "A", "shelf_name": "Top", "shelf_index": 1,
|
|
"position_index": 1, "position_label": "A-1", "position_kind": "miner_slot",
|
|
"position_status": "occupied", "occupied": True,
|
|
"expected_mac": "de:ad:be:ef:00:01",
|
|
"equipment_label": None, "notes": None, "source_file": None,
|
|
"source_cell": None, "conflict_reason": None, "miner_id": miner_id,
|
|
"asset_id": '<img src=x onerror="alert(1)">', "current_mac": "aa:bb:cc:dd:ee:ff",
|
|
"current_ip": "192.0.2.10", "miner_status": "online", "hostname": "miner-a",
|
|
"miner_sn": "SERIAL-1", "power_sn": None, "model": "M50",
|
|
"miner_type": "WhatsMiner", "firmware_version": "1.0",
|
|
"lifecycle_status": "in_repair", "last_seen_at": NOW,
|
|
"asset_state": "shelved", "observed_at": NOW, "hashrate_ths": 120.5,
|
|
"power_w": 3320.0, "chip_temp_max_c": 71.0, "chip_temp_avg_c": 66.0,
|
|
"hash_stable": True, "pool_rejected_percent": 0.2,
|
|
"active_repair_id": repair_id, "active_repair_status": "OPEN",
|
|
"active_repair_reason": "CONTROL_BOARD",
|
|
}, {
|
|
"location_position_id": uuid4(), "site_id": site_a, "site_name": "North",
|
|
"room_id": room_a, "room_name": "Room 1", "rack_name": "A",
|
|
"shelf_name": "Top", "shelf_index": 1, "position_index": 2,
|
|
"position_label": "A-2", "position_kind": "miner_slot",
|
|
"position_status": "available", "occupied": False,
|
|
"active_repair_id": None, "active_repair_status": None,
|
|
"active_repair_reason": None,
|
|
}],
|
|
"virtual_miners": [],
|
|
}
|
|
|
|
|
|
class FakeLocationRepository:
|
|
def __init__(self):
|
|
self.rows = map_rows()
|
|
self.calls = []
|
|
|
|
def read_map(self, *, site_id=None, room_id=None):
|
|
self.calls.append((site_id, room_id))
|
|
return self.rows
|
|
|
|
|
|
class MineMapServiceTests(unittest.TestCase):
|
|
def test_service_builds_typed_bulk_map_with_repair_and_empty_position(self):
|
|
repository = FakeLocationRepository()
|
|
response = LocationMapService(repository).get_map(include_miner_details=True)
|
|
self.assertEqual(len(repository.calls), 1)
|
|
self.assertEqual(response.data.rooms[0].name, response.data.rooms[1].name)
|
|
self.assertNotEqual(response.data.rooms[0].site_id, response.data.rooms[1].site_id)
|
|
self.assertTrue(response.data.positions[0].occupied)
|
|
self.assertEqual(response.data.positions[0].active_repair.status, "OPEN")
|
|
self.assertFalse(response.data.positions[1].occupied)
|
|
self.assertEqual(response.data.positions[0].hashrate_ths, 120.5)
|
|
|
|
def test_location_only_user_receives_layout_without_miner_details(self):
|
|
response = LocationMapService(FakeLocationRepository()).get_map(
|
|
include_miner_details=False
|
|
)
|
|
position = response.data.positions[0]
|
|
self.assertTrue(position.occupied)
|
|
self.assertIsNone(position.miner_id)
|
|
self.assertIsNone(position.asset_id)
|
|
self.assertIsNone(position.expected_mac)
|
|
for field in ("current_mac", "current_ip", "miner_status", "hostname",
|
|
"miner_sn", "power_sn", "model", "miner_type",
|
|
"firmware_version", "lifecycle_status", "last_seen_at",
|
|
"asset_state", "observed_at", "hashrate_ths", "power_w",
|
|
"chip_temp_max_c", "chip_temp_avg_c", "hash_stable",
|
|
"pool_rejected_percent"):
|
|
self.assertIsNone(getattr(position, field), field)
|
|
self.assertIsNone(position.active_repair)
|
|
self.assertEqual(response.data.virtual_miners, [])
|
|
self.assertFalse(response.meta.miner_details_included)
|
|
|
|
|
|
def test_service_preserves_legacy_position_without_assigning_durable_scope(self):
|
|
repository = FakeLocationRepository()
|
|
legacy = dict(repository.rows["positions"][1])
|
|
legacy.update({
|
|
"location_position_id": uuid4(), "site_id": None, "site_name": None,
|
|
"room_id": None, "room_name": "Imported Room", "rack_name": "A",
|
|
"shelf_name": "Top", "position_index": 3,
|
|
})
|
|
repository.rows["positions"].append(legacy)
|
|
response = LocationMapService(repository).get_map(include_miner_details=True)
|
|
position = response.data.positions[-1]
|
|
self.assertIsNone(position.site_id)
|
|
self.assertIsNone(position.room_id)
|
|
self.assertEqual(position.room_name, "Imported Room")
|
|
|
|
|
|
class StubMapService:
|
|
def __init__(self):
|
|
self.repository = FakeLocationRepository()
|
|
self.service = LocationMapService(self.repository)
|
|
self.include = None
|
|
|
|
def get_map(self, **kwargs):
|
|
self.include = kwargs["include_miner_details"]
|
|
return self.service.get_map(**kwargs)
|
|
|
|
|
|
class MineMapRouterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.service = StubMapService()
|
|
self.app = FastAPI()
|
|
self.app.include_router(router)
|
|
self.app.dependency_overrides[get_location_map_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_map_requires_authentication_and_locations_permission(self):
|
|
self.assertEqual(self.client.get("/api/location/map").status_code, 401)
|
|
self.authenticate(("miners.view",))
|
|
self.assertEqual(self.client.get("/api/location/map").status_code, 403)
|
|
|
|
def test_map_returns_typed_envelope_and_permission_aware_miner_shape(self):
|
|
self.authenticate(("locations.view", "miners.view"))
|
|
response = self.client.get("/api/location/map")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()["meta"]["miner_details_included"])
|
|
self.assertEqual(response.json()["data"]["positions"][0]["asset_id"],
|
|
'<img src=x onerror="alert(1)">')
|
|
self.assertIsNone(response.json()["error"])
|
|
self.authenticate(("locations.view",))
|
|
response = self.client.get("/api/location/map")
|
|
redacted = response.json()["data"]["positions"][0]
|
|
self.assertIsNone(redacted["miner_id"])
|
|
self.assertIsNone(redacted["expected_mac"])
|
|
self.assertIsNone(redacted["active_repair"])
|
|
self.assertEqual(response.json()["data"]["virtual_miners"], [])
|
|
self.assertFalse(response.json()["meta"]["miner_details_included"])
|
|
|
|
def test_uuid_filters_are_validated_and_forwarded(self):
|
|
self.authenticate(("locations.view", "miners.view"))
|
|
site_id, room_id = uuid4(), uuid4()
|
|
response = self.client.get(
|
|
"/api/location/map", params={"site_id": site_id, "room_id": room_id}
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(self.service.repository.calls[-1], (site_id, room_id))
|
|
self.assertEqual(self.client.get("/api/location/map?site_id=bad").status_code, 422)
|
|
|
|
def test_router_has_no_sql_or_database_driver_imports(self):
|
|
path = Path(__file__).parents[1] / "app" / "locations" / "router.py"
|
|
source = path.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())
|
|
|
|
|
|
class MineMapRepositoryIntegrationTests(unittest.TestCase):
|
|
def test_set_based_query_scopes_same_named_rooms_by_uuid_and_site(self):
|
|
connection = psycopg.connect(
|
|
host=settings.postgres_host,
|
|
port=settings.postgres_port,
|
|
dbname=settings.postgres_db,
|
|
user=settings.postgres_user,
|
|
password=settings.postgres_password,
|
|
)
|
|
try:
|
|
repository = LocationMapRepository(connection=connection)
|
|
site_a, site_b, room_a, room_b = uuid4(), uuid4(), uuid4(), uuid4()
|
|
miner_id = uuid4()
|
|
room_name = "Map Same Name " + uuid4().hex[:8]
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO sites (id, name) VALUES (%s, %s), (%s, %s)",
|
|
(site_a, "Map North " + uuid4().hex[:8],
|
|
site_b, "Map South " + uuid4().hex[:8]),
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO rooms (id, site_id, name, sort_order) "
|
|
"VALUES (%s, %s, %s, 1), (%s, %s, %s, 1)",
|
|
(room_a, site_a, room_name, room_b, site_b, room_name),
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO miners (id, site_id, room_id, asset_tag, status, "
|
|
"lifecycle_status) VALUES (%s, %s, %s, %s, 'online', 'active')",
|
|
(miner_id, site_a, room_a, "MAP-" + uuid4().hex[:8]),
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO location_positions "
|
|
"(room_id, room_name, rack_name, shelf_name, shelf_index, "
|
|
"position_index, position_kind, position_status, miner_id) "
|
|
"VALUES (%s, %s, 'A', '1', 1, 1, 'miner_slot', 'occupied', %s), "
|
|
"(%s, %s, 'B', '1', 1, 1, 'miner_slot', 'available', NULL)",
|
|
(room_a, room_name, miner_id, room_b, room_name),
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO miner_status_snapshots "
|
|
"(miner_id, mhs_av, power_w, chip_temp_max_c, hash_stable) "
|
|
"VALUES (%s, 120000000, 3300, 70, true)",
|
|
(miner_id,),
|
|
)
|
|
result = repository.read_map(site_id=site_a, room_id=room_a)
|
|
self.assertEqual(len(result["positions"]), 1)
|
|
row = result["positions"][0]
|
|
self.assertEqual(row["site_id"], site_a)
|
|
self.assertEqual(row["room_id"], room_a)
|
|
self.assertTrue(row["occupied"])
|
|
self.assertEqual(row["hashrate_ths"], 120.0)
|
|
matching_rooms = [room for room in result["rooms"] if room["name"] == room_name]
|
|
self.assertEqual({room["site_id"] for room in matching_rooms}, {site_a, site_b})
|
|
finally:
|
|
connection.rollback()
|
|
connection.close()
|
|
|
|
|
|
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 StatusSnapshotPermissionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from app.main import list_miner_status_snapshots
|
|
self.app = FastAPI()
|
|
self.app.get("/api/miners/{miner_id}/status-snapshots")(
|
|
list_miner_status_snapshots
|
|
)
|
|
self.client = TestClient(self.app)
|
|
|
|
def tearDown(self):
|
|
self.app.dependency_overrides.pop(get_current_user, None)
|
|
self.client.close()
|
|
|
|
def test_status_snapshots_require_authentication_permission_and_allow_reader(self):
|
|
miner_id = uuid4()
|
|
self.assertEqual(self.client.get(f"/api/miners/{miner_id}/status-snapshots").status_code, 401)
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": str(uuid4()), "role": "Custom", "permissions": ["locations.view"]
|
|
}
|
|
self.assertEqual(self.client.get(f"/api/miners/{miner_id}/status-snapshots").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(f"/api/miners/{miner_id}/status-snapshots")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"], [])
|
|
|
|
def test_established_roles_and_custom_roles_follow_miners_view_permission(self):
|
|
miner_id = uuid4()
|
|
allowed = (("Admin", []), ("Manager", ["miners.view"]),
|
|
("Technician", ["miners.view"]), ("Read Only", ["miners.view"]),
|
|
("Custom", ["miners.view"]))
|
|
with patch("app.main.get_db_connection", return_value=DummyConnection()):
|
|
for role, permissions in allowed:
|
|
with self.subTest(role=role):
|
|
self.app.dependency_overrides[get_current_user] = lambda r=role, p=permissions: {
|
|
"id": str(uuid4()), "role": r, "permissions": p
|
|
}
|
|
response = self.client.get(f"/api/miners/{miner_id}/status-snapshots")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": str(uuid4()), "role": "Custom", "permissions": []
|
|
}
|
|
self.assertEqual(
|
|
self.client.get(f"/api/miners/{miner_id}/status-snapshots").status_code, 403
|
|
)
|
|
|
|
|
|
class MineMapFrontendAuditTests(unittest.TestCase):
|
|
def test_map_module_uses_one_bulk_endpoint_and_safe_dom(self):
|
|
source = (STATIC_ROOT / "mine-map.js").read_text(encoding="utf-8")
|
|
self.assertEqual(source.count('fetch("/api/location/map"'), 1)
|
|
self.assertNotIn("/api/inventory/miners", source)
|
|
self.assertNotIn("status-snapshots", source)
|
|
for unsafe in ("innerHTML", "outerHTML", "insertAdjacentHTML",
|
|
"document.write", "eval(", "Function("):
|
|
self.assertNotIn(unsafe, source)
|
|
self.assertIn("textContent", source)
|
|
self.assertIn("encodeURIComponent(position.miner_id)", source)
|
|
self.assertIn('window.addEventListener("popstate"', source)
|
|
self.assertIn('el.mapWorkspace.addEventListener("click"', source)
|
|
self.assertIn('el.mapWorkspace.addEventListener("keydown"', source)
|
|
self.assertIn('tile.setAttribute("aria-pressed"', source)
|
|
self.assertIn("syncTileTabStops", source)
|
|
self.assertIn("renderLegacyPositions", source)
|
|
self.assertIn("minerDetailsIncluded", source)
|
|
self.assertIn("mapProblemsOnly", source)
|
|
|
|
def test_map_has_factual_status_rules_without_thresholds_or_mutation(self):
|
|
source = (STATIC_ROOT / "mine-map.js").read_text(encoding="utf-8")
|
|
for invented in ("> 90", ">= 90", "staleMinutes", "degradedThreshold"):
|
|
self.assertNotIn(invented, source)
|
|
self.assertIn('runtime === "offline"', source)
|
|
self.assertIn("position.hash_stable === false", source)
|
|
self.assertNotIn('method: "POST"', source)
|
|
self.assertNotIn('method: "PATCH"', source)
|
|
self.assertNotIn('method: "DELETE"', source)
|
|
self.assertIn('if (!state.minerDetailsIncluded)', source)
|
|
self.assertIn('Number.isFinite(value)', source)
|
|
|
|
def test_problems_and_measurement_layers_use_factual_relative_values(self):
|
|
source = (STATIC_ROOT / "mine-map.js").read_text(encoding="utf-8")
|
|
css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8")
|
|
self.assertIn("const PROBLEM_KEYS = new Set", source)
|
|
for state in ("offline", "unknown", "unstable", "repair"):
|
|
self.assertIn(f'"{state}"', source)
|
|
self.assertIn("state.problemsOnly", source)
|
|
self.assertIn('"--relative-percent"', source)
|
|
self.assertIn("Relative range", source)
|
|
self.assertIn('data-problems-muted="true"', css)
|
|
self.assertIn(".relative-scale-track", css)
|
|
|
|
def test_map_repository_uses_constant_set_based_queries(self):
|
|
source = (Path(__file__).parents[1] / "app" / "locations" /
|
|
"repository.py").read_text(encoding="utf-8")
|
|
|
|
read_map_source = source.split(" def read_map(", 1)[1].split(
|
|
" def summarize_locations(", 1
|
|
)[0]
|
|
|
|
latest_telemetry_sql = source.split(
|
|
'LATEST_TELEMETRY_SQL = """', 1
|
|
)[1].split('"""', 1)[0]
|
|
|
|
active_repair_sql = source.split(
|
|
'ACTIVE_REPAIR_SQL = """', 1
|
|
)[1].split('"""', 1)[0]
|
|
|
|
self.assertEqual(read_map_source.count("cur.execute("), 4)
|
|
self.assertEqual(
|
|
read_map_source.count("{LATEST_TELEMETRY_SQL}"),
|
|
2,
|
|
)
|
|
self.assertEqual(
|
|
read_map_source.count("{ACTIVE_REPAIR_SQL}"),
|
|
2,
|
|
)
|
|
|
|
self.assertIn("LEFT JOIN LATERAL", latest_telemetry_sql)
|
|
self.assertIn("LEFT JOIN LATERAL", active_repair_sql)
|
|
|
|
self.assertIn(
|
|
"ORDER BY snapshots.observed_at DESC, snapshots.id DESC",
|
|
latest_telemetry_sql,
|
|
)
|
|
self.assertIn(
|
|
"ORDER BY repairs.opened_at, repairs.id",
|
|
active_repair_sql,
|
|
)
|
|
|
|
self.assertNotIn("for miner", read_map_source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|