810 lines
28 KiB
Python
810 lines
28 KiB
Python
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()
|