import ast import unittest from datetime import datetime, timezone from pathlib import Path from uuid import UUID, uuid4 from fastapi import FastAPI from fastapi.testclient import TestClient from app.identity import get_current_user from app.inventory.dependencies import get_inventory_service from app.inventory.exceptions import ( DuplicateAssetID, InventoryValidationError, MinerNotFound, ) from app.inventory.models import ( InventoryAssetIDHistoryEntry, InventoryCreateRequest, InventoryMetadataUpdateRequest, InventoryMinerDetail, InventoryMinerSummary, InventorySearchFilters, ) from app.inventory.router import router class StubInventoryService: def __init__(self): self.search_result: list[InventoryMinerSummary] = [] self.detail_result: InventoryMinerDetail | None = None self.search_filters: InventorySearchFilters | None = None self.detail_miner_id: UUID | None = None self.search_error: InventoryValidationError | None = None self.create_result: InventoryMinerDetail | None = None self.create_error: Exception | None = None self.create_request: InventoryCreateRequest | None = None self.update_result: InventoryMinerDetail | None = None self.update_error: Exception | None = None self.update_miner_id: UUID | None = None self.update_request: InventoryMetadataUpdateRequest | None = None self.update_actor_user_id: UUID | None = None self.history_result: list[InventoryAssetIDHistoryEntry] = [] self.history_error: Exception | None = None self.history_miner_id: UUID | None = None self.include_location_details = False self.location_options = {"sites": [], "rooms": []} def search_miners( self, filters: InventorySearchFilters, *, include_location_details: bool = False, ) -> list[InventoryMinerSummary]: self.search_filters = filters self.include_location_details = include_location_details if self.search_error is not None: raise self.search_error return self.search_result def list_location_filter_options(self): return self.location_options def create_miner( self, request: InventoryCreateRequest, ) -> InventoryMinerDetail: self.create_request = request if self.create_error is not None: raise self.create_error if self.create_result is None: raise AssertionError("create_result was not configured") return self.create_result def update_metadata( self, miner_id: UUID, request: InventoryMetadataUpdateRequest, *, actor_user_id: UUID, ) -> InventoryMinerDetail: self.update_miner_id = miner_id self.update_request = request self.update_actor_user_id = actor_user_id if self.update_error is not None: raise self.update_error if self.update_result is None: raise AssertionError("update_result was not configured") return self.update_result def get_miner(self, miner_id: UUID) -> InventoryMinerDetail: self.detail_miner_id = miner_id if self.detail_result is None or self.detail_result.id != miner_id: raise MinerNotFound(miner_id) return self.detail_result def get_asset_id_history( self, miner_id: UUID, ) -> list[InventoryAssetIDHistoryEntry]: self.history_miner_id = miner_id if self.history_error is not None: raise self.history_error return self.history_result def miner_summary(miner_id: UUID | None = None) -> InventoryMinerSummary: now = datetime.now(timezone.utc) return InventoryMinerSummary( id=miner_id or uuid4(), asset_id="ASSET-001", model="M50", lifecycle_status="active", runtime_status="unknown", created_at=now, updated_at=now, ) def miner_detail(miner_id: UUID | None = None) -> InventoryMinerDetail: values = miner_summary(miner_id).model_dump() values["miner_sn"] = "MINER-001" return InventoryMinerDetail( **values, control_board_sn="CB-001", psu_serial="PSU-001", ) def history_entry( *, old_asset_id: str | None, new_asset_id: str, changed_at: datetime, ) -> InventoryAssetIDHistoryEntry: return InventoryAssetIDHistoryEntry( old_asset_id=old_asset_id, new_asset_id=new_asset_id, changed_at=changed_at, changed_by=UUID("00000000-0000-0000-0000-000000000951"), changed_by_display_name="Inventory Test User", ) class InventoryRouterTests(unittest.TestCase): def setUp(self): self.service = StubInventoryService() self.app = FastAPI() self.app.include_router(router) self.app.dependency_overrides[get_inventory_service] = ( lambda: self.service ) self.client = TestClient(self.app) def tearDown(self): self.client.close() def authenticate(self, permissions=("miners.view",)): self.app.dependency_overrides[get_current_user] = lambda: { "id": str(uuid4()), "username": "reader", "role": "Read Only", "permissions": list(permissions), } def test_create_requires_authentication(self): response = self.client.post( "/api/inventory/miners", json={"asset_id": "ASSET-NEW"}, ) self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["detail"], "Login required") def test_create_requires_miners_update_permission(self): self.authenticate(permissions=("miners.view",)) response = self.client.post( "/api/inventory/miners", json={"asset_id": "ASSET-NEW"}, ) self.assertEqual(response.status_code, 403) self.assertIsNone(self.service.create_request) def test_create_returns_new_miner_in_typed_envelope(self): self.authenticate(permissions=("miners.update",)) created = miner_detail() self.service.create_result = created response = self.client.post( "/api/inventory/miners", json={ "asset_id": " ASSET-NEW ", "lifecycle_status": "spare", "site_id": str(uuid4()), "tags": ["Priority", "Repair"], }, ) self.assertEqual(response.status_code, 201) self.assertEqual(response.json()["data"]["id"], str(created.id)) self.assertEqual(response.json()["meta"], {}) self.assertIsNone(response.json()["error"]) request = self.service.create_request self.assertIsNotNone(request) self.assertEqual(request.asset_id, " ASSET-NEW ") self.assertEqual(request.lifecycle_status, "spare") self.assertEqual(request.tags, ["Priority", "Repair"]) def test_duplicate_asset_id_returns_conflict(self): self.authenticate(permissions=("miners.update",)) self.service.create_error = DuplicateAssetID("ASSET-001") response = self.client.post( "/api/inventory/miners", json={"asset_id": "ASSET-001"}, ) self.assertEqual(response.status_code, 409) self.assertEqual( response.json()["detail"], "Asset ID already exists: ASSET-001", ) def test_invalid_create_body_returns_validation_error(self): self.authenticate(permissions=("miners.update",)) for body in ( {}, {"asset_id": "A" * 121}, {"asset_id": "ASSET-001", "site_id": "not-a-uuid"}, ): with self.subTest(body=body): response = self.client.post( "/api/inventory/miners", json=body, ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.create_request) def test_create_maps_domain_validation_error(self): self.authenticate(permissions=("miners.update",)) self.service.create_error = InventoryValidationError( "Tag names cannot be blank" ) response = self.client.post( "/api/inventory/miners", json={"asset_id": "ASSET-001", "tags": [" "]}, ) self.assertEqual(response.status_code, 422) self.assertEqual( response.json()["detail"], "Tag names cannot be blank", ) def test_patch_requires_authentication(self): response = self.client.patch( f"/api/inventory/miners/{uuid4()}", json={"model": "M60"}, ) self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["detail"], "Login required") def test_patch_requires_miners_update_permission(self): self.authenticate(permissions=("miners.view",)) response = self.client.patch( f"/api/inventory/miners/{uuid4()}", json={"model": "M60"}, ) self.assertEqual(response.status_code, 403) self.assertIsNone(self.service.update_request) def test_patch_unknown_miner_returns_not_found(self): self.authenticate(permissions=("miners.update",)) miner_id = uuid4() self.service.update_error = MinerNotFound(miner_id) response = self.client.patch( f"/api/inventory/miners/{miner_id}", json={"model": "M60"}, ) self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["detail"], "Miner not found") def test_patch_malformed_uuid_returns_validation_error(self): self.authenticate(permissions=("miners.update",)) response = self.client.patch( "/api/inventory/miners/not-a-uuid", json={"model": "M60"}, ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.update_request) def test_empty_patch_maps_domain_validation_error(self): self.authenticate(permissions=("miners.update",)) self.service.update_error = InventoryValidationError( "At least one metadata field must be provided" ) response = self.client.patch( f"/api/inventory/miners/{uuid4()}", json={}, ) self.assertEqual(response.status_code, 422) self.assertEqual( response.json()["detail"], "At least one metadata field must be provided", ) def test_patch_rejects_unknown_and_operational_fields(self): self.authenticate(permissions=("miners.update",)) for body in ( {"unknown_field": "value"}, {"lifecycle_status": "retired"}, {"current_ip": "192.0.2.10"}, {"current_mac": "AA:BB:CC:DD:EE:FF"}, {"control_board_sn": "CB-NEW"}, {"site_id": str(uuid4())}, {"room_id": str(uuid4())}, {"tags": ["Priority"]}, {"firmware_version": "1.2.3"}, {"hostname": "miner-host"}, ): with self.subTest(body=body): response = self.client.patch( f"/api/inventory/miners/{uuid4()}", json=body, ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.update_request) def test_valid_metadata_patch_returns_typed_envelope(self): self.authenticate(permissions=("miners.update",)) updated = miner_detail() self.service.update_result = updated response = self.client.patch( f"/api/inventory/miners/{updated.id}", json={ "asset_id": " ASSET-NEW ", "miner_sn": "MINER-NEW", "psu_serial": None, "model": "M60", "notes": "Bench verified", }, ) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["id"], str(updated.id)) self.assertEqual(body["meta"], {}) self.assertIsNone(body["error"]) self.assertEqual(self.service.update_miner_id, updated.id) self.assertIsInstance(self.service.update_actor_user_id, UUID) request = self.service.update_request self.assertIsNotNone(request) self.assertEqual( request.model_fields_set, {"asset_id", "miner_sn", "psu_serial", "model", "notes"}, ) self.assertIsNone(request.psu_serial) def test_patch_duplicate_asset_id_returns_conflict(self): self.authenticate(permissions=("miners.update",)) self.service.update_error = DuplicateAssetID("ASSET-TAKEN") response = self.client.patch( f"/api/inventory/miners/{uuid4()}", json={"asset_id": "ASSET-TAKEN"}, ) self.assertEqual(response.status_code, 409) self.assertEqual( response.json()["detail"], "Asset ID already exists: ASSET-TAKEN", ) def test_list_requires_authentication(self): response = self.client.get("/api/inventory/miners") self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["detail"], "Login required") def test_detail_requires_authentication(self): response = self.client.get(f"/api/inventory/miners/{uuid4()}") self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["detail"], "Login required") def test_asset_id_history_requires_authentication(self): response = self.client.get( f"/api/inventory/miners/{uuid4()}/asset-id-history" ) self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["detail"], "Login required") def test_list_requires_miners_view_permission(self): self.authenticate(permissions=()) response = self.client.get("/api/inventory/miners") self.assertEqual(response.status_code, 403) def test_asset_id_history_requires_miners_view_permission(self): self.authenticate(permissions=()) response = self.client.get( f"/api/inventory/miners/{uuid4()}/asset-id-history" ) self.assertEqual(response.status_code, 403) self.assertIsNone(self.service.history_miner_id) def test_authenticated_list_returns_typed_envelope(self): self.authenticate() self.service.search_result = [miner_summary()] response = self.client.get("/api/inventory/miners") self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(len(body["data"]), 1) self.assertEqual(body["data"][0]["asset_id"], "ASSET-001") self.assertEqual( body["meta"], {"limit": 100, "offset": 0, "returned": 1, "location_options": {"sites": [], "rooms": []}}, ) self.assertIsNone(body["error"]) def test_list_projection_and_location_permission_are_explicit(self): miner_id = uuid4() site_id = uuid4() room_id = uuid4() position_id = uuid4() self.service.search_result = [ InventoryMinerSummary( id=miner_id, asset_id="FLEET-001", model="M50", miner_sn="SN-001", lifecycle_status="active", runtime_status="online", site_id=site_id, site_name="North Site", room_id=room_id, room_name="Room A", location_position_id=position_id, rack_label="Rack 1", shelf_label="Shelf 2", position_index=3, observed_at=datetime.now(timezone.utc), hashrate_ths=120.0, power_w=3400.0, chip_temp_max_c=72.0, hash_stable=True, created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) ] self.service.location_options = { "sites": [{"id": site_id, "name": "North Site"}], "rooms": [{"id": room_id, "site_id": site_id, "name": "Room A"}], } self.authenticate(("miners.view", "locations.view")) response = self.client.get("/api/inventory/miners") data = response.json()["data"][0] self.assertEqual(response.status_code, 200) self.assertEqual(data["miner_sn"], "SN-001") self.assertEqual(data["location_position_id"], str(position_id)) self.assertEqual(data["hashrate_ths"], 120.0) self.assertTrue(self.service.include_location_details) self.assertEqual( response.json()["meta"]["location_options"]["sites"][0]["name"], "North Site", ) self.authenticate(("miners.view",)) self.service.search_result = [miner_summary(miner_id)] response = self.client.get("/api/inventory/miners") data = response.json()["data"][0] self.assertIsNone(data["site_id"]) self.assertIsNone(data["room_id"]) self.assertIsNone(data["location_position_id"]) self.assertEqual(response.json()["meta"]["location_options"], {"sites": [], "rooms": []}) self.assertFalse(self.service.include_location_details) def test_location_filters_require_locations_view(self): self.authenticate(("miners.view",)) response = self.client.get( "/api/inventory/miners", params={"site_id": str(uuid4())} ) self.assertEqual(response.status_code, 403) self.assertIsNone(self.service.search_filters) def test_empty_list_returns_successful_empty_collection(self): self.authenticate() response = self.client.get("/api/inventory/miners") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"], []) self.assertEqual(response.json()["meta"]["returned"], 0) def test_search_filters_are_passed_to_service(self): self.authenticate() response = self.client.get( "/api/inventory/miners", params=[ ("query", "M50"), ("asset_id", "ASSET-001"), ("lifecycle_status", "spare"), ("model", "M50"), ("runtime_status", "offline"), ("repair", "active"), ("operational_state", "repair"), ("problems", "true"), ("tags", "Priority"), ("tags", "Repair"), ("limit", "25"), ("offset", "50"), ], ) self.assertEqual(response.status_code, 200) filters = self.service.search_filters self.assertIsNotNone(filters) self.assertEqual(filters.query, "M50") self.assertEqual(filters.asset_id, "ASSET-001") self.assertEqual(filters.lifecycle_status, "spare") self.assertEqual(filters.model, "M50") self.assertEqual(filters.runtime_status, "offline") self.assertEqual(filters.repair, "active") self.assertEqual(filters.operational_state, "repair") self.assertTrue(filters.problems) self.assertEqual(filters.tags, ["Priority", "Repair"]) self.assertEqual(filters.limit, 25) self.assertEqual(filters.offset, 50) def test_invalid_lifecycle_filter_returns_client_error(self): self.authenticate() self.service.search_error = InventoryValidationError( "Lifecycle status must be active, spare, or retired" ) response = self.client.get( "/api/inventory/miners", params={"lifecycle_status": "broken"}, ) self.assertEqual(response.status_code, 422) self.assertIn("Lifecycle status", response.json()["detail"]) def test_pagination_is_validated_and_bounded(self): self.authenticate() for params in ( {"limit": 0}, {"limit": 501}, {"offset": -1}, ): with self.subTest(params=params): response = self.client.get( "/api/inventory/miners", params=params, ) self.assertEqual(response.status_code, 422) def test_existing_uuid_returns_inventory_detail(self): self.authenticate() expected = miner_detail() self.service.detail_result = expected response = self.client.get( f"/api/inventory/miners/{expected.id}" ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["id"], str(expected.id)) self.assertEqual(response.json()["data"]["psu_serial"], "PSU-001") self.assertEqual(self.service.detail_miner_id, expected.id) def test_unknown_uuid_returns_not_found(self): self.authenticate() response = self.client.get(f"/api/inventory/miners/{uuid4()}") self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["detail"], "Miner not found") def test_malformed_uuid_returns_client_error(self): self.authenticate() response = self.client.get("/api/inventory/miners/not-a-uuid") self.assertEqual(response.status_code, 422) def test_asset_id_history_malformed_uuid_returns_validation_error(self): self.authenticate() response = self.client.get( "/api/inventory/miners/not-a-uuid/asset-id-history" ) self.assertEqual(response.status_code, 422) self.assertIsNone(self.service.history_miner_id) def test_asset_id_history_unknown_miner_returns_not_found(self): self.authenticate() miner_id = uuid4() self.service.history_error = MinerNotFound(miner_id) response = self.client.get( f"/api/inventory/miners/{miner_id}/asset-id-history" ) self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["detail"], "Miner not found") def test_asset_id_history_empty_collection_uses_typed_envelope(self): self.authenticate() miner_id = uuid4() response = self.client.get( f"/api/inventory/miners/{miner_id}/asset-id-history" ) self.assertEqual(response.status_code, 200) self.assertEqual( response.json(), {"data": [], "meta": {}, "error": None}, ) self.assertEqual(self.service.history_miner_id, miner_id) def test_asset_id_history_preserves_chronological_service_order(self): self.authenticate() miner_id = uuid4() first = datetime(2026, 7, 27, 12, 0, tzinfo=timezone.utc) second = datetime(2026, 7, 28, 12, 0, tzinfo=timezone.utc) self.service.history_result = [ history_entry( old_asset_id="ASSET-FIRST", new_asset_id="ASSET-SECOND", changed_at=first, ), history_entry( old_asset_id="ASSET-SECOND", new_asset_id="ASSET-FIRST", changed_at=second, ), ] response = self.client.get( f"/api/inventory/miners/{miner_id}/asset-id-history" ) self.assertEqual(response.status_code, 200) data = response.json()["data"] self.assertEqual( [entry["new_asset_id"] for entry in data], ["ASSET-SECOND", "ASSET-FIRST"], ) self.assertEqual( data[0]["changed_by"], "00000000-0000-0000-0000-000000000951", ) self.assertEqual( data[0]["changed_by_display_name"], "Inventory Test User", ) self.assertEqual(response.json()["meta"], {}) self.assertIsNone(response.json()["error"]) def test_router_has_no_sql_or_database_driver_imports(self): router_path = ( Path(__file__).parents[1] / "app" / "inventory" / "router.py" ) source = router_path.read_text(encoding="utf-8") tree = ast.parse(source) 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) database_modules = ("psycopg", "asyncpg", "sqlalchemy") self.assertFalse( any( module == name or module.startswith(f"{name}.") for module in imported_modules for name in database_modules ) ) upper_source = source.upper() for keyword in ("SELECT ", "INSERT ", "UPDATE ", "DELETE "): self.assertNotIn(keyword, upper_source) if __name__ == "__main__": unittest.main()