import ast from contextlib import contextmanager from copy import deepcopy from datetime import datetime, timezone from decimal import Decimal from pathlib import Path import unittest from uuid import uuid4 from fastapi import FastAPI from fastapi.testclient import TestClient from pydantic import ValidationError from app.components.dependencies import get_component_service from app.components.exceptions import ComponentValidationError from app.components.models import ComponentCreateRequest, ComponentDetail from app.components.router import router as component_router from app.components.service import ComponentService from app.identity import get_current_user, required_api_permission from app.repairs.dependencies import get_repair_service from app.repairs.exceptions import RepairPartAlreadyVoided from app.repairs.models import ( ConsumableInput, RepairPartComponentSnapshot, RepairPartConsumable, RepairPartDetail, RepairPartList, RepairPartRecordRequest, RepairPartVoidRequest, ) from app.repairs.router import router as repair_router from app.repairs.service import RepairService NOW = datetime(2026, 8, 2, 12, 0, tzinfo=timezone.utc) class FakeComponentRepository: def __init__(self): self.rows = {} self.calls = [] @contextmanager def transaction(self): self.calls.append("transaction") yield self def create_component(self, **values): component_id = uuid4() row = { "component_id": component_id, "created_at": NOW, **values, } self.rows[component_id] = row return row def get_component(self, component_id): return self.rows.get(component_id) class ComponentServiceTests(unittest.TestCase): def test_create_normalizes_metadata_and_uses_authoritative_fields(self): repository = FakeComponentRepository() actor = uuid4() component = ComponentService(repository).create_component( ComponentCreateRequest( component_type="hashboard", asset_id=" HB-0042 ", manufacturer=" MicroBT ", ), actor_user_id=actor, ) self.assertEqual(component.asset_id, "HB-0042") self.assertEqual(component.manufacturer, "MicroBT") self.assertEqual(component.created_by, actor) self.assertIsNotNone(component.component_id) self.assertEqual(repository.calls, ["transaction"]) def test_component_request_rejects_blank_unknown_and_server_fields(self): for body, exception in ( ({"component_type": "other"}, ValidationError), ({"component_type": "fan", "asset_id": " "}, ComponentValidationError), ({"component_type": "fan", "created_by": str(uuid4())}, ValidationError), ): with self.subTest(body=body), self.assertRaises(exception): request = ComponentCreateRequest(**body) ComponentService(FakeComponentRepository()).create_component( request, actor_user_id=uuid4() ) class FakeRepairPartRepository: def __init__(self): self.repair_id = uuid4() self.miner_id = uuid4() self.actor_id = uuid4() self.component_id = uuid4() self.miner = {"id": self.miner_id, "lifecycle_status": "in_repair"} self.repair = { "id": self.repair_id, "miner_id": self.miner_id, "status": "OPEN", } self.component = { "id": self.component_id, "component_type": "control_board", "asset_id": "CB-42", "manufacturer": "MicroBT", "model": None, "manufacturer_serial": "SER-42", } self.events = [{"sequence": 1}] self.parts = {} self.calls = [] self.fail_at = None @contextmanager def transaction(self): self.calls.append("transaction") snapshot = deepcopy((self.events, self.parts)) try: yield self except Exception: self.events, self.parts = snapshot raise def _call(self, name): self.calls.append(name) if self.fail_at == name: raise RuntimeError(name) def get_repair_identity(self, repair_id): self._call("repair_identity") return {"miner_id": self.miner_id} if repair_id == self.repair_id else None def lock_miner_for_update(self, miner_id): self._call("lock_miner") return self.miner if miner_id == self.miner_id else None def lock_repair_for_update(self, repair_id): self._call("lock_repair") return self.repair if repair_id == self.repair_id else None def get_active_repair(self, miner_id): self._call("active_repair") return self.repair if miner_id == self.miner_id else None def lock_component_for_update(self, component_id): self._call("lock_component") return self.component if component_id == self.component_id else None def get_next_event_sequence(self, repair_id): self._call("next_sequence") return max(event["sequence"] for event in self.events) + 1 def create_event(self, repair_id, **values): self._call("create_event") row = {"id": uuid4(), "repair_id": repair_id, "created_at": NOW, **values} self.events.append(row) return row def create_repair_part(self, **values): self._call("create_part") component = values["component"] row = { "id": values["repair_part_id"], "repair_id": values["repair_id"], "consumable_description": values["consumable_description"], "quantity": values["quantity"], "unit": values["unit"], "component_id": component["id"] if component else None, "component_type_snapshot": component["component_type"] if component else None, "component_asset_id_snapshot": component["asset_id"] if component else None, "component_manufacturer_snapshot": component["manufacturer"] if component else None, "component_model_snapshot": component["model"] if component else None, "component_manufacturer_serial_snapshot": component["manufacturer_serial"] if component else None, "notes": values["notes"], "recorded_by": values["recorded_event"]["created_by"], "recorded_at": values["recorded_event"]["created_at"], "recorded_event_id": values["recorded_event"]["id"], "voided_by": None, "voided_at": None, "void_reason": None, "void_event_id": None, } self.parts[row["id"]] = row return row def get_repair_part(self, repair_id, repair_part_id): self._call("get_part") row = self.parts.get(repair_part_id) return row if row and row["repair_id"] == repair_id else None def get_repair_parts(self, repair_id): rows = [row for row in self.parts.values() if row["repair_id"] == repair_id] return rows if repair_id == self.repair_id else None def lock_repair_part_for_update(self, repair_id, repair_part_id): self._call("lock_part") return self.get_repair_part(repair_id, repair_part_id) def void_repair_part(self, repair_id, repair_part_id, *, reason, void_event): self._call("void_part") row = self.get_repair_part(repair_id, repair_part_id) if row is None or row["voided_at"] is not None: return None row.update( voided_by=void_event["created_by"], voided_at=void_event["created_at"], void_reason=reason, void_event_id=void_event["id"], ) return row class RepairPartServiceTests(unittest.TestCase): def setUp(self): self.repository = FakeRepairPartRepository() self.service = RepairService(self.repository) def test_request_xor_precision_units_and_trim(self): valid = ConsumableInput(description=" Paste ", quantity="1.250", unit="gram") self.assertEqual(valid.description, "Paste") self.assertEqual(valid.quantity, Decimal("1.250")) invalid = ( {"consumable": None, "tracked_component_uuid": None}, {"consumable": valid, "tracked_component_uuid": uuid4()}, {"consumable": {"description": "x", "quantity": "1.0001", "unit": "gram"}}, {"consumable": {"description": "x", "quantity": "1.5", "unit": "each"}}, {"consumable": {"description": "x", "quantity": "1", "unit": "kg"}}, ) for body in invalid: with self.subTest(body=body), self.assertRaises(ValidationError): RepairPartRecordRequest(**body) def test_consumable_record_is_atomic_and_sequenced(self): result = self.service.record_repair_part( self.repository.repair_id, RepairPartRecordRequest( consumable={"description": " Paste ", "quantity": "2", "unit": "gram"}, notes=" Used on board ", ), actor_user_id=self.repository.actor_id, ) self.assertEqual(result.entry_kind, "consumable") self.assertEqual(result.consumable.description, "Paste") event = self.repository.events[-1] self.assertEqual(event["event_type"], "REPAIR_PART_RECORDED") self.assertEqual(event["sequence"], 2) self.assertEqual(event["created_by"], self.repository.actor_id) self.assertEqual(event["payload"]["quantity"], "2") self.assertEqual(self.repository.calls[:5], [ "transaction", "repair_identity", "lock_miner", "lock_repair", "active_repair" ]) def test_tracked_component_uses_locked_snapshot_without_mutation(self): before = deepcopy(self.repository.component) result = self.service.record_repair_part( self.repository.repair_id, RepairPartRecordRequest(tracked_component_uuid=self.repository.component_id), actor_user_id=self.repository.actor_id, ) self.assertEqual(result.component_id, self.repository.component_id) self.assertEqual(result.component_snapshot.asset_id, "CB-42") self.assertEqual(self.repository.component, before) self.assertLess( self.repository.calls.index("lock_repair"), self.repository.calls.index("lock_component"), ) def test_void_is_one_way_and_rollbacks_leave_no_partial_event(self): part = self.service.record_repair_part( self.repository.repair_id, RepairPartRecordRequest( consumable={"description": "Fuse", "quantity": 1, "unit": "each"} ), actor_user_id=self.repository.actor_id, ) voided = self.service.void_repair_part( self.repository.repair_id, part.repair_part_id, RepairPartVoidRequest(reason=" Wrong repair "), actor_user_id=self.repository.actor_id, ) self.assertEqual(voided.void_reason, "Wrong repair") with self.assertRaises(RepairPartAlreadyVoided): self.service.void_repair_part( self.repository.repair_id, part.repair_part_id, RepairPartVoidRequest(reason="Again"), actor_user_id=self.repository.actor_id, ) self.assertEqual( [event.get("event_type") for event in self.repository.events].count("REPAIR_PART_VOIDED"), 1, ) def test_insert_failure_rolls_back_event_and_part(self): self.repository.fail_at = "create_part" with self.assertRaises(RuntimeError): self.service.record_repair_part( self.repository.repair_id, RepairPartRecordRequest( consumable={"description": "Fuse", "quantity": 1, "unit": "each"} ), actor_user_id=self.repository.actor_id, ) self.assertEqual(self.repository.events, [{"sequence": 1}]) self.assertEqual(self.repository.parts, {}) def part_detail(repair_id, actor_id): return RepairPartDetail( repair_part_id=uuid4(), repair_id=repair_id, entry_kind="consumable", consumable=RepairPartConsumable(description="Fuse", quantity=1, unit="each"), component_id=None, component_snapshot=None, notes=None, recorded_by=actor_id, recorded_at=NOW, recorded_event_id=uuid4(), ) class StubComponentService: def __init__(self, actor): self.actor = actor def create_component(self, request, *, actor_user_id): self.actor_seen = actor_user_id return ComponentDetail( component_id=uuid4(), component_type=request.component_type, created_at=NOW, created_by=actor_user_id, ) def get_component(self, component_id): return ComponentDetail( component_id=component_id, component_type="fan", created_at=NOW, created_by=self.actor, ) class StubRepairService: def __init__(self, repair_id, actor): self.repair_id = repair_id self.actor = actor self.detail = part_detail(repair_id, actor) def record_repair_part(self, repair_id, request, *, actor_user_id): self.actor_seen = actor_user_id return self.detail def list_repair_parts(self, repair_id): return RepairPartList(repair_id=repair_id, entries=[self.detail]) def void_repair_part(self, repair_id, repair_part_id, request, *, actor_user_id): self.actor_seen = actor_user_id return self.detail.model_copy(update={ "voided_by": actor_user_id, "voided_at": NOW, "void_reason": request.reason, "void_event_id": uuid4(), }) class Task005RouterTests(unittest.TestCase): def setUp(self): self.actor = uuid4() self.repair_id = uuid4() self.component_service = StubComponentService(self.actor) self.repair_service = StubRepairService(self.repair_id, self.actor) self.app = FastAPI() self.app.include_router(component_router) self.app.include_router(repair_router) self.app.dependency_overrides[get_component_service] = lambda: self.component_service self.app.dependency_overrides[get_repair_service] = lambda: self.repair_service self.client = TestClient(self.app) def tearDown(self): self.client.close() def authenticate(self, permissions, role="Technician"): self.app.dependency_overrides[get_current_user] = lambda: { "id": str(self.actor), "role": role, "permissions": list(permissions) } def test_authentication_permissions_and_actor_authority(self): path = f"/api/repairs/{self.repair_id}/parts" body = {"consumable": {"description": "Fuse", "quantity": 1, "unit": "each"}} self.assertEqual(self.client.post(path, json=body).status_code, 401) self.authenticate(("repairs.read",), role="Read Only") self.assertEqual(self.client.post(path, json=body).status_code, 403) self.assertEqual(self.client.get(path).status_code, 200) self.authenticate(("repairs.parts.manage",)) self.assertEqual(self.client.post(path, json=body).status_code, 201) self.assertEqual(self.repair_service.actor_seen, self.actor) def test_component_routes_permissions_and_envelopes(self): self.authenticate(("components.read",), role="Read Only") self.assertEqual(self.client.post("/api/components", json={"component_type": "fan"}).status_code, 403) self.assertEqual(self.client.get(f"/api/components/{uuid4()}").status_code, 200) self.authenticate(("components.create",)) response = self.client.post("/api/components", json={"component_type": "fan"}) self.assertEqual(response.status_code, 201) self.assertEqual(response.json()["meta"], {}) self.assertIsNone(response.json()["error"]) def test_exact_routes_and_permission_mapping(self): component_routes = {(route.path, tuple(sorted(route.methods))) for route in component_router.routes} self.assertEqual(component_routes, { ("/api/components", ("POST",)), ("/api/components/{component_id}", ("GET",)), }) part_routes = { (route.path, tuple(sorted(route.methods))) for route in repair_router.routes if "/parts" in route.path } self.assertEqual(part_routes, { ("/api/repairs/{repair_id}/parts", ("GET",)), ("/api/repairs/{repair_id}/parts", ("POST",)), ("/api/repairs/{repair_id}/parts/{repair_part_id}/void", ("POST",)), }) repair_id = uuid4() part_id = uuid4() self.assertEqual(required_api_permission("GET", "/api/components"), "components.read") self.assertEqual(required_api_permission("POST", "/api/components"), "components.create") self.assertEqual(required_api_permission("GET", f"/api/repairs/{repair_id}/parts"), "repairs.read") self.assertEqual(required_api_permission("POST", f"/api/repairs/{repair_id}/parts"), "repairs.parts.manage") self.assertEqual(required_api_permission("POST", f"/api/repairs/{repair_id}/parts/{part_id}/void"), "repairs.parts.manage") def test_architecture_boundaries(self): root = Path(__file__).parents[1] / "app" for relative in ("components/router.py", "repairs/router.py"): source = (root / relative).read_text(encoding="utf-8") tree = ast.parse(source) imports = [ node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module ] self.assertFalse(any(name.startswith(("psycopg", "sqlalchemy")) for name in imports)) self.assertNotIn("SELECT ", source.upper()) self.assertNotIn(".COMMIT(", source.upper()) for relative in ( "components/service.py", "components/repository.py", "repairs/service.py", "repairs/repository.py", ): tree = ast.parse((root / relative).read_text(encoding="utf-8")) imports = [] for node in ast.walk(tree): if isinstance(node, ast.Import): imports.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: imports.append(node.module) self.assertFalse( any(name == "fastapi" or name.startswith("fastapi.") for name in imports) ) if __name__ == "__main__": unittest.main()