456 lines
18 KiB
Python
456 lines
18 KiB
Python
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()
|