431 lines
19 KiB
Python
431 lines
19 KiB
Python
import inspect
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.frontend.router import router as frontend_router
|
|
from app.identity import get_current_user, required_api_permission
|
|
from app.repairs.center_repository import RepairCenterRepository
|
|
from app.repairs.center_service import RepairCenterService
|
|
from app.repairs.models import RepairCenterAccess, RepairCenterFilters
|
|
|
|
|
|
NOW = datetime(2026, 8, 11, 12, 0, tzinfo=timezone.utc)
|
|
STATIC_ROOT = Path(__file__).parents[1] / "app" / "frontend" / "static"
|
|
|
|
|
|
class FakeCenterRepository:
|
|
def __init__(self):
|
|
self.arguments = None
|
|
self.rows = []
|
|
|
|
def list_repair_center(self, filters, **permissions):
|
|
self.arguments = (filters, permissions)
|
|
return {
|
|
"items": self.rows,
|
|
"summary": {"total": len(self.rows), "active": len(self.rows),
|
|
"historical": 0, "status_counts": {"OPEN": len(self.rows)}},
|
|
"location_options": {"sites": [], "rooms": []},
|
|
}
|
|
|
|
|
|
def center_row(source="modular", active=True):
|
|
repair_id = uuid4() if source == "modular" else None
|
|
return {
|
|
"repair_id": repair_id,
|
|
"source": source,
|
|
"active": active,
|
|
"opened_at": NOW,
|
|
"closed_at": None,
|
|
"repair_status": "OPEN" if source == "modular" else "testing",
|
|
"reason": "HARDWARE_FAILURE" if source == "modular" else None,
|
|
"reported_problem": "Hashboard not detected",
|
|
"latest_context": "Testing replacement board",
|
|
"latest_context_at": NOW,
|
|
"opened_by_display_name": "Repair Tech" if source == "modular" else None,
|
|
"miner_id": uuid4(),
|
|
"asset_id": "MINER-100",
|
|
"model": "M60S",
|
|
"miner_sn": "SN-100",
|
|
"power_sn": "PSU-100",
|
|
"current_mac": "00:11:22:33:44:55",
|
|
"current_ip": "192.0.2.10",
|
|
"lifecycle_status": "in_repair",
|
|
"runtime_status": "offline",
|
|
"site_id": uuid4(),
|
|
"site_name": "Site A",
|
|
"room_id": uuid4(),
|
|
"room_name": "Room 1",
|
|
"rack_label": "12",
|
|
"shelf_label": "3",
|
|
"position_id": uuid4(),
|
|
"position_label": "P04",
|
|
"position_index": 4,
|
|
"parts_count": 1 if source == "modular" else None,
|
|
"parts_summary": ["Fan"] if source == "modular" else [],
|
|
"legacy_event_status": "testing" if source == "legacy" else None,
|
|
"legacy_asset_state": "repair" if source == "legacy" else None,
|
|
}
|
|
|
|
|
|
class RepairCenterServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.repository = FakeCenterRepository()
|
|
self.service = RepairCenterService(self.repository)
|
|
|
|
def test_modular_capabilities_use_repair_uuid_and_permissions(self):
|
|
self.repository.rows = [center_row()]
|
|
result = self.service.list_repair_center(
|
|
RepairCenterFilters(),
|
|
RepairCenterAccess(can_read_repairs=True, can_view_miners=True,
|
|
can_view_locations=True, can_complete_repairs=True,
|
|
can_manage_parts=True, can_manage_legacy=True),
|
|
)
|
|
item = result.data.items[0]
|
|
self.assertTrue(item.capabilities.can_read_modular_timeline)
|
|
self.assertTrue(item.capabilities.can_manage_modular_parts)
|
|
self.assertTrue(item.capabilities.can_complete_modular_repair)
|
|
self.assertFalse(item.capabilities.can_manage_legacy)
|
|
self.assertTrue(item.capabilities.can_open_map)
|
|
|
|
def test_part_management_requires_modular_read_access(self):
|
|
self.repository.rows = [center_row()]
|
|
result = self.service.list_repair_center(
|
|
RepairCenterFilters(),
|
|
RepairCenterAccess(can_manage_parts=True),
|
|
)
|
|
self.assertFalse(
|
|
result.data.items[0].capabilities.can_manage_modular_parts
|
|
)
|
|
|
|
def test_legacy_capabilities_never_enable_modular_mutations(self):
|
|
self.repository.rows = [center_row("legacy")]
|
|
result = self.service.list_repair_center(
|
|
RepairCenterFilters(),
|
|
RepairCenterAccess(can_read_repairs=True, can_view_miners=True,
|
|
can_view_locations=True, can_complete_repairs=True,
|
|
can_manage_parts=True, can_manage_legacy=True),
|
|
)
|
|
item = result.data.items[0]
|
|
self.assertTrue(item.capabilities.can_manage_legacy)
|
|
self.assertTrue(item.capabilities.can_mark_legacy_repaired)
|
|
self.assertFalse(item.capabilities.can_complete_modular_repair)
|
|
self.assertFalse(item.capabilities.can_manage_modular_parts)
|
|
|
|
def test_repository_receives_explicit_redaction_flags(self):
|
|
self.service.list_repair_center(
|
|
RepairCenterFilters(query=" fan "), RepairCenterAccess()
|
|
)
|
|
filters, permissions = self.repository.arguments
|
|
self.assertEqual(filters["query"], "fan")
|
|
self.assertFalse(permissions["include_miner_details"])
|
|
self.assertFalse(permissions["include_location_details"])
|
|
|
|
def test_pagination_and_exact_summary_are_preserved(self):
|
|
self.repository.rows = [center_row()]
|
|
result = self.service.list_repair_center(
|
|
RepairCenterFilters(limit=25, offset=50), RepairCenterAccess()
|
|
)
|
|
self.assertEqual(result.meta.limit, 25)
|
|
self.assertEqual(result.meta.offset, 50)
|
|
self.assertEqual(result.meta.returned, 1)
|
|
self.assertEqual(result.meta.total, 1)
|
|
self.assertEqual(result.data.summary.status_counts, {"OPEN": 1})
|
|
|
|
|
|
|
|
class RepairCenterDatabaseTests(unittest.TestCase):
|
|
def test_empty_database_projection_executes_as_typed_set_query(self):
|
|
result = RepairCenterRepository().list_repair_center(
|
|
{"scope": "active", "limit": 25, "offset": 0},
|
|
include_miner_details=True,
|
|
include_location_details=True,
|
|
)
|
|
self.assertEqual(result["items"], [])
|
|
self.assertEqual(result["summary"]["total"], 0)
|
|
self.assertEqual(result["location_options"], {"sites": [], "rooms": []})
|
|
|
|
def test_projection_consolidates_sources_filters_and_pages_exactly(self):
|
|
repository = RepairCenterRepository()
|
|
actor_id = uuid4()
|
|
site_id = uuid4()
|
|
room_id = uuid4()
|
|
modular_miner_id = uuid4()
|
|
legacy_miner_id = uuid4()
|
|
completed_miner_id = uuid4()
|
|
connection = repository._connection_factory()
|
|
try:
|
|
with connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO users (
|
|
id, username, display_name, password_hash, role_id
|
|
)
|
|
SELECT %s, %s, 'Repair Center Actor', 'not-used', id
|
|
FROM roles WHERE name = 'Admin'
|
|
""",
|
|
(actor_id, f"repair-center-{actor_id}"),
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO sites (id, name) VALUES (%s, %s)",
|
|
(site_id, f"Repair Center Site {site_id}"),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO rooms (id, site_id, name)
|
|
VALUES (%s, %s, 'Repair Room')
|
|
""",
|
|
(room_id, site_id),
|
|
)
|
|
for miner_id, asset_id, lifecycle in (
|
|
(modular_miner_id, "RC-MODULAR", "in_repair"),
|
|
(legacy_miner_id, "RC-LEGACY", "in_repair"),
|
|
(completed_miner_id, "RC-COMPLETED", "active"),
|
|
):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miners (
|
|
id, asset_tag, model, status, lifecycle_status,
|
|
first_seen_at, last_seen_at
|
|
)
|
|
VALUES (%s, %s, 'M60S', 'offline', %s, NULL, NULL)
|
|
""",
|
|
(miner_id, asset_id, lifecycle),
|
|
)
|
|
for index, miner_id in enumerate(
|
|
(modular_miner_id, legacy_miner_id), start=1
|
|
):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO location_positions (
|
|
room_id, room_name, rack_name, shelf_name,
|
|
position_index, position_label, miner_id
|
|
)
|
|
VALUES (%s, 'Repair Room', 'Rack 1', 'Shelf A',
|
|
%s, %s, %s)
|
|
""",
|
|
(room_id, index, f"P{index:02d}", miner_id),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_location_states (
|
|
miner_id, asset_state, state_notes,
|
|
updated_by_user_id
|
|
)
|
|
VALUES (%s, 'repair', 'Legacy repair context', %s)
|
|
""",
|
|
(miner_id, actor_id),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_maintenance_events (
|
|
miner_id, event_type, event_status, event_title,
|
|
description, created_by_user_id
|
|
)
|
|
VALUES (%s, 'repair', 'waiting_parts',
|
|
'Legacy board issue', 'Waiting for board', %s)
|
|
""",
|
|
(miner_id, actor_id),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO repairs (miner_id, reason, opened_by)
|
|
VALUES (%s, 'CONTROL_BOARD', %s)
|
|
""",
|
|
(modular_miner_id, actor_id),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO repairs (
|
|
miner_id, status, reason, opened_by, completed_at,
|
|
completed_by, resulting_lifecycle
|
|
)
|
|
VALUES (%s, 'COMPLETED', 'INSPECTION', %s, now(),
|
|
%s, 'active')
|
|
""",
|
|
(completed_miner_id, actor_id, actor_id),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
from app.main import minerops_repair_reject_modular_mutation
|
|
|
|
boundary_connection = repository._connection_factory()
|
|
try:
|
|
with boundary_connection.cursor() as cur:
|
|
with self.assertRaises(HTTPException) as raised:
|
|
minerops_repair_reject_modular_mutation(
|
|
cur, str(modular_miner_id)
|
|
)
|
|
self.assertEqual(raised.exception.status_code, 409)
|
|
minerops_repair_reject_modular_mutation(
|
|
cur, str(legacy_miner_id)
|
|
)
|
|
finally:
|
|
boundary_connection.close()
|
|
|
|
active = repository.list_repair_center(
|
|
{"scope": "active", "limit": 1, "offset": 0},
|
|
include_miner_details=True,
|
|
include_location_details=True,
|
|
)
|
|
self.assertEqual(active["summary"]["total"], 2)
|
|
self.assertEqual(len(active["items"]), 1)
|
|
self.assertEqual(active["summary"]["active"], 2)
|
|
self.assertEqual(active["location_options"]["sites"][0]["id"], str(site_id))
|
|
|
|
all_active = repository.list_repair_center(
|
|
{"scope": "active", "limit": 25, "offset": 0},
|
|
include_miner_details=True,
|
|
include_location_details=True,
|
|
)
|
|
by_miner = {row["miner_id"]: row for row in all_active["items"]}
|
|
self.assertEqual(set(by_miner), {modular_miner_id, legacy_miner_id})
|
|
self.assertEqual(by_miner[modular_miner_id]["source"], "modular")
|
|
self.assertEqual(by_miner[legacy_miner_id]["source"], "legacy")
|
|
self.assertEqual(by_miner[legacy_miner_id]["repair_event_count"], 1)
|
|
self.assertEqual(
|
|
by_miner[legacy_miner_id]["latest_event_title"],
|
|
"Legacy board issue",
|
|
)
|
|
self.assertEqual(all_active["summary"]["model_counts"], {"M60S": 2})
|
|
self.assertEqual(
|
|
all_active["summary"]["error_counts"],
|
|
{"CONTROL BOARD": 1, "Legacy board issue": 1},
|
|
)
|
|
|
|
filtered = repository.list_repair_center(
|
|
{"scope": "all", "query": "RC-COMPLETED", "model": "m60s",
|
|
"limit": 25, "offset": 0},
|
|
include_miner_details=True,
|
|
include_location_details=True,
|
|
)
|
|
self.assertEqual(filtered["summary"]["total"], 1)
|
|
self.assertFalse(filtered["items"][0]["active"])
|
|
|
|
redacted = repository.list_repair_center(
|
|
{"scope": "active", "limit": 25, "offset": 0},
|
|
include_miner_details=False,
|
|
include_location_details=False,
|
|
)
|
|
self.assertEqual(redacted["summary"]["total"], 2)
|
|
self.assertTrue(all(row["asset_id"] is None for row in redacted["items"]))
|
|
self.assertTrue(all(row["site_id"] is None for row in redacted["items"]))
|
|
self.assertEqual(redacted["location_options"], {"sites": [], "rooms": []})
|
|
|
|
class RepairCenterRepositoryAuditTests(unittest.TestCase):
|
|
def test_projection_prevents_duplicate_active_legacy_rows(self):
|
|
source = inspect.getsource(RepairCenterRepository._ctes)
|
|
self.assertIn("NOT EXISTS", source)
|
|
self.assertIn("active_repair.miner_id = miner.id", source)
|
|
self.assertIn("UNION ALL", source)
|
|
self.assertIn("part_rollup", source)
|
|
self.assertIn("placement", source)
|
|
|
|
def test_projection_uses_constant_set_based_queries(self):
|
|
source = inspect.getsource(RepairCenterRepository.list_repair_center)
|
|
self.assertEqual(source.count("cur.execute("), 3)
|
|
self.assertNotIn("for item in", source)
|
|
self.assertIn("count(*) FILTER", source)
|
|
self.assertIn("LIMIT %s OFFSET %s", source)
|
|
|
|
def test_filter_sql_is_parameterized_and_complete_result_scoped(self):
|
|
where, params = RepairCenterRepository._where({
|
|
"scope": "historical", "status": "COMPLETED", "query": "fan",
|
|
"site_id": uuid4(), "room_id": uuid4(),
|
|
})
|
|
self.assertIn("NOT candidate.active", where)
|
|
self.assertIn("candidate.repair_status = %s", where)
|
|
self.assertIn("candidate.site_id = %s", where)
|
|
self.assertTrue(params[-1].startswith("%fan"))
|
|
|
|
|
|
class RepairCenterFrontendTests(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": "tech", "display_name": "Repair Tech",
|
|
"role": "Technician", "permissions": ["repairs.view", "repairs.read",
|
|
"repairs.manage", "repairs.complete", "repairs.parts.manage",
|
|
"miners.view", "locations.view"],
|
|
}
|
|
self.client = TestClient(self.app)
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
|
|
def test_page_has_semantic_queue_filters_paging_and_inspector(self):
|
|
response = self.client.get("/operations/repairs")
|
|
self.assertEqual(response.status_code, 200)
|
|
for value in ('id="repairFilters"', 'id="repairRows"',
|
|
'id="repairInspector"', 'id="repairPrevious"', 'id="repairNext"',
|
|
'<th>Reported problem</th>', '/assets/minerops-repair-center.js'):
|
|
self.assertIn(value, response.text)
|
|
|
|
def test_location_filters_are_not_rendered_without_permission(self):
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": str(uuid4()), "role": "Custom", "permissions": ["repairs.view"]
|
|
}
|
|
text = self.client.get("/operations/repairs").text
|
|
self.assertNotIn('id="repairSite"', text)
|
|
self.assertNotIn('id="repairModel"', text)
|
|
self.assertNotIn('id="repairRoom"', text)
|
|
|
|
def test_restricted_filter_parameters_are_rejected(self):
|
|
from app.main import minerops_repair_queue_api
|
|
|
|
arguments = {
|
|
"query": None, "scope": "active", "status_filter": None,
|
|
"model": None, "lifecycle_status": None, "site_id": None,
|
|
"room_id": None, "limit": 50, "offset": 0,
|
|
"current_user": {"permissions": ["repairs.view"]},
|
|
}
|
|
for field, value in (("model", "M60S"), ("site_id", uuid4())):
|
|
with self.subTest(field=field):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
minerops_repair_queue_api(**{**arguments, field: value})
|
|
self.assertEqual(raised.exception.status_code, 403)
|
|
|
|
def test_frontend_is_one_initial_read_without_polling_or_unsafe_html(self):
|
|
source = (STATIC_ROOT / "repair-center.js").read_text(encoding="utf-8")
|
|
self.assertEqual(source.count('api("/api/repairs/queue?"'), 1)
|
|
self.assertNotIn("setInterval", source)
|
|
for unsafe in ("innerHTML", "insertAdjacentHTML", "document.write"):
|
|
self.assertNotIn(unsafe, source)
|
|
self.assertIn("Promise.all", source)
|
|
self.assertIn("encodeURIComponent", source)
|
|
self.assertIn("URLSearchParams", source)
|
|
|
|
def test_mutations_are_source_specific_and_gets_are_read_only(self):
|
|
source = (STATIC_ROOT / "repair-center.js").read_text(encoding="utf-8")
|
|
self.assertIn('/complete"', source)
|
|
self.assertIn('/parts"', source)
|
|
self.assertIn('"/note" : "/status"', source)
|
|
self.assertIn('/state"', source)
|
|
self.assertNotIn('method: "GET"', source)
|
|
self.assertIn("Type DECOMMISSION", source)
|
|
|
|
def test_miner_and_map_links_use_durable_safe_identifiers(self):
|
|
source = (STATIC_ROOT / "repair-center.js").read_text(encoding="utf-8")
|
|
self.assertIn('encodeURIComponent(item.miner_id) + "/detail"', source)
|
|
self.assertIn("site_id: item.site_id", source)
|
|
self.assertIn("room_id: item.room_id", source)
|
|
self.assertIn("miner_id: item.miner_id", source)
|
|
|
|
def test_history_get_requires_repairs_read(self):
|
|
self.assertEqual(required_api_permission(
|
|
"GET", f"/api/miners/{uuid4()}/maintenance-events"), "repairs.read")
|
|
self.assertEqual(required_api_permission(
|
|
"PATCH", f"/api/miners/{uuid4()}/maintenance-events/{uuid4()}"),
|
|
"repairs.manage")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|