import ast import unittest from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from uuid import UUID, uuid4 from app.inventory import ( DuplicateAssetID, InventoryCreateRequest, 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 @contextmanager def transaction(self): yield self def get_miner_by_id(self, miner_id): return self.rows.get(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): 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_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) 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_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( asset_id=" ASSET-NEW ", psu_serial=" PSU-NEW ", lifecycle_status="retired", ), ) self.assertEqual(updated.asset_id, "ASSET-NEW") self.assertEqual(updated.psu_serial, "PSU-NEW") self.assertEqual(updated.lifecycle_status, "retired") 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): with self.repository.transaction() as repository: with repository._connection.cursor() as cur: cur.execute("DELETE FROM miner_tags") cur.execute("DELETE FROM tags") cur.execute("DELETE FROM miners") 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_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_package_has_no_http_layer_or_router(self): import app.inventory as inventory inventory_path = Path(__file__).parents[1] / "app" / "inventory" for source_path in inventory_path.glob("*.py"): 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, ) self.assertFalse(hasattr(inventory, "router")) if __name__ == "__main__": unittest.main()