import ast import unittest from copy import deepcopy from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from uuid import UUID, uuid4 from psycopg.errors import ForeignKeyViolation, UniqueViolation from app.inventory import ( DuplicateAssetID, InventoryCreateRequest, InventoryMetadataUpdateRequest, InventoryRepository, InventorySearchFilters, InventoryService, InventoryUpdateRequest, InventoryValidationError, ) def inventory_row(**overrides): now = datetime.now(timezone.utc) row = { "id": uuid4(), "site_id": None, "room_id": None, "asset_tag": "ASSET-001", "current_mac": None, "current_ip": None, "hostname": None, "miner_sn": None, "control_board_sn": None, "power_sn": None, "model": None, "firmware_version": None, "lifecycle_status": "active", "status": "unknown", "notes": None, "first_seen_at": None, "last_seen_at": None, "created_at": now, "updated_at": now, "tags": [], } row.update(overrides) return row class FakeInventoryRepository: def __init__(self): self.rows = {} self.created_values = None self.asset_id_history = [] self.update_error = None @contextmanager def transaction(self): rows_snapshot = deepcopy(self.rows) history_snapshot = deepcopy(self.asset_id_history) try: yield self except Exception: self.rows = rows_snapshot self.asset_id_history = history_snapshot raise def get_miner_by_id(self, miner_id): return self.rows.get(miner_id) def get_miner_by_id_for_update(self, miner_id): return self.get_miner_by_id(miner_id) def get_miner_by_asset_id(self, asset_id): key = asset_id.strip().casefold() return next( ( row for row in self.rows.values() if str(row.get("asset_tag") or "").strip().casefold() == key ), None, ) def list_miners(self, *, limit=100, offset=0): return list(self.rows.values())[offset : offset + limit] def search_miners(self, **kwargs): rows = list(self.rows.values()) lifecycle = kwargs.get("lifecycle_status") if lifecycle: rows = [ row for row in rows if row["lifecycle_status"] == lifecycle ] return rows[kwargs.get("offset", 0) : kwargs.get("limit", 100)] def create_miner(self, values): self.created_values = dict(values) row = inventory_row(**values) self.rows[row["id"]] = row return row def update_miner(self, miner_id, values): if self.update_error is not None: raise self.update_error row = self.rows.get(miner_id) if not row: return None row.update(values) row["updated_at"] = datetime.now(timezone.utc) return row def add_asset_id_history( self, miner_id, old_asset_id, new_asset_id, changed_by ): history = { "id": uuid4(), "miner_id": miner_id, "old_asset_id": old_asset_id, "new_asset_id": new_asset_id, "changed_at": datetime.now(timezone.utc), "changed_by": changed_by, } self.asset_id_history.append(history) return history def list_asset_id_history(self, miner_id): return [ row for row in self.asset_id_history if row["miner_id"] == miner_id ] def add_tag(self, miner_id, tag_name): now = datetime.now(timezone.utc) tag = { "id": uuid4(), "name": tag_name, "created_at": now, "updated_at": now, } existing = { item["name"].strip().casefold() for item in self.rows[miner_id]["tags"] } if tag_name.casefold() not in existing: self.rows[miner_id]["tags"].append(tag) return tag def remove_tag(self, miner_id, tag_id): before = len(self.rows[miner_id]["tags"]) self.rows[miner_id]["tags"] = [ tag for tag in self.rows[miner_id]["tags"] if tag["id"] != tag_id ] return len(self.rows[miner_id]["tags"]) != before class InventoryServiceTests(unittest.TestCase): def setUp(self): self.repository = FakeInventoryRepository() self.service = InventoryService(self.repository) self.actor_user_id = uuid4() def test_create_normalizes_asset_id_tags_and_never_seen(self): created = self.service.create_miner( InventoryCreateRequest( asset_id=" Mine-001 ", psu_serial=" PSU-9 ", tags=[" Priority ", "priority", "\tRepair\n"], ) ) self.assertEqual(created.asset_id, "Mine-001") self.assertEqual(created.psu_serial, "PSU-9") self.assertEqual([tag.name for tag in created.tags], ["Priority", "Repair"]) self.assertIsNone(created.first_seen_at) self.assertIsNone(created.last_seen_at) self.assertIsNone(self.repository.created_values["first_seen_at"]) self.assertIsNone(self.repository.created_values["last_seen_at"]) def test_duplicate_asset_id_is_case_and_whitespace_insensitive(self): row = inventory_row(asset_tag="Mine-001") self.repository.rows[row["id"]] = row with self.assertRaises(DuplicateAssetID): self.service.create_miner( InventoryCreateRequest(asset_id=" mine-001 ") ) def test_lifecycle_validation_is_owned_by_service(self): self.assertEqual( self.service.validate_lifecycle_status(" SPARE "), "spare", ) with self.assertRaises(InventoryValidationError): self.service.create_miner( InventoryCreateRequest( asset_id="ASSET-BAD", lifecycle_status="warehouse", ) ) def test_blank_asset_and_tags_are_rejected(self): with self.assertRaises(InventoryValidationError): self.service.normalize_asset_id(" ") with self.assertRaises(InventoryValidationError): self.service.normalize_tags(["valid", "\t\n"]) def test_metadata_update_preserves_unspecified_fields(self): row = inventory_row( asset_tag="ASSET-001", miner_sn="MINER-OLD", power_sn="PSU-OLD", model="M50", notes="Old notes", ) self.repository.rows[row["id"]] = row updated = self.service.update_metadata( row["id"], InventoryMetadataUpdateRequest( model="M60", notes=None, ), actor_user_id=self.actor_user_id, ) self.assertEqual(updated.id, row["id"]) self.assertEqual(updated.asset_id, "ASSET-001") self.assertEqual(updated.miner_sn, "MINER-OLD") self.assertEqual(updated.psu_serial, "PSU-OLD") self.assertEqual(updated.model, "M60") self.assertIsNone(updated.notes) self.assertEqual(self.repository.asset_id_history, []) def test_asset_id_change_preserves_uuid_and_creates_history(self): row = inventory_row(asset_tag="ASSET-OLD") self.repository.rows[row["id"]] = row updated = self.service.update_metadata( row["id"], InventoryMetadataUpdateRequest(asset_id=" ASSET-NEW "), actor_user_id=self.actor_user_id, ) self.assertEqual(updated.id, row["id"]) self.assertEqual(updated.asset_id, "ASSET-NEW") self.assertEqual(len(self.repository.asset_id_history), 1) history = self.repository.asset_id_history[0] self.assertEqual(history["miner_id"], row["id"]) self.assertEqual(history["old_asset_id"], "ASSET-OLD") self.assertEqual(history["new_asset_id"], "ASSET-NEW") self.assertEqual(history["changed_by"], self.actor_user_id) def test_unchanged_normalized_asset_id_creates_no_history(self): row = inventory_row(asset_tag="ASSET-001") original_updated_at = row["updated_at"] self.repository.rows[row["id"]] = row updated = self.service.update_metadata( row["id"], InventoryMetadataUpdateRequest(asset_id=" ASSET-001 "), actor_user_id=self.actor_user_id, ) self.assertEqual(updated.asset_id, "ASSET-001") self.assertEqual(updated.updated_at, original_updated_at) self.assertEqual(self.repository.asset_id_history, []) def test_duplicate_asset_id_leaves_miner_and_history_unchanged(self): row = inventory_row(asset_tag="ASSET-OLD") duplicate = inventory_row(asset_tag="ASSET-TAKEN") self.repository.rows[row["id"]] = row self.repository.rows[duplicate["id"]] = duplicate with self.assertRaises(DuplicateAssetID): self.service.update_metadata( row["id"], InventoryMetadataUpdateRequest(asset_id=" asset-taken "), actor_user_id=self.actor_user_id, ) self.assertEqual( self.repository.rows[row["id"]]["asset_tag"], "ASSET-OLD", ) self.assertEqual(self.repository.asset_id_history, []) def test_empty_or_invalid_metadata_update_does_not_write(self): row = inventory_row(asset_tag="ASSET-OLD", model="M50") self.repository.rows[row["id"]] = row for request in ( InventoryMetadataUpdateRequest(), InventoryMetadataUpdateRequest(asset_id=" ", model="M60"), ): with self.subTest(request=request): with self.assertRaises(InventoryValidationError): self.service.update_metadata( row["id"], request, actor_user_id=self.actor_user_id, ) self.assertEqual(self.repository.rows[row["id"]]["model"], "M50") self.assertEqual(self.repository.asset_id_history, []) def test_uniqueness_race_becomes_duplicate_asset_id(self): row = inventory_row(asset_tag="ASSET-OLD") self.repository.rows[row["id"]] = row self.repository.update_error = UniqueViolation("duplicate Asset ID") with self.assertRaises(DuplicateAssetID): self.service.update_metadata( row["id"], InventoryMetadataUpdateRequest(asset_id="ASSET-NEW"), actor_user_id=self.actor_user_id, ) self.assertEqual( self.repository.rows[row["id"]]["asset_tag"], "ASSET-OLD", ) self.assertEqual(self.repository.asset_id_history, []) def test_update_maps_product_names_without_repository_rules(self): row = inventory_row(asset_tag="ASSET-OLD", power_sn="PSU-OLD") self.repository.rows[row["id"]] = row updated = self.service.update_miner( row["id"], InventoryUpdateRequest( psu_serial=" PSU-NEW ", lifecycle_status="retired", ), ) self.assertEqual(updated.asset_id, "ASSET-OLD") self.assertEqual(updated.psu_serial, "PSU-NEW") self.assertEqual(updated.lifecycle_status, "retired") def test_legacy_update_rejects_asset_id_change_without_actor(self): row = inventory_row(asset_tag="ASSET-OLD") self.repository.rows[row["id"]] = row with self.assertRaisesRegex( InventoryValidationError, "actor-attributed metadata update", ): self.service.update_miner( row["id"], InventoryUpdateRequest(asset_id="ASSET-NEW"), ) self.assertEqual( self.repository.rows[row["id"]]["asset_tag"], "ASSET-OLD", ) self.assertEqual(self.repository.asset_id_history, []) def test_search_validation_precedes_repository(self): with self.assertRaises(InventoryValidationError): self.service.search_miners( InventorySearchFilters(lifecycle_status="broken") ) class InventoryRepositoryIntegrationTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.repository = InventoryRepository() def setUp(self): self.actor_user_id = UUID("00000000-0000-0000-0000-000000000951") with self.repository.transaction() as repository: with repository._connection.cursor() as cur: cur.execute("DELETE FROM miner_asset_id_history") cur.execute("DELETE FROM miner_tags") cur.execute("DELETE FROM tags") cur.execute("DELETE FROM miners") cur.execute( """ INSERT INTO users ( id, username, display_name, password_hash, role_id ) SELECT %s, %s, %s, %s, id FROM roles WHERE name = 'Admin' ON CONFLICT (id) DO NOTHING """, ( self.actor_user_id, "inventory-history-test", "Inventory History Test", "not-used", ), ) def test_repository_crud_search_and_tags(self): created = self.repository.create_miner( { "asset_tag": "REPO-001", "current_ip": "192.0.2.10", "model": "M50", "miner_sn": "MINER-001", "control_board_sn": "CB-001", "power_sn": "PSU-001", "lifecycle_status": "active", "status": "unknown", "first_seen_at": None, "last_seen_at": None, } ) miner_id = created["id"] self.assertIsInstance(miner_id, UUID) self.assertIsNone(created["first_seen_at"]) self.assertIsNone(created["last_seen_at"]) self.assertEqual( self.repository.get_miner_by_asset_id(" repo-001 ")["id"], miner_id, ) self.assertEqual(self.repository.get_miner_by_id(miner_id)["model"], "M50") self.assertEqual(len(self.repository.list_miners()), 1) self.assertEqual( self.repository.search_miners(query="CB-001")[0]["id"], miner_id, ) updated = self.repository.update_miner( miner_id, {"lifecycle_status": "spare", "notes": "Bench stock"}, ) self.assertEqual(updated["lifecycle_status"], "spare") self.assertEqual( self.repository.search_miners(lifecycle_status="spare")[0]["id"], miner_id, ) tag = self.repository.add_tag(miner_id, "Priority") repeated = self.repository.add_tag(miner_id, " priority ") self.assertEqual(tag["id"], repeated["id"]) self.assertEqual(len(self.repository.list_tags()), 1) self.assertEqual( self.repository.search_miners(tag_names=["priority"])[0]["id"], miner_id, ) self.assertEqual( len(self.repository.get_miner_by_id(miner_id)["tags"]), 1, ) self.assertTrue(self.repository.remove_tag(miner_id, tag["id"])) self.assertEqual( self.repository.get_miner_by_id(miner_id)["tags"], [], ) def test_repeated_asset_id_changes_preserve_chronological_history(self): service = InventoryService(self.repository) created = service.create_miner( InventoryCreateRequest(asset_id="ASSET-FIRST") ) second = service.update_metadata( created.id, InventoryMetadataUpdateRequest(asset_id="ASSET-SECOND"), actor_user_id=self.actor_user_id, ) third = service.update_metadata( created.id, InventoryMetadataUpdateRequest(asset_id="ASSET-THIRD"), actor_user_id=self.actor_user_id, ) self.assertEqual(created.id, second.id) self.assertEqual(second.id, third.id) history = self.repository.list_asset_id_history(created.id) self.assertEqual(len(history), 2) self.assertEqual( [ (row["old_asset_id"], row["new_asset_id"]) for row in history ], [ ("ASSET-FIRST", "ASSET-SECOND"), ("ASSET-SECOND", "ASSET-THIRD"), ], ) self.assertLessEqual( history[0]["changed_at"], history[1]["changed_at"], ) def test_history_failure_rolls_back_asset_id_update(self): service = InventoryService(self.repository) created = service.create_miner( InventoryCreateRequest(asset_id="ASSET-ROLLBACK-OLD") ) with self.assertRaises(ForeignKeyViolation): service.update_metadata( created.id, InventoryMetadataUpdateRequest( asset_id="ASSET-ROLLBACK-NEW" ), actor_user_id=uuid4(), ) persisted = self.repository.get_miner_by_id(created.id) self.assertEqual(persisted["asset_tag"], "ASSET-ROLLBACK-OLD") self.assertEqual( self.repository.list_asset_id_history(created.id), [], ) def test_transaction_rolls_back_repository_writes(self): with self.assertRaises(RuntimeError): with self.repository.transaction() as repository: repository.create_miner( { "asset_tag": "ROLLBACK-001", "status": "unknown", "first_seen_at": None, "last_seen_at": None, } ) raise RuntimeError("rollback") self.assertIsNone( self.repository.get_miner_by_asset_id("ROLLBACK-001") ) class InventoryArchitectureTests(unittest.TestCase): def test_inventory_domain_layers_have_no_http_dependency(self): inventory_path = Path(__file__).parents[1] / "app" / "inventory" domain_files = ( "dependencies.py", "exceptions.py", "models.py", "repository.py", "service.py", ) for filename in domain_files: source_path = inventory_path / filename tree = ast.parse(source_path.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.startswith("fastapi") for module in imported_modules), source_path, ) if __name__ == "__main__": unittest.main()