import ast from concurrent.futures import ThreadPoolExecutor from threading import Barrier from unittest.mock import patch import unittest from contextlib import contextmanager from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from uuid import UUID, uuid4 from psycopg.errors import CheckViolation, RaiseException, UniqueViolation from psycopg.rows import dict_row from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from app.identity import get_current_user, required_api_permission from app.repairs.dependencies import get_repair_service from app.repairs.exceptions import ( ControlBoardReplacementConflict, InvalidMacAddress, MacOwnershipConflict, RepairMinerNotFound, RepairMinerNotInRepair, ) from app.repairs.models import ( BeginRepairRequest, ControlBoardReplacementRequest, ControlBoardReplacementResult, ) from app.repairs.repository import RepairRepository from app.repairs.router import control_board_router from app.repairs.service import RepairService NOW = datetime(2026, 8, 2, 12, 0, tzinfo=timezone.utc) def request(**overrides): values = { "new_mac_address": "02:AA:BB:CC:DD:02", "new_control_board_serial": " CB-NEW ", "board_model": " Model X ", "board_revision": " Rev B ", "firmware_version": " 1.2.3 ", "removed_board_disposition": "sent_for_repair", "replacement_reason": "failed", "notes": " Board would not boot. ", } values.update(overrides) return ControlBoardReplacementRequest(**values) class FakeReplacementRepository: def __init__(self): self.miner_id = uuid4() self.repair_id = uuid4() self.actor_id = uuid4() self.miner = { "id": self.miner_id, "asset_tag": "MINER-004", "lifecycle_status": "in_repair", "current_mac": "02:AA:BB:CC:DD:01", "control_board_sn": "CB-OLD", "firmware_version": "1.0.0", } self.repair = {"id": self.repair_id, "miner_id": self.miner_id} self.installations = [] self.events = [{"sequence": 1}] self.identities = [] self.calls = [] self.fail_at = None self.owner = None @contextmanager def transaction(self): self.calls.append("transaction") snapshot = deepcopy( (self.miner, self.installations, self.events, self.identities) ) try: yield self except Exception: self.miner, self.installations, self.events, self.identities = snapshot raise def _fail(self, name): self.calls.append(name) if self.fail_at == name: raise RuntimeError(name) def lock_miner_for_control_board_replacement(self, miner_id): self._fail("lock_miner") return self.miner if miner_id == self.miner_id else None def lock_active_repair_for_miner(self, miner_id): self._fail("lock_repair") return self.repair if miner_id == self.miner_id else None def lock_current_control_board(self, miner_id): self._fail("lock_board") current = [row for row in self.installations if row["removed_at"] is None] return current[0] if current else None def create_legacy_control_board(self, miner): self._fail("create_legacy_board") row = { "id": uuid4(), "miner_id": miner["id"], "mac_address": miner["current_mac"], "board_serial": miner["control_board_sn"], "board_model": None, "board_revision": None, "removed_at": None, } self.installations.append(row) return row def ensure_network_identity_history(self, miner_id, **values): self._fail("ensure_old_identity") self.identities.append({"miner_id": miner_id, "is_current": False, **values}) def claim_mac_ownership(self, normalized_mac, miner_id, actor_user_id): self._fail("claim_mac") if self.owner is not None and self.owner != miner_id: raise MacOwnershipConflict(normalized_mac) self.owner = miner_id def close_control_board(self, installation_id, **values): self._fail("close_board") row = next(row for row in self.installations if row["id"] == installation_id) row.update(removed_at=NOW, **values) return row def create_control_board_installation(self, **values): self._fail("create_board") row = {"id": uuid4(), "removed_at": None, **values, "installed_at": NOW} self.installations.append(row) return row def set_current_network_identity(self, miner_id, **values): self._fail("set_identity") self.identities.append({"miner_id": miner_id, **values}) return self.identities[-1] def set_current_control_board(self, miner_id, *, mac_address, board_serial): self._fail("set_projection") self.miner.update(current_mac=mac_address, control_board_sn=board_serial) def get_next_event_sequence(self, repair_id): self._fail("next_sequence") return max(event["sequence"] for event in self.events) + 1 def create_event(self, repair_id, **values): self._fail("create_event") row = {"id": uuid4(), "repair_id": repair_id, "created_at": NOW, **values} self.events.append(row) return row class ControlBoardReplacementServiceTests(unittest.TestCase): def setUp(self): self.repository = FakeReplacementRepository() self.service = RepairService(self.repository) def replace(self, replacement=None): return self.service.replace_control_board( self.repository.miner_id, replacement or request(), actor_user_id=self.repository.actor_id, ) def test_replacement_is_atomic_and_preserves_durable_identity(self): miner_id = self.repository.miner["id"] asset_id = self.repository.miner["asset_tag"] result = self.replace() self.assertEqual(result.miner_id, miner_id) self.assertEqual(result.asset_id, asset_id) self.assertEqual(result.current_mac, "02:AA:BB:CC:DD:02") self.assertEqual(result.control_board_serial, "CB-NEW") self.assertEqual(self.repository.miner["id"], miner_id) self.assertEqual(self.repository.miner["asset_tag"], asset_id) self.assertEqual(len([x for x in self.repository.installations if x["removed_at"] is None]), 1) event = self.repository.events[-1] self.assertEqual(event["event_type"], "CONTROL_BOARD_REPLACED") self.assertEqual(event["sequence"], 2) self.assertEqual(event["created_by"], self.repository.actor_id) self.assertEqual(event["payload"]["old_mac_address"], "02:AA:BB:CC:DD:01") self.assertEqual(event["payload"]["old_control_board_serial"], "CB-OLD") self.assertEqual(event["payload"]["notes"], "Board would not boot.") def test_mac_normalization_and_prohibited_values(self): self.assertEqual( self.service.normalize_mac_address("02-aA-bb-cc-dd-02"), ("02:AA:BB:CC:DD:02", "02aabbccdd02"), ) for value in ( "bad", "00:00:00:00:00:00", "FF:FF:FF:FF:FF:FF", "01:00:5E:00:00:01", ): with self.subTest(value=value), self.assertRaises(InvalidMacAddress): self.service.normalize_mac_address(value) def test_conflicting_states_do_not_write(self): cases = ("active", "spare", "retired") for lifecycle in cases: with self.subTest(lifecycle=lifecycle): repository = FakeReplacementRepository() repository.miner["lifecycle_status"] = lifecycle with self.assertRaises(RepairMinerNotInRepair): RepairService(repository).replace_control_board( repository.miner_id, request(), actor_user_id=repository.actor_id, ) self.assertEqual(repository.events, [{"sequence": 1}]) def test_same_and_historically_owned_mac_conflict(self): with self.assertRaises(ControlBoardReplacementConflict): self.replace(request(new_mac_address="02:AA:BB:CC:DD:01")) repository = FakeReplacementRepository() repository.owner = uuid4() with self.assertRaises(MacOwnershipConflict): RepairService(repository).replace_control_board( repository.miner_id, request(), actor_user_id=repository.actor_id ) def test_missing_miner_and_active_repair_conflict(self): with self.assertRaises(RepairMinerNotFound): self.service.replace_control_board( uuid4(), request(), actor_user_id=self.repository.actor_id ) self.repository.repair = None with self.assertRaises(ControlBoardReplacementConflict): self.replace() def test_each_meaningful_write_failure_rolls_back(self): for step in ( "claim_mac", "close_board", "create_board", "set_identity", "set_projection", "create_event", ): with self.subTest(step=step): repository = FakeReplacementRepository() repository.fail_at = step original = deepcopy(repository.miner) with self.assertRaises(RuntimeError): RepairService(repository).replace_control_board( repository.miner_id, request(), actor_user_id=repository.actor_id, ) self.assertEqual(repository.miner, original) self.assertEqual(repository.installations, []) self.assertEqual(repository.events, [{"sequence": 1}]) class StubReplacementService: def __init__(self): self.error = None self.result = None self.request = None self.actor = None def replace_control_board(self, miner_id, request, *, actor_user_id): self.request = request self.actor = actor_user_id if self.error: raise self.error return self.result class ControlBoardReplacementRouterTests(unittest.TestCase): def setUp(self): self.service = StubReplacementService() self.app = FastAPI() self.app.include_router(control_board_router) self.app.dependency_overrides[get_repair_service] = lambda: self.service self.client = TestClient(self.app) self.actor = uuid4() self.miner = uuid4() def tearDown(self): self.client.close() def authenticate(self, permissions=("miners.replace_control_board",), role="Technician"): self.app.dependency_overrides[get_current_user] = lambda: { "id": str(self.actor), "role": role, "permissions": list(permissions), } def post(self, body=None): return self.client.post( f"/api/miners/{self.miner}/replace-control-board", json=(request().model_dump(mode="json") if body is None else body), ) def test_authentication_and_permission_are_required(self): self.assertEqual(self.post().status_code, 401) self.authenticate(permissions=("miners.update",)) self.assertEqual(self.post().status_code, 403) def test_success_uses_typed_envelope_and_authenticated_actor(self): self.authenticate() self.service.result = ControlBoardReplacementResult( miner_id=self.miner, asset_id="MINER-004", repair_id=uuid4(), installation_id=uuid4(), event_id=uuid4(), event_sequence=4, current_mac="02:AA:BB:CC:DD:02", control_board_serial="CB-NEW", replaced_at=NOW, replaced_by=self.actor, ) response = self.post() self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["miner_id"], str(self.miner)) self.assertEqual(response.json()["meta"], {}) self.assertIsNone(response.json()["error"]) self.assertEqual(self.service.actor, self.actor) def test_validation_and_domain_error_mapping(self): self.authenticate() self.assertEqual(self.post({"new_mac_address": "bad"}).status_code, 422) self.service.error = RepairMinerNotFound(self.miner) self.assertEqual(self.post().status_code, 404) self.service.error = ControlBoardReplacementConflict("conflict") self.assertEqual(self.post().status_code, 409) self.service.error = InvalidMacAddress() self.assertEqual(self.post().status_code, 422) def test_route_and_permission_mapping_are_exact(self): routes = [ (route.path, route.methods) for route in control_board_router.routes if hasattr(route, "methods") ] self.assertEqual(routes, [ ("/api/miners/{miner_id}/replace-control-board", {"POST"}) ]) path = f"/api/miners/{self.miner}/replace-control-board" self.assertEqual( required_api_permission("POST", path), "miners.replace_control_board", ) def test_allowed_roles_and_read_only_matrix(self): self.service.result = ControlBoardReplacementResult( miner_id=self.miner, asset_id="MINER-004", repair_id=uuid4(), installation_id=uuid4(), event_id=uuid4(), event_sequence=2, current_mac="02:AA:BB:CC:DD:02", control_board_serial=None, replaced_at=NOW, replaced_by=self.actor, ) for role in ("Admin", "Manager", "Technician"): with self.subTest(role=role): self.authenticate(role=role) self.assertEqual(self.post().status_code, 200) self.authenticate(permissions=(), role="Read Only") self.assertEqual(self.post().status_code, 403) def test_request_contract_rejects_invalid_shapes(self): self.authenticate() valid = request().model_dump(mode="json") invalid_bodies = [] missing_serial = dict(valid) missing_serial.pop("new_control_board_serial") invalid_bodies.append(missing_serial) for field, value in ( ("new_mac_address", ""), ("notes", "x" * 4001), ("replacement_reason", "invented"), ("removed_board_disposition", "invented"), ): body = dict(valid) body[field] = value invalid_bodies.append(body) unknown = dict(valid) unknown["created_by"] = str(self.actor) invalid_bodies.append(unknown) for body in invalid_bodies: with self.subTest(body=body): self.assertEqual(self.post(body).status_code, 422) def test_router_has_no_sql_driver_or_transaction_management(self): path = Path(__file__).parents[1] / "app" / "repairs" / "router.py" source = path.read_text(encoding="utf-8") tree = ast.parse(source) 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(name.startswith(("psycopg", "sqlalchemy", "asyncpg")) for name in modules)) for keyword in ("SELECT ", "INSERT ", "UPDATE ", "DELETE ", ".COMMIT(", ".ROLLBACK("): self.assertNotIn(keyword, source.upper()) class LegacyChangeMacCompatibilityTests(unittest.TestCase): def setUp(self): from app import main as main_module self.main = main_module self.service = StubReplacementService() self.actor = uuid4() self.miner = uuid4() self.service.result = ControlBoardReplacementResult( miner_id=self.miner, asset_id="MINER-004", repair_id=uuid4(), installation_id=uuid4(), event_id=uuid4(), event_sequence=3, current_mac="02:AA:BB:CC:DD:02", control_board_serial="CB-NEW", replaced_at=NOW, replaced_by=self.actor, ) def legacy_request(self, **overrides): values = { "new_mac": "02:AA:BB:CC:DD:02", "component_serial": "CB-NEW", "description": "Replaced during repair.", "replacement_reason": "failed", "removed_board_disposition": "sent_for_repair", } values.update(overrides) return self.main.MinerMacChangeRequest(**values) def call(self, legacy_request): with patch.object(self.main, "RepairRepository"), patch.object( self.main, "RepairService", return_value=self.service ): return self.main.change_miner_mac_address( str(self.miner), legacy_request, {"id": str(self.actor)} ) def test_alias_uses_same_permission_and_domain_workflow(self): path = f"/api/miners/{self.miner}/change-mac" self.assertEqual( required_api_permission("POST", path), "miners.replace_control_board", ) response = self.call(self.legacy_request()) self.assertEqual(self.service.actor, self.actor) self.assertEqual(self.service.request.replacement_reason, "failed") self.assertEqual( self.service.request.removed_board_disposition, "sent_for_repair" ) self.assertEqual( response["data"]["miner"]["current_mac"], "02:AA:BB:CC:DD:02", ) self.assertIn("control_board_replacement", response["data"]) self.assertNotIn("maintenance_event", response["data"]) def test_alias_does_not_fabricate_required_workflow_fields(self): for changes in ( {"replacement_reason": None}, {"removed_board_disposition": None}, {"new_ip": "192.0.2.4"}, {"downtime_minutes": 10}, ): with self.subTest(changes=changes), self.assertRaises(HTTPException) as caught: self.call(self.legacy_request(**changes)) self.assertEqual(caught.exception.status_code, 422) def test_alias_maps_domain_conflicts_without_internal_error(self): self.service.error = ControlBoardReplacementConflict("Miner has no active repair") with self.assertRaises(HTTPException) as caught: self.call(self.legacy_request()) self.assertEqual(caught.exception.status_code, 409) class ControlBoardReplacementDatabaseTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.repository = RepairRepository() def setUp(self): self.actor = uuid4() with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute("DROP TRIGGER IF EXISTS test_board_event_failure ON repair_events") cur.execute("DROP FUNCTION IF EXISTS test_fail_board_event()") cur.execute( "TRUNCATE miner_control_board_installations, " "miner_network_identities, miner_mac_ownerships" ) cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history") cur.execute("DELETE FROM repair_events") cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history") 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 'board-test-%'") cur.execute( """ INSERT INTO users ( id, username, display_name, password_hash, role_id ) SELECT %s, %s, 'Board Test Actor', 'unused', id FROM roles WHERE name = 'Admin' """, (self.actor, f"board-test-{self.actor}"), ) def tearDown(self): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute("DROP TRIGGER IF EXISTS test_board_event_failure ON repair_events") cur.execute("DROP FUNCTION IF EXISTS test_fail_board_event()") cur.execute( "TRUNCATE miner_control_board_installations, " "miner_network_identities, miner_mac_ownerships" ) cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history") cur.execute("DELETE FROM repair_events") cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history") 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 'board-test-%'") def create_repairing_miner(self, mac_address, serial): asset_id = f"BOARD-{uuid4()}" 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, current_mac, control_board_sn, status, lifecycle_status, first_seen_at, last_seen_at ) VALUES (%s, %s, %s, 'unknown', 'active', NULL, NULL) RETURNING id """, (asset_id, mac_address, serial), ) miner_id = cur.fetchone()["id"] repair = RepairService(self.repository).begin_repair( BeginRepairRequest(miner_id=miner_id, reason="CONTROL_BOARD"), actor_user_id=self.actor, ) return miner_id, asset_id, repair.repair_id def replace(self, miner_id, mac_address="02:AA:BB:CC:EE:02"): return RepairService(self.repository).replace_control_board( miner_id, request(new_mac_address=mac_address), actor_user_id=self.actor, ) def fetchall(self, sql, values=()): with self.repository._connection_scope() as connection: with connection.cursor(row_factory=dict_row) as cur: cur.execute(sql, values) return [dict(row) for row in cur.fetchall()] def test_database_success_preserves_identity_and_authoritative_history(self): miner_id, asset_id, repair_id = self.create_repairing_miner( "02:AA:BB:CC:EE:01", "CB-OLD" ) result = self.replace(miner_id) miner = self.fetchall( "SELECT * FROM miners WHERE id = %s", (miner_id,) )[0] boards = self.fetchall( "SELECT * FROM miner_control_board_installations " "WHERE miner_id = %s ORDER BY created_at, id", (miner_id,) ) identities = self.fetchall( "SELECT * FROM miner_network_identities " "WHERE miner_id = %s ORDER BY is_current, mac_address", (miner_id,) ) events = self.fetchall( "SELECT * FROM repair_events WHERE repair_id = %s " "AND event_type = 'CONTROL_BOARD_REPLACED'", (repair_id,) ) self.assertEqual(result.miner_id, miner_id) self.assertEqual(miner["asset_tag"], asset_id) self.assertEqual(miner["current_mac"], "02:AA:BB:CC:EE:02") self.assertEqual(len(boards), 2) self.assertEqual(sum(row["removed_at"] is None for row in boards), 1) legacy = next(row for row in boards if row["record_origin"] == "legacy_projection") current = next(row for row in boards if row["removed_at"] is None) self.assertIsNone(legacy["installed_at"]) self.assertIsNone(legacy["installed_by"]) self.assertEqual(current["record_origin"], "replacement") self.assertEqual(current["installed_by"], self.actor) self.assertEqual(current["normalized_mac"], "02aabbccee02") self.assertEqual(len(identities), 2) self.assertEqual(sum(row["is_current"] for row in identities), 1) self.assertEqual( next(row for row in identities if row["is_current"])["mac_address"], "02:AA:BB:CC:EE:02", ) self.assertTrue(any(row["mac_address"] == "02:AA:BB:CC:EE:01" for row in identities)) self.assertEqual(len(events), 1) self.assertEqual(events[0]["created_by"], self.actor) self.assertEqual(events[0]["sequence"], result.event_sequence) def test_database_allows_omitted_optional_notes(self): miner_id, _, repair_id = self.create_repairing_miner( "02:AA:BB:CC:EA:01", "CB-OLD" ) RepairService(self.repository).replace_control_board( miner_id, request(new_mac_address="02:AA:BB:CC:EA:02", notes=None), actor_user_id=self.actor, ) event = self.fetchall( "SELECT payload FROM repair_events WHERE repair_id = %s " "AND event_type = 'CONTROL_BOARD_REPLACED'", (repair_id,), )[0] self.assertIn("notes", event["payload"]) self.assertIsNone(event["payload"]["notes"]) def test_database_constraints_prevent_drift_and_history_mutation(self): miner_id, _, repair_id = self.create_repairing_miner( "02:AA:BB:CC:ED:01", "CB-OLD" ) self.replace(miner_id, "02:AA:BB:CC:ED:02") with self.assertRaises(UniqueViolation): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( """ INSERT INTO miner_control_board_installations ( miner_id, record_origin, board_serial ) VALUES (%s, 'legacy_projection', 'SECOND') """, (miner_id,), ) with self.assertRaises(CheckViolation): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( "UPDATE miners SET control_board_sn = 'DRIFT' " "WHERE id = %s", (miner_id,), ) with self.assertRaises(CheckViolation): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( "UPDATE miner_network_identities SET is_current = false " "WHERE miner_id = %s AND is_current", (miner_id,), ) with self.assertRaises(RaiseException): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( "DELETE FROM miner_network_identities " "WHERE miner_id = %s", (miner_id,), ) with self.assertRaises(RaiseException): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( "DELETE FROM miner_control_board_installations " "WHERE miner_id = %s", (miner_id,), ) for statement in ( "UPDATE miner_mac_ownerships SET miner_id = gen_random_uuid() " "WHERE normalized_mac = '02aabbcced02'", "DELETE FROM miner_mac_ownerships " "WHERE normalized_mac = '02aabbcced02'", ): with self.assertRaises(RaiseException): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute(statement) with self.assertRaises(UniqueViolation): 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_by, payload ) SELECT repair_id, sequence, event_type, created_by, payload FROM repair_events WHERE repair_id = %s ORDER BY sequence LIMIT 1 """, (repair_id,), ) def test_database_rejects_prohibited_component_mac(self): miner_id, _, _ = self.create_repairing_miner( "02:AA:BB:CC:EC:01", "CB-OLD" ) for mac_address, normalized_mac in ( ("01:00:5E:00:00:01", "01005e000001"), ("00:00:00:00:00:00", "000000000000"), ("FF:FF:FF:FF:FF:FF", "ffffffffffff"), ("02-AA-BB-CC-EC-02", "02aabbccec02"), ): with self.subTest(mac_address=mac_address), self.assertRaises(CheckViolation): with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( """ INSERT INTO miner_control_board_installations ( miner_id, record_origin, mac_address, normalized_mac ) VALUES (%s, 'legacy_projection', %s, %s) """, (miner_id, mac_address, normalized_mac), ) def test_event_failure_rolls_back_every_replacement_write(self): miner_id, _, repair_id = self.create_repairing_miner( "02:AA:BB:CC:EF:01", "CB-OLD" ) with self.repository._connection_scope(write=True) as connection: with connection.cursor() as cur: cur.execute( """ CREATE FUNCTION test_fail_board_event() RETURNS trigger AS $$ BEGIN IF NEW.event_type = 'CONTROL_BOARD_REPLACED' THEN RAISE EXCEPTION 'forced replacement event failure'; END IF; RETURN NEW; END $$ LANGUAGE plpgsql """ ) cur.execute( "CREATE TRIGGER test_board_event_failure BEFORE INSERT ON " "repair_events FOR EACH ROW EXECUTE FUNCTION test_fail_board_event()" ) with self.assertRaises(RaiseException): self.replace(miner_id, "02:AA:BB:CC:EF:02") miner = self.fetchall("SELECT * FROM miners WHERE id = %s", (miner_id,))[0] self.assertEqual(miner["current_mac"], "02:AA:BB:CC:EF:01") self.assertEqual(self.fetchall( "SELECT * FROM miner_control_board_installations WHERE miner_id = %s", (miner_id,), ), []) self.assertEqual(self.fetchall( "SELECT * FROM miner_network_identities WHERE miner_id = %s", (miner_id,), ), []) self.assertEqual(len(self.fetchall( "SELECT * FROM repair_events WHERE repair_id = %s", (repair_id,) )), 1) owners = self.fetchall( "SELECT normalized_mac FROM miner_mac_ownerships WHERE miner_id = %s", (miner_id,), ) self.assertEqual(owners, [{"normalized_mac": "02aabbccef01"}]) def test_concurrent_replacements_on_one_miner_succeed_once(self): miner_id, _, repair_id = self.create_repairing_miner( "02:AA:BB:CC:F0:01", "CB-OLD" ) barrier = Barrier(2) def replace(mac_address): barrier.wait() return self.replace(miner_id, mac_address) outcomes = [] with ThreadPoolExecutor(max_workers=2) as executor: futures = [ executor.submit(replace, "02:AA:BB:CC:F0:02"), executor.submit(replace, "02:AA:BB:CC:F0:04"), ] for future in futures: try: outcomes.append(future.result(timeout=15)) except ControlBoardReplacementConflict as exc: outcomes.append(exc) self.assertEqual(sum(not isinstance(item, Exception) for item in outcomes), 1) self.assertEqual(sum(isinstance(item, Exception) for item in outcomes), 1) self.assertEqual(len(self.fetchall( "SELECT * FROM miner_control_board_installations " "WHERE miner_id = %s AND removed_at IS NULL", (miner_id,) )), 1) self.assertEqual(len(self.fetchall( "SELECT * FROM repair_events WHERE repair_id = %s " "AND event_type = 'CONTROL_BOARD_REPLACED'", (repair_id,) )), 1) def test_two_miners_cannot_concurrently_claim_one_mac(self): first, _, _ = self.create_repairing_miner("02:AA:BB:CC:F1:01", "CB-1") second, _, _ = self.create_repairing_miner("02:AA:BB:CC:F1:02", "CB-2") barrier = Barrier(2) def replace(miner_id): barrier.wait() return self.replace(miner_id, "02:AA:BB:CC:F1:04") outcomes = [] with ThreadPoolExecutor(max_workers=2) as executor: futures = [executor.submit(replace, first), executor.submit(replace, second)] for future in futures: try: outcomes.append(future.result(timeout=15)) except ControlBoardReplacementConflict as exc: outcomes.append(exc) self.assertEqual(sum(not isinstance(item, Exception) for item in outcomes), 1) self.assertEqual(sum(isinstance(item, Exception) for item in outcomes), 1) owner = self.fetchall( "SELECT miner_id FROM miner_mac_ownerships " "WHERE normalized_mac = '02aabbccf104'" ) self.assertEqual(len(owner), 1) if __name__ == "__main__": unittest.main()