import ast import inspect import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from uuid import UUID, uuid4 from fastapi import FastAPI from fastapi.testclient import TestClient from psycopg.rows import dict_row from starlette.routing import Match from app.identity import get_current_user, required_api_permission from app.repairs.dependencies import get_repair_service from app.repairs.exceptions import RepairNotFound from app.repairs.models import ( BeginRepairRequest, BeginRepairResult, RepairActor, RepairDetail, RepairTimeline, RepairTimelineEvent, ) from app.repairs.repository import RepairRepository from app.repairs.router import router from app.repairs.service import RepairService OPENED_AT = datetime(2026, 7, 29, 12, 0, tzinfo=timezone.utc) def detail_result( repair_id: UUID | None = None, miner_id: UUID | None = None, ) -> RepairDetail: return RepairDetail( repair_id=repair_id or uuid4(), miner_id=miner_id or uuid4(), miner_asset_id="ASSET-TIMELINE-001", status="OPEN", reason="INSPECTION", opened_at=OPENED_AT, opened_by=RepairActor( user_id=uuid4(), display_name="Timeline Technician", ), closed_at=None, closed_by=None, miner_lifecycle="in_repair", latest_event_sequence=2, ) def timeline_result(repair_id: UUID | None = None) -> RepairTimeline: return RepairTimeline( repair_id=repair_id or uuid4(), events=[ RepairTimelineEvent( event_id=uuid4(), sequence=1, event_type="REPAIR_STARTED", actor=RepairActor( user_id=uuid4(), display_name="Timeline Technician", ), occurred_at=OPENED_AT, payload={ "reason": "INSPECTION", "from_lifecycle": "active", "to_lifecycle": "in_repair", }, ), RepairTimelineEvent( event_id=uuid4(), sequence=2, event_type="INITIAL_NOTE", actor=RepairActor( user_id=uuid4(), display_name="Timeline Manager", ), occurred_at=OPENED_AT, payload={"note": "Inspect control board."}, ), ], ) class FakeTimelineRepository: def __init__(self): self.detail = None self.events = None self.detail_repair_id = None self.events_repair_id = None def get_repair_detail(self, repair_id): self.detail_repair_id = repair_id return self.detail def get_repair_events(self, repair_id): self.events_repair_id = repair_id return self.events class RepairTimelineServiceTests(unittest.TestCase): def setUp(self): self.repository = FakeTimelineRepository() self.service = RepairService(self.repository) self.repair_id = uuid4() self.miner_id = uuid4() self.opened_by = uuid4() self.closed_by = uuid4() def detail_row(self): return { "repair_id": self.repair_id, "miner_id": self.miner_id, "miner_asset_id": "ASSET-SERVICE-001", "status": "COMPLETED", "reason": "PSU", "opened_at": OPENED_AT, "opened_by_user_id": self.opened_by, "opened_by_display_name": "Opening Technician", "closed_at": OPENED_AT + timedelta(hours=2), "closed_by_user_id": self.closed_by, "closed_by_display_name": "Closing Manager", "miner_lifecycle": "active", "latest_event_sequence": 3, } def test_detail_maps_public_fields_and_actors(self): self.repository.detail = self.detail_row() detail = self.service.get_repair(self.repair_id) self.assertEqual(detail.repair_id, self.repair_id) self.assertEqual(detail.miner_id, self.miner_id) self.assertEqual(detail.miner_asset_id, "ASSET-SERVICE-001") self.assertEqual(detail.status, "COMPLETED") self.assertEqual(detail.reason, "PSU") self.assertEqual(detail.opened_by.user_id, self.opened_by) self.assertEqual(detail.closed_by.user_id, self.closed_by) self.assertEqual(detail.closed_by.display_name, "Closing Manager") self.assertEqual(detail.latest_event_sequence, 3) self.assertEqual(self.repository.detail_repair_id, self.repair_id) def test_detail_preserves_nullable_asset_and_empty_sequence(self): row = self.detail_row() row.update( { "miner_asset_id": None, "status": "OPEN", "closed_at": None, "closed_by_user_id": None, "closed_by_display_name": None, "miner_lifecycle": "in_repair", "latest_event_sequence": None, } ) self.repository.detail = row detail = self.service.get_repair(self.repair_id) self.assertIsNone(detail.miner_asset_id) self.assertIsNone(detail.closed_at) self.assertIsNone(detail.closed_by) self.assertIsNone(detail.latest_event_sequence) def test_missing_detail_raises_domain_not_found(self): with self.assertRaises(RepairNotFound): self.service.get_repair(self.repair_id) def test_timeline_maps_structured_events_without_reordering(self): actor_id = uuid4() self.repository.events = [ { "event_id": uuid4(), "sequence": 1, "event_type": "REPAIR_STARTED", "actor_user_id": actor_id, "actor_display_name": "Timeline Actor", "occurred_at": OPENED_AT, "payload": { "reason": "OTHER", "from_lifecycle": "spare", "to_lifecycle": "in_repair", }, } ] timeline = self.service.get_repair_timeline(self.repair_id) self.assertEqual(timeline.repair_id, self.repair_id) self.assertEqual(len(timeline.events), 1) self.assertEqual(timeline.events[0].sequence, 1) self.assertEqual(timeline.events[0].actor.user_id, actor_id) self.assertIsInstance(timeline.events[0].payload, dict) self.assertEqual( timeline.events[0].payload["from_lifecycle"], "spare", ) def test_empty_timeline_is_successful(self): self.repository.events = [] timeline = self.service.get_repair_timeline(self.repair_id) self.assertEqual(timeline.events, []) def test_missing_timeline_raises_domain_not_found(self): with self.assertRaises(RepairNotFound): self.service.get_repair_timeline(self.repair_id) class RepairTimelineRepositoryTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.repository = RepairRepository() def setUp(self): self.opened_actor = uuid4() self.note_actor = uuid4() with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute("DELETE FROM repair_events") cur.execute("DELETE FROM repairs") cur.execute("DELETE FROM miner_tags") cur.execute("DELETE FROM miner_asset_id_history") cur.execute("DELETE FROM miners") cur.execute( "DELETE FROM users WHERE username LIKE " "'repair-timeline-test-%'" ) for actor_id, suffix, display_name in ( ( self.opened_actor, "opened", "Timeline Opening Technician", ), ( self.note_actor, "note", "Timeline Note Manager", ), ): cur.execute( """ INSERT INTO users ( id, username, display_name, password_hash, role_id ) SELECT %s, %s, %s, 'not-used', id FROM roles WHERE name = 'Admin' """, ( actor_id, f"repair-timeline-test-{suffix}", display_name, ), ) def create_miner(self, asset_id="ASSET-TIMELINE-DB"): with self.repository._connection_scope(write=True) as connection: with connection.cursor(row_factory=dict_row) as cur: cur.execute( """ INSERT INTO miners ( asset_tag, status, lifecycle_status, first_seen_at, last_seen_at ) VALUES (%s, 'unknown', 'in_repair', NULL, NULL) RETURNING id """, (asset_id,), ) return cur.fetchone()["id"] def create_repair(self, miner_id): with self.repository._connection_scope(write=True) as connection: with connection.cursor(row_factory=dict_row) as cur: cur.execute( """ INSERT INTO repairs (miner_id, reason, opened_by) VALUES (%s, 'INSPECTION', %s) RETURNING id """, (miner_id, self.opened_actor), ) return cur.fetchone()["id"] def insert_non_monotonic_events(self, repair_id): later = OPENED_AT + timedelta(hours=1) earlier = OPENED_AT with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( """ INSERT INTO repair_events ( repair_id, sequence, event_type, created_at, created_by, payload ) VALUES ( %s, 1, 'REPAIR_STARTED', %s, %s, %s::jsonb ), ( %s, 2, 'INITIAL_NOTE', %s, %s, %s::jsonb ) """, ( repair_id, later, self.opened_actor, '{"reason":"INSPECTION","from_lifecycle":' '"active","to_lifecycle":"in_repair"}', repair_id, earlier, self.note_actor, '{"note":"Timestamp is earlier than sequence one."}', ), ) def counts(self): with self.repository._connection_scope() as connection: with connection.cursor() as cur: cur.execute( """ SELECT (SELECT count(*) FROM repairs), (SELECT count(*) FROM repair_events) """ ) return cur.fetchone() def test_detail_joins_asset_lifecycle_actor_and_latest_sequence(self): miner_id = self.create_miner() repair_id = self.create_repair(miner_id) self.insert_non_monotonic_events(repair_id) detail = self.repository.get_repair_detail(repair_id) self.assertEqual(detail["repair_id"], repair_id) self.assertEqual(detail["miner_id"], miner_id) self.assertEqual(detail["miner_asset_id"], "ASSET-TIMELINE-DB") self.assertEqual(detail["miner_lifecycle"], "in_repair") self.assertEqual(detail["opened_by_user_id"], self.opened_actor) self.assertEqual( detail["opened_by_display_name"], "Timeline Opening Technician", ) self.assertIsNone(detail["closed_at"]) self.assertIsNone(detail["closed_by_user_id"]) self.assertEqual(detail["latest_event_sequence"], 2) def test_events_order_by_sequence_not_timestamp_and_join_actors(self): miner_id = self.create_miner() repair_id = self.create_repair(miner_id) self.insert_non_monotonic_events(repair_id) events = self.repository.get_repair_events(repair_id) self.assertEqual([event["sequence"] for event in events], [1, 2]) self.assertGreater( events[0]["occurred_at"], events[1]["occurred_at"], ) self.assertEqual( [event["actor_display_name"] for event in events], [ "Timeline Opening Technician", "Timeline Note Manager", ], ) self.assertIsInstance(events[0]["payload"], dict) self.assertEqual( events[1]["payload"], {"note": "Timestamp is earlier than sequence one."}, ) def test_known_repair_with_no_events_returns_empty_timeline(self): repair_id = self.create_repair(self.create_miner()) self.assertEqual(self.repository.get_repair_events(repair_id), []) detail = self.repository.get_repair_detail(repair_id) self.assertIsNone(detail["latest_event_sequence"]) def test_missing_repair_is_distinct_from_empty_timeline(self): repair_id = uuid4() self.assertIsNone(self.repository.get_repair_detail(repair_id)) self.assertIsNone(self.repository.get_repair_events(repair_id)) def test_reads_do_not_modify_repairs_or_events(self): repair_id = self.create_repair(self.create_miner()) self.insert_non_monotonic_events(repair_id) before = self.counts() self.repository.get_repair_detail(repair_id) self.repository.get_repair_events(repair_id) self.assertEqual(self.counts(), before) class StubTimelineService: def __init__(self): self.detail = None self.timeline = None self.error = None self.detail_repair_id = None self.timeline_repair_id = None self.begin_called = False def get_repair(self, repair_id): self.detail_repair_id = repair_id if self.error is not None: raise self.error return self.detail def get_repair_timeline(self, repair_id): self.timeline_repair_id = repair_id if self.error is not None: raise self.error return self.timeline def begin_repair(self, request: BeginRepairRequest, *, actor_user_id): self.begin_called = True return BeginRepairResult( repair_id=uuid4(), miner_id=request.miner_id, status="OPEN", reason=request.reason, opened_at=OPENED_AT, opened_by=actor_user_id, ) class RepairTimelineRouterTests(unittest.TestCase): def setUp(self): self.service = StubTimelineService() self.app = FastAPI() self.app.include_router(router) self.app.dependency_overrides[get_repair_service] = ( lambda: self.service ) self.client = TestClient(self.app) def tearDown(self): self.client.close() def authenticate( self, role="Read Only", permissions=("repairs.read",), ): self.app.dependency_overrides[get_current_user] = lambda: { "id": str(uuid4()), "username": role.lower().replace(" ", "-"), "role": role, "permissions": list(permissions), } def test_timeline_routes_require_authentication(self): for path in ( f"/api/repairs/{uuid4()}", f"/api/repairs/{uuid4()}/events", ): with self.subTest(path=path): response = self.client.get(path) self.assertEqual(response.status_code, 401) def test_timeline_routes_require_repairs_read(self): self.authenticate(permissions=("repairs.view",)) for path in ( f"/api/repairs/{uuid4()}", f"/api/repairs/{uuid4()}/events", ): with self.subTest(path=path): response = self.client.get(path) self.assertEqual(response.status_code, 403) def test_all_default_roles_can_read_detail_and_events(self): repair_id = uuid4() self.service.detail = detail_result(repair_id) self.service.timeline = timeline_result(repair_id) for role in ("Admin", "Manager", "Technician", "Read Only"): with self.subTest(role=role): self.authenticate(role=role) self.assertEqual( self.client.get( f"/api/repairs/{repair_id}" ).status_code, 200, ) self.assertEqual( self.client.get( f"/api/repairs/{repair_id}/events" ).status_code, 200, ) def test_malformed_repair_uuid_returns_validation_error(self): self.authenticate() for path in ( "/api/repairs/not-a-uuid", "/api/repairs/not-a-uuid/events", ): with self.subTest(path=path): response = self.client.get(path) self.assertEqual(response.status_code, 422) def test_missing_repair_returns_not_found(self): self.authenticate() repair_id = uuid4() self.service.error = RepairNotFound(repair_id) for path in ( f"/api/repairs/{repair_id}", f"/api/repairs/{repair_id}/events", ): with self.subTest(path=path): response = self.client.get(path) self.assertEqual(response.status_code, 404) self.assertEqual( response.json()["detail"], "Repair not found", ) def test_detail_returns_typed_public_envelope(self): self.authenticate() repair_id = uuid4() expected = detail_result(repair_id) self.service.detail = expected response = self.client.get(f"/api/repairs/{repair_id}") self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["repair_id"], str(repair_id)) self.assertEqual(body["data"]["miner_asset_id"], "ASSET-TIMELINE-001") self.assertEqual(body["data"]["miner_lifecycle"], "in_repair") self.assertEqual(body["data"]["latest_event_sequence"], 2) self.assertEqual( body["data"]["opened_by"]["display_name"], "Timeline Technician", ) self.assertIsNone(body["data"]["closed_at"]) self.assertIsNone(body["data"]["closed_by"]) self.assertEqual(body["meta"], {}) self.assertIsNone(body["error"]) def test_events_return_structured_payload_in_service_order(self): self.authenticate() repair_id = uuid4() expected = timeline_result(repair_id) self.service.timeline = expected response = self.client.get( f"/api/repairs/{repair_id}/events" ) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["repair_id"], str(repair_id)) self.assertEqual( [event["sequence"] for event in body["data"]["events"]], [1, 2], ) self.assertIsInstance(body["data"]["events"][0]["payload"], dict) self.assertEqual( body["data"]["events"][1]["actor"]["display_name"], "Timeline Manager", ) def test_known_empty_timeline_returns_success(self): self.authenticate() repair_id = uuid4() self.service.timeline = RepairTimeline( repair_id=repair_id, events=[], ) response = self.client.get( f"/api/repairs/{repair_id}/events" ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["events"], []) def test_read_only_cannot_begin_repair(self): self.authenticate(role="Read Only", permissions=("repairs.read",)) response = self.client.post( "/api/repairs", json={"miner_id": str(uuid4()), "reason": "OTHER"}, ) self.assertEqual(response.status_code, 403) self.assertFalse(self.service.begin_called) def test_exactly_two_timeline_get_routes_are_registered(self): get_routes = { route.path for route in router.routes if hasattr(route, "methods") and "GET" in route.methods } self.assertEqual( get_routes, { "/api/repairs/{repair_id}", "/api/repairs/{repair_id}/events", }, ) class RepairTimelineArchitectureTests(unittest.TestCase): @staticmethod def full_route_matches(app, method, path): scope = { "type": "http", "method": method, "path": path, "root_path": "", } return [ route for route in app.routes if route.matches(scope)[0] is Match.FULL ] def test_permission_mapper_scopes_new_and_legacy_routes(self): repair_id = uuid4() self.assertEqual( required_api_permission( "GET", f"/api/repairs/{repair_id}", ), "repairs.read", ) self.assertEqual( required_api_permission( "GET", f"/api/repairs/{repair_id}/events", ), "repairs.read", ) self.assertEqual( required_api_permission("GET", "/api/repairs/queue"), "repairs.view", ) self.assertEqual( required_api_permission( "GET", "/api/repairs/parts-summary", ), "repairs.view", ) self.assertEqual( required_api_permission("POST", "/api/repairs"), "repairs.create", ) def test_repair_router_is_registered_after_legacy_static_routes(self): main_path = Path(__file__).parents[1] / "app" / "main.py" source = main_path.read_text(encoding="utf-8") self.assertGreater( source.rfind("app.include_router(repairs_router)"), source.find('@app.get("/api/repairs/parts-summary")'), ) def test_live_application_repair_route_inventory(self): from app.main import app routes = { (method, route.path, route.endpoint.__name__) for route in app.routes if route.path.startswith("/api/repairs") for method in route.methods } self.assertEqual( routes, { ("GET", "/api/repairs/queue", "minerops_repair_queue_api"), ( "POST", "/api/repairs/{miner_id}/note", "minerops_repair_add_note_api", ), ( "POST", "/api/repairs/{miner_id}/status", "minerops_repair_update_status_api", ), ( "POST", "/api/repairs/{miner_id}/state", "minerops_repair_update_state_api", ), ( "GET", "/api/repairs/parts-summary", "minerops_repair_parts_summary_api", ), ("POST", "/api/repairs", "begin_repair"), ( "POST", "/api/repairs/{repair_id}/complete", "complete_repair", ), ("GET", "/api/repairs/{repair_id}", "get_repair"), ( "GET", "/api/repairs/{repair_id}/events", "get_repair_events", ), }, ) def test_static_repair_routes_resolve_before_uuid_detail_route(self): from app.main import app for path, endpoint_name in ( ("/api/repairs/queue", "minerops_repair_queue_api"), ( "/api/repairs/parts-summary", "minerops_repair_parts_summary_api", ), ): with self.subTest(path=path): matches = self.full_route_matches(app, "GET", path) self.assertGreaterEqual(len(matches), 2) self.assertEqual(matches[0].path, path) self.assertEqual(matches[0].endpoint.__name__, endpoint_name) self.assertEqual(matches[1].endpoint.__name__, "get_repair") def test_uuid_repair_routes_resolve_to_timeline_handlers(self): from app.main import app repair_id = uuid4() for path, endpoint_name in ( (f"/api/repairs/{repair_id}", "get_repair"), ( f"/api/repairs/{repair_id}/events", "get_repair_events", ), ): with self.subTest(path=path): matches = self.full_route_matches(app, "GET", path) self.assertEqual(len(matches), 1) self.assertEqual(matches[0].endpoint.__name__, endpoint_name) def test_repository_event_read_is_one_joined_ordered_query(self): source = inspect.getsource(RepairRepository.get_repair_events) self.assertEqual(source.count("cur.execute("), 1) self.assertIn("JOIN users AS actor", source) self.assertIn("ORDER BY event.sequence ASC", source) for keyword in ("INSERT ", "UPDATE ", "DELETE "): self.assertNotIn(keyword, source.upper()) def test_service_and_repository_have_no_fastapi_dependency(self): repairs_path = Path(__file__).parents[1] / "app" / "repairs" for filename in ("service.py", "repository.py"): tree = ast.parse( (repairs_path / filename).read_text(encoding="utf-8") ) modules = [] for node in ast.walk(tree): 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( module == "fastapi" or module.startswith("fastapi.") for module in modules ), filename, ) def test_timeline_has_no_mutation_route_or_repository_method(self): methods = { method for route in router.routes if hasattr(route, "methods") for method in route.methods } self.assertNotIn("PATCH", methods) self.assertNotIn("PUT", methods) self.assertNotIn("DELETE", methods) self.assertFalse(hasattr(RepairRepository, "update_repair_event")) self.assertFalse(hasattr(RepairRepository, "delete_repair_event")) if __name__ == "__main__": unittest.main()