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": '
', "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"],
'')
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)
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_map_repository_uses_constant_set_based_queries(self):
source = (Path(__file__).parents[1] / "app" / "locations" /
"repository.py").read_text(encoding="utf-8")
self.assertEqual(source.count("cur.execute("), 4)
self.assertEqual(source.count("LEFT JOIN LATERAL"), 2)
self.assertIn("ORDER BY snapshots.observed_at DESC, snapshots.id DESC", source)
self.assertIn("ORDER BY repairs.opened_at, repairs.id", source)
self.assertNotIn("for miner", source)
if __name__ == "__main__":
unittest.main()