import ast import unittest from datetime import datetime, timezone from pathlib import Path from uuid import UUID, uuid4 from fastapi import FastAPI from fastapi.testclient import TestClient from app.identity import get_current_user from app.repairs.dependencies import get_repair_service from app.repairs.exceptions import ( ActiveRepairConflict, IncompatibleMinerLifecycle, InconsistentRepairState, RepairMinerNotFound, ) from app.repairs.models import ( BeginRepairRequest, BeginRepairResult, ) from app.repairs.router import router class StubRepairService: def __init__(self): self.result = None self.error = None self.request = None self.actor_user_id = None def begin_repair( self, request: BeginRepairRequest, *, actor_user_id: UUID, ) -> BeginRepairResult: self.request = request self.actor_user_id = actor_user_id if self.error is not None: raise self.error if self.result is None: raise AssertionError("result was not configured") return self.result def repair_result( miner_id=None, actor_user_id=None, ) -> BeginRepairResult: return BeginRepairResult( repair_id=uuid4(), miner_id=miner_id or uuid4(), status="OPEN", reason="CONTROL_BOARD", opened_at=datetime(2026, 7, 29, 12, 0, tzinfo=timezone.utc), opened_by=actor_user_id or uuid4(), ) class RepairRouterTests(unittest.TestCase): def setUp(self): self.service = StubRepairService() self.app = FastAPI() self.app.include_router(router) self.app.dependency_overrides[get_repair_service] = ( lambda: self.service ) self.client = TestClient(self.app) self.actor_user_id = uuid4() def tearDown(self): self.client.close() def authenticate(self, permissions=("repairs.create",)): self.app.dependency_overrides[get_current_user] = lambda: { "id": str(self.actor_user_id), "username": "technician", "role": "Technician", "permissions": list(permissions), } def test_begin_repair_requires_authentication(self): response = self.client.post( "/api/repairs", json={ "miner_id": str(uuid4()), "reason": "CONTROL_BOARD", }, ) self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["detail"], "Login required") def test_begin_repair_requires_repairs_create(self): self.authenticate(permissions=("repairs.view", "repairs.manage")) response = self.client.post( "/api/repairs", json={ "miner_id": str(uuid4()), "reason": "CONTROL_BOARD", }, ) self.assertEqual(response.status_code, 403) self.assertIsNone(self.service.request) def test_malformed_miner_uuid_returns_validation_error(self): self.authenticate() response = self.client.post( "/api/repairs", json={ "miner_id": "not-a-uuid", "reason": "CONTROL_BOARD", }, ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.request) def test_invalid_reason_returns_validation_error(self): self.authenticate() response = self.client.post( "/api/repairs", json={ "miner_id": str(uuid4()), "reason": "FREE_FORM_REASON", }, ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.request) def test_unknown_miner_returns_not_found(self): self.authenticate() miner_id = uuid4() self.service.error = RepairMinerNotFound(miner_id) response = self.client.post( "/api/repairs", json={ "miner_id": str(miner_id), "reason": "OTHER", }, ) self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["detail"], "Miner not found") def test_active_repair_returns_conflict(self): self.authenticate() miner_id = uuid4() self.service.error = ActiveRepairConflict(miner_id) response = self.client.post( "/api/repairs", json={ "miner_id": str(miner_id), "reason": "PSU", }, ) self.assertEqual(response.status_code, 409) self.assertIn("active repair", response.json()["detail"]) def test_lifecycle_conflicts_return_conflict(self): self.authenticate() miner_id = uuid4() for error in ( IncompatibleMinerLifecycle(miner_id, "retired"), InconsistentRepairState(miner_id), ): with self.subTest(error=error): self.service.error = error response = self.client.post( "/api/repairs", json={ "miner_id": str(miner_id), "reason": "OTHER", }, ) self.assertEqual(response.status_code, 409) def test_success_returns_typed_envelope_and_authenticated_actor(self): self.authenticate() miner_id = uuid4() expected = repair_result(miner_id, self.actor_user_id) self.service.result = expected response = self.client.post( "/api/repairs", json={ "miner_id": str(miner_id), "reason": "CONTROL_BOARD", "notes": "Miner will not boot.", }, ) self.assertEqual(response.status_code, 201) body = response.json() self.assertEqual(body["data"]["repair_id"], str(expected.repair_id)) self.assertEqual(body["data"]["miner_id"], str(miner_id)) self.assertEqual(body["data"]["status"], "OPEN") self.assertEqual(body["data"]["reason"], "CONTROL_BOARD") self.assertEqual(body["data"]["opened_by"], str(self.actor_user_id)) self.assertEqual( body["data"]["opened_at"], "2026-07-29T12:00:00Z", ) self.assertEqual(body["meta"], {}) self.assertIsNone(body["error"]) self.assertEqual(self.service.actor_user_id, self.actor_user_id) self.assertEqual( self.service.request.notes, "Miner will not boot.", ) def test_client_cannot_override_server_fields_or_actor(self): self.authenticate() base = { "miner_id": str(uuid4()), "reason": "OTHER", } forbidden = { "repair_id": str(uuid4()), "status": "COMPLETED", "opened_at": "2026-01-01T00:00:00Z", "opened_by": str(uuid4()), "event_id": str(uuid4()), "event_timestamp": "2026-01-01T00:00:00Z", "event_actor": str(uuid4()), "lifecycle_status": "retired", } for field, value in forbidden.items(): with self.subTest(field=field): response = self.client.post( "/api/repairs", json={**base, field: value}, ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.request) def test_router_registers_exactly_one_post_route(self): routes = [ (route.path, route.methods) for route in router.routes if hasattr(route, "methods") and "POST" in route.methods ] self.assertEqual(routes, [("/api/repairs", {"POST"})]) def test_router_has_no_sql_driver_or_transaction_management(self): router_path = ( Path(__file__).parents[1] / "app" / "repairs" / "router.py" ) source = router_path.read_text(encoding="utf-8") tree = ast.parse(source) imported_modules = [] for node in ast.walk(tree): if isinstance(node, ast.Import): imported_modules.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: imported_modules.append(node.module) database_modules = ("psycopg", "asyncpg", "sqlalchemy") self.assertFalse( any( module == name or module.startswith(f"{name}.") for module in imported_modules for name in database_modules ) ) upper_source = source.upper() for keyword in ( "SELECT ", "INSERT ", "UPDATE ", "DELETE ", ".COMMIT(", ".ROLLBACK(", ): self.assertNotIn(keyword, upper_source) if __name__ == "__main__": unittest.main()