738 lines
25 KiB
Python
738 lines
25 KiB
Python
import ast
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import contextmanager
|
|
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from threading import Barrier
|
|
from uuid import UUID, uuid4
|
|
|
|
from psycopg.errors import RaiseException
|
|
from psycopg.rows import dict_row
|
|
|
|
from app.repairs import (
|
|
ActiveRepairConflict,
|
|
BeginRepairRequest,
|
|
IncompatibleMinerLifecycle,
|
|
InconsistentRepairState,
|
|
RepairActorRequired,
|
|
RepairMinerNotFound,
|
|
RepairRepository,
|
|
RepairService,
|
|
)
|
|
|
|
|
|
def repair_row(miner_id, actor_user_id, reason="CONTROL_BOARD"):
|
|
return {
|
|
"id": uuid4(),
|
|
"miner_id": miner_id,
|
|
"status": "OPEN",
|
|
"reason": reason,
|
|
"opened_at": datetime.now(timezone.utc),
|
|
"opened_by": actor_user_id,
|
|
"completed_at": None,
|
|
"completed_by": None,
|
|
}
|
|
|
|
|
|
class FakeRepairRepository:
|
|
def __init__(self):
|
|
self.miners = {}
|
|
self.repairs = []
|
|
self.events = []
|
|
self.calls = []
|
|
self.create_error = None
|
|
self.event_error = None
|
|
self.lifecycle_error = None
|
|
|
|
@contextmanager
|
|
def transaction(self):
|
|
self.calls.append("transaction")
|
|
snapshot = deepcopy((self.miners, self.repairs, self.events))
|
|
try:
|
|
yield self
|
|
except Exception:
|
|
self.miners, self.repairs, self.events = snapshot
|
|
raise
|
|
|
|
def lock_miner_for_update(self, miner_id):
|
|
self.calls.append("lock_miner")
|
|
return self.miners.get(miner_id)
|
|
|
|
def get_active_repair(self, miner_id):
|
|
self.calls.append("get_active_repair")
|
|
return next(
|
|
(
|
|
repair
|
|
for repair in self.repairs
|
|
if repair["miner_id"] == miner_id
|
|
and repair["status"]
|
|
in {
|
|
"OPEN",
|
|
"IN_PROGRESS",
|
|
"WAITING_FOR_PARTS",
|
|
"READY_FOR_TEST",
|
|
}
|
|
),
|
|
None,
|
|
)
|
|
|
|
def create_repair(self, miner_id, reason, opened_by):
|
|
self.calls.append("create_repair")
|
|
if self.create_error is not None:
|
|
raise self.create_error
|
|
row = repair_row(miner_id, opened_by, reason)
|
|
self.repairs.append(row)
|
|
return row
|
|
|
|
def create_event(
|
|
self,
|
|
repair_id,
|
|
*,
|
|
sequence,
|
|
event_type,
|
|
created_by,
|
|
payload,
|
|
):
|
|
self.calls.append(f"create_event:{event_type}")
|
|
if self.event_error is not None:
|
|
raise self.event_error
|
|
event = {
|
|
"id": uuid4(),
|
|
"repair_id": repair_id,
|
|
"sequence": sequence,
|
|
"event_type": event_type,
|
|
"created_at": datetime.now(timezone.utc),
|
|
"created_by": created_by,
|
|
"payload": dict(payload),
|
|
}
|
|
self.events.append(event)
|
|
return event
|
|
|
|
def set_miner_lifecycle(self, miner_id, lifecycle_status):
|
|
self.calls.append("set_miner_lifecycle")
|
|
if self.lifecycle_error is not None:
|
|
raise self.lifecycle_error
|
|
self.miners[miner_id]["lifecycle_status"] = lifecycle_status
|
|
|
|
|
|
class RepairServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.repository = FakeRepairRepository()
|
|
self.service = RepairService(self.repository)
|
|
self.actor_user_id = uuid4()
|
|
|
|
def add_miner(self, lifecycle_status="active"):
|
|
miner_id = uuid4()
|
|
self.repository.miners[miner_id] = {
|
|
"id": miner_id,
|
|
"lifecycle_status": lifecycle_status,
|
|
}
|
|
return miner_id
|
|
|
|
def test_begin_repair_creates_record_event_and_lifecycle_atomically(self):
|
|
miner_id = self.add_miner()
|
|
|
|
result = self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="CONTROL_BOARD",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertEqual(result.miner_id, miner_id)
|
|
self.assertEqual(result.status, "OPEN")
|
|
self.assertEqual(result.reason, "CONTROL_BOARD")
|
|
self.assertEqual(result.opened_by, self.actor_user_id)
|
|
self.assertEqual(len(self.repository.repairs), 1)
|
|
self.assertEqual(len(self.repository.events), 1)
|
|
self.assertEqual(
|
|
self.repository.events[0]["event_type"],
|
|
"REPAIR_STARTED",
|
|
)
|
|
self.assertEqual(
|
|
self.repository.events[0]["payload"],
|
|
{
|
|
"reason": "CONTROL_BOARD",
|
|
"from_lifecycle": "active",
|
|
"to_lifecycle": "in_repair",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
self.repository.miners[miner_id]["lifecycle_status"],
|
|
"in_repair",
|
|
)
|
|
self.assertEqual(
|
|
self.repository.calls,
|
|
[
|
|
"transaction",
|
|
"lock_miner",
|
|
"get_active_repair",
|
|
"create_repair",
|
|
"create_event:REPAIR_STARTED",
|
|
"set_miner_lifecycle",
|
|
],
|
|
)
|
|
|
|
def test_begin_repair_normalizes_and_records_initial_note(self):
|
|
miner_id = self.add_miner("spare")
|
|
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="INSPECTION",
|
|
notes=" Check fan harness. ",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[event["event_type"] for event in self.repository.events],
|
|
["REPAIR_STARTED", "INITIAL_NOTE"],
|
|
)
|
|
self.assertEqual(
|
|
[event["sequence"] for event in self.repository.events],
|
|
[1, 2],
|
|
)
|
|
self.assertEqual(
|
|
self.repository.events[0]["payload"],
|
|
{
|
|
"reason": "INSPECTION",
|
|
"from_lifecycle": "spare",
|
|
"to_lifecycle": "in_repair",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
self.repository.events[1]["payload"],
|
|
{"note": "Check fan harness."},
|
|
)
|
|
|
|
def test_absent_and_blank_notes_create_no_note_event(self):
|
|
for note in (None, " \t\n "):
|
|
with self.subTest(note=note):
|
|
repository = FakeRepairRepository()
|
|
miner_id = uuid4()
|
|
repository.miners[miner_id] = {
|
|
"id": miner_id,
|
|
"lifecycle_status": "active",
|
|
}
|
|
RepairService(repository).begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
notes=note,
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
self.assertEqual(len(repository.events), 1)
|
|
|
|
def test_unknown_miner_raises_not_found_without_writes(self):
|
|
with self.assertRaises(RepairMinerNotFound):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=uuid4(),
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
self.assertEqual(self.repository.repairs, [])
|
|
self.assertEqual(self.repository.events, [])
|
|
|
|
def test_active_repair_conflict_prevents_partial_changes(self):
|
|
miner_id = self.add_miner()
|
|
self.repository.repairs.append(
|
|
repair_row(miner_id, self.actor_user_id)
|
|
)
|
|
snapshot = deepcopy(self.repository.repairs)
|
|
|
|
with self.assertRaises(ActiveRepairConflict):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="PSU",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertEqual(self.repository.repairs, snapshot)
|
|
self.assertEqual(self.repository.events, [])
|
|
self.assertEqual(
|
|
self.repository.miners[miner_id]["lifecycle_status"],
|
|
"active",
|
|
)
|
|
|
|
def test_in_repair_without_active_record_is_inconsistent(self):
|
|
miner_id = self.add_miner("in_repair")
|
|
with self.assertRaises(InconsistentRepairState):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
def test_incompatible_lifecycle_is_rejected(self):
|
|
miner_id = self.add_miner("retired")
|
|
with self.assertRaises(IncompatibleMinerLifecycle):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
def test_actor_is_required_before_transaction_starts(self):
|
|
miner_id = self.add_miner()
|
|
with self.assertRaises(RepairActorRequired):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=None,
|
|
)
|
|
self.assertEqual(self.repository.calls, [])
|
|
|
|
def test_repository_domain_conflict_propagates(self):
|
|
miner_id = self.add_miner()
|
|
self.repository.create_error = ActiveRepairConflict(miner_id)
|
|
with self.assertRaises(ActiveRepairConflict):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
def test_event_failure_rolls_back_every_change(self):
|
|
miner_id = self.add_miner()
|
|
self.repository.event_error = RuntimeError("event failed")
|
|
with self.assertRaisesRegex(RuntimeError, "event failed"):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
self.assertEqual(self.repository.repairs, [])
|
|
self.assertEqual(self.repository.events, [])
|
|
self.assertEqual(
|
|
self.repository.miners[miner_id]["lifecycle_status"],
|
|
"active",
|
|
)
|
|
|
|
def test_lifecycle_failure_rolls_back_repair_and_events(self):
|
|
miner_id = self.add_miner()
|
|
self.repository.lifecycle_error = RuntimeError("lifecycle failed")
|
|
with self.assertRaisesRegex(RuntimeError, "lifecycle failed"):
|
|
self.service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
notes="Initial note",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
self.assertEqual(self.repository.repairs, [])
|
|
self.assertEqual(self.repository.events, [])
|
|
self.assertEqual(
|
|
self.repository.miners[miner_id]["lifecycle_status"],
|
|
"active",
|
|
)
|
|
|
|
|
|
class RepairRepositoryIntegrationTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.repository = RepairRepository()
|
|
cls.actor_user_id = UUID(
|
|
"00000000-0000-0000-0000-000000001101"
|
|
)
|
|
|
|
def setUp(self):
|
|
with self.repository._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"DROP TRIGGER IF EXISTS test_repair_event_failure "
|
|
"ON repair_events"
|
|
)
|
|
cur.execute(
|
|
"DROP TRIGGER IF EXISTS test_repair_lifecycle_failure "
|
|
"ON miners"
|
|
)
|
|
cur.execute(
|
|
"DROP FUNCTION IF EXISTS test_fail_repair_event()"
|
|
)
|
|
cur.execute(
|
|
"DROP FUNCTION IF EXISTS test_fail_repair_lifecycle()"
|
|
)
|
|
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 = 'repair-domain-test'"
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO users (
|
|
id, username, display_name, password_hash, role_id
|
|
)
|
|
SELECT %s, %s, %s, %s, id
|
|
FROM roles
|
|
WHERE name = 'Admin'
|
|
""",
|
|
(
|
|
self.actor_user_id,
|
|
"repair-domain-test",
|
|
"Repair Domain Test",
|
|
"not-used",
|
|
),
|
|
)
|
|
|
|
def create_miner(self, lifecycle_status="active"):
|
|
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', %s, NULL, NULL)
|
|
RETURNING id
|
|
""",
|
|
(f"REPAIR-{uuid4()}", lifecycle_status),
|
|
)
|
|
return cur.fetchone()["id"]
|
|
|
|
def fetch_events(self, repair_id):
|
|
with self.repository._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT *
|
|
FROM repair_events
|
|
WHERE repair_id = %s
|
|
ORDER BY sequence
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
def count_repairs(self, miner_id):
|
|
with self.repository._connection_scope() as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT COUNT(*) FROM repairs WHERE miner_id = %s",
|
|
(miner_id,),
|
|
)
|
|
return cur.fetchone()[0]
|
|
|
|
def lifecycle(self, miner_id):
|
|
with self.repository._connection_scope() as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT lifecycle_status FROM miners WHERE id = %s",
|
|
(miner_id,),
|
|
)
|
|
return cur.fetchone()[0]
|
|
|
|
def test_begin_repair_persists_actor_reason_timestamp_and_started_event(self):
|
|
miner_id = self.create_miner()
|
|
before = datetime.now(timezone.utc)
|
|
|
|
result = RepairService(self.repository).begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="HASHBOARD",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertGreaterEqual(result.opened_at, before)
|
|
self.assertEqual(result.opened_by, self.actor_user_id)
|
|
self.assertEqual(result.reason, "HASHBOARD")
|
|
self.assertEqual(result.status, "OPEN")
|
|
self.assertEqual(self.lifecycle(miner_id), "in_repair")
|
|
active = self.repository.get_active_repair(miner_id)
|
|
self.assertEqual(active["id"], result.repair_id)
|
|
events = self.fetch_events(result.repair_id)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["event_type"], "REPAIR_STARTED")
|
|
self.assertEqual(events[0]["sequence"], 1)
|
|
self.assertEqual(events[0]["created_by"], self.actor_user_id)
|
|
self.assertIsNotNone(events[0]["created_at"])
|
|
self.assertEqual(
|
|
events[0]["payload"],
|
|
{
|
|
"reason": "HASHBOARD",
|
|
"from_lifecycle": "active",
|
|
"to_lifecycle": "in_repair",
|
|
},
|
|
)
|
|
|
|
def test_initial_note_is_normalized_and_ordered_after_started_event(self):
|
|
miner_id = self.create_miner("spare")
|
|
result = RepairService(self.repository).begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="INSPECTION",
|
|
notes=" Inspect harness. ",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
events = self.fetch_events(result.repair_id)
|
|
self.assertEqual(
|
|
[(event["sequence"], event["event_type"]) for event in events],
|
|
[(1, "REPAIR_STARTED"), (2, "INITIAL_NOTE")],
|
|
)
|
|
self.assertEqual(
|
|
events[0]["payload"],
|
|
{
|
|
"reason": "INSPECTION",
|
|
"from_lifecycle": "spare",
|
|
"to_lifecycle": "in_repair",
|
|
},
|
|
)
|
|
self.assertEqual(events[1]["payload"], {"note": "Inspect harness."})
|
|
|
|
def test_database_rejects_duplicate_active_repair(self):
|
|
miner_id = self.create_miner()
|
|
service = RepairService(self.repository)
|
|
service.begin_repair(
|
|
BeginRepairRequest(miner_id=miner_id, reason="FAN"),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
with self.assertRaises(ActiveRepairConflict):
|
|
self.repository.create_repair(
|
|
miner_id,
|
|
"PSU",
|
|
self.actor_user_id,
|
|
)
|
|
self.assertEqual(self.count_repairs(miner_id), 1)
|
|
|
|
def test_repairs_are_isolated_by_miner(self):
|
|
first_miner = self.create_miner()
|
|
second_miner = self.create_miner()
|
|
service = RepairService(self.repository)
|
|
|
|
first = service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=first_miner,
|
|
reason="LOW_HASHRATE",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
second = service.begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=second_miner,
|
|
reason="NETWORKING",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertNotEqual(first.repair_id, second.repair_id)
|
|
self.assertEqual(
|
|
self.repository.get_active_repair(first_miner)["id"],
|
|
first.repair_id,
|
|
)
|
|
self.assertEqual(
|
|
self.repository.get_active_repair(second_miner)["id"],
|
|
second.repair_id,
|
|
)
|
|
|
|
def test_event_insertion_failure_rolls_back_repair_and_lifecycle(self):
|
|
miner_id = self.create_miner()
|
|
with self.repository._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
CREATE FUNCTION test_fail_repair_event()
|
|
RETURNS trigger AS $$
|
|
BEGIN
|
|
RAISE EXCEPTION 'forced repair event failure';
|
|
END;
|
|
$$ LANGUAGE plpgsql
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"""
|
|
CREATE TRIGGER test_repair_event_failure
|
|
BEFORE INSERT ON repair_events
|
|
FOR EACH ROW EXECUTE FUNCTION test_fail_repair_event()
|
|
"""
|
|
)
|
|
|
|
with self.assertRaises(RaiseException):
|
|
RepairService(self.repository).begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertEqual(self.count_repairs(miner_id), 0)
|
|
self.assertEqual(self.lifecycle(miner_id), "active")
|
|
|
|
def test_lifecycle_failure_rolls_back_repair_and_events(self):
|
|
miner_id = self.create_miner()
|
|
with self.repository._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
CREATE FUNCTION test_fail_repair_lifecycle()
|
|
RETURNS trigger AS $$
|
|
BEGIN
|
|
IF NEW.lifecycle_status = 'in_repair' THEN
|
|
RAISE EXCEPTION 'forced lifecycle failure';
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"""
|
|
CREATE TRIGGER test_repair_lifecycle_failure
|
|
BEFORE UPDATE ON miners
|
|
FOR EACH ROW
|
|
EXECUTE FUNCTION test_fail_repair_lifecycle()
|
|
"""
|
|
)
|
|
|
|
with self.assertRaises(RaiseException):
|
|
RepairService(self.repository).begin_repair(
|
|
BeginRepairRequest(
|
|
miner_id=miner_id,
|
|
reason="OTHER",
|
|
notes="Must roll back.",
|
|
),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
self.assertEqual(self.count_repairs(miner_id), 0)
|
|
self.assertEqual(self.lifecycle(miner_id), "active")
|
|
|
|
def test_simultaneous_requests_create_only_one_active_repair(self):
|
|
miner_id = self.create_miner()
|
|
barrier = Barrier(2)
|
|
|
|
def begin(reason):
|
|
barrier.wait()
|
|
return RepairService(RepairRepository()).begin_repair(
|
|
BeginRepairRequest(miner_id=miner_id, reason=reason),
|
|
actor_user_id=self.actor_user_id,
|
|
)
|
|
|
|
outcomes = []
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
futures = [
|
|
executor.submit(begin, "PSU"),
|
|
executor.submit(begin, "FAN"),
|
|
]
|
|
for future in futures:
|
|
try:
|
|
outcomes.append(future.result(timeout=10))
|
|
except ActiveRepairConflict as exc:
|
|
outcomes.append(exc)
|
|
|
|
self.assertEqual(
|
|
sum(not isinstance(item, Exception) for item in outcomes),
|
|
1,
|
|
)
|
|
self.assertEqual(
|
|
sum(isinstance(item, ActiveRepairConflict) for item in outcomes),
|
|
1,
|
|
)
|
|
self.assertEqual(self.count_repairs(miner_id), 1)
|
|
self.assertEqual(self.lifecycle(miner_id), "in_repair")
|
|
|
|
def test_repairs_create_permission_matrix_and_idempotent_seed(self):
|
|
with self.repository._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO permissions (key, description)
|
|
VALUES (
|
|
'repairs.create',
|
|
'Begin repairs for miner assets'
|
|
)
|
|
ON CONFLICT (key) DO UPDATE
|
|
SET description = EXCLUDED.description
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO role_permissions (role_id, permission_id)
|
|
SELECT roles.id, permissions.id
|
|
FROM roles
|
|
JOIN permissions
|
|
ON permissions.key = 'repairs.create'
|
|
WHERE roles.name IN (
|
|
'Admin', 'Manager', 'Technician'
|
|
)
|
|
ON CONFLICT DO NOTHING
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"""
|
|
SELECT roles.name
|
|
FROM role_permissions
|
|
JOIN roles ON roles.id = role_permissions.role_id
|
|
JOIN permissions
|
|
ON permissions.id = role_permissions.permission_id
|
|
WHERE permissions.key = 'repairs.create'
|
|
ORDER BY roles.name
|
|
"""
|
|
)
|
|
role_names = [row[0] for row in cur.fetchall()]
|
|
|
|
self.assertEqual(
|
|
role_names,
|
|
["Admin", "Manager", "Technician"],
|
|
)
|
|
|
|
|
|
class RepairArchitectureTests(unittest.TestCase):
|
|
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")
|
|
)
|
|
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)
|
|
self.assertFalse(
|
|
any(
|
|
module == "fastapi"
|
|
or module.startswith("fastapi.")
|
|
for module in imported_modules
|
|
),
|
|
filename,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|