diff --git a/backend/app/inventory/models.py b/backend/app/inventory/models.py index 3a15b4c..b193911 100644 --- a/backend/app/inventory/models.py +++ b/backend/app/inventory/models.py @@ -43,6 +43,14 @@ class InventoryMinerDetail(InventoryMinerSummary): first_seen_at: datetime | None = None +class InventoryAssetIDHistoryEntry(BaseModel): + old_asset_id: str | None = None + new_asset_id: str + changed_at: datetime + changed_by: UUID + changed_by_display_name: str + + class InventoryCreateRequest(BaseModel): asset_id: str = Field(min_length=1, max_length=120) current_mac: str | None = Field(default=None, max_length=64) @@ -112,3 +120,9 @@ class InventoryMinerDetailResponse(BaseModel): data: InventoryMinerDetail meta: dict[str, object] = Field(default_factory=dict) error: None = None + + +class InventoryAssetIDHistoryResponse(BaseModel): + data: list[InventoryAssetIDHistoryEntry] + meta: dict[str, object] = Field(default_factory=dict) + error: None = None diff --git a/backend/app/inventory/repository.py b/backend/app/inventory/repository.py index 7b7035e..c1ec443 100644 --- a/backend/app/inventory/repository.py +++ b/backend/app/inventory/repository.py @@ -347,15 +347,19 @@ class InventoryRepository: cur.execute( """ SELECT - id, - miner_id, - old_asset_id, - new_asset_id, - changed_at, - changed_by - FROM miner_asset_id_history - WHERE miner_id = %s - ORDER BY changed_at, id + history.id, + history.miner_id, + history.old_asset_id, + history.new_asset_id, + history.changed_at, + history.changed_by, + changed_by_user.display_name + AS changed_by_display_name + FROM miner_asset_id_history AS history + JOIN users AS changed_by_user + ON changed_by_user.id = history.changed_by + WHERE history.miner_id = %s + ORDER BY history.changed_at, history.id """, (miner_id,), ) diff --git a/backend/app/inventory/router.py b/backend/app/inventory/router.py index 262b933..980cf22 100644 --- a/backend/app/inventory/router.py +++ b/backend/app/inventory/router.py @@ -13,6 +13,7 @@ from app.inventory.exceptions import ( MinerNotFound, ) from app.inventory.models import ( + InventoryAssetIDHistoryResponse, InventoryCreateRequest, InventoryListMeta, InventoryMetadataUpdateRequest, @@ -120,6 +121,27 @@ def list_inventory_miners( ) +@router.get( + "/{miner_id}/asset-id-history", + response_model=InventoryAssetIDHistoryResponse, +) +def get_inventory_miner_asset_id_history( + miner_id: UUID, + service: InventoryServiceDependency, + current_user: InventoryReadUser, +) -> InventoryAssetIDHistoryResponse: + del current_user + try: + history = service.get_asset_id_history(miner_id) + except MinerNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Miner not found", + ) from exc + + return InventoryAssetIDHistoryResponse(data=history) + + @router.get("/{miner_id}", response_model=InventoryMinerDetailResponse) def get_inventory_miner( miner_id: UUID, diff --git a/backend/app/inventory/service.py b/backend/app/inventory/service.py index 18bd20a..f0706b8 100644 --- a/backend/app/inventory/service.py +++ b/backend/app/inventory/service.py @@ -12,6 +12,7 @@ from app.inventory.exceptions import ( MinerNotFound, ) from app.inventory.models import ( + InventoryAssetIDHistoryEntry, InventoryCreateRequest, InventoryMetadataUpdateRequest, InventoryMinerDetail, @@ -112,6 +113,17 @@ class InventoryService: raise MinerNotFound(miner_id) return self._detail_model(row) + def get_asset_id_history( + self, + miner_id: UUID, + ) -> list[InventoryAssetIDHistoryEntry]: + if not self.repository.get_miner_by_id(miner_id): + raise MinerNotFound(miner_id) + return [ + InventoryAssetIDHistoryEntry.model_validate(row) + for row in self.repository.list_asset_id_history(miner_id) + ] + def list_miners( self, *, diff --git a/backend/tests/test_inventory_domain.py b/backend/tests/test_inventory_domain.py index 6840ac4..091c091 100644 --- a/backend/tests/test_inventory_domain.py +++ b/backend/tests/test_inventory_domain.py @@ -17,7 +17,9 @@ from app.inventory import ( InventoryService, InventoryUpdateRequest, InventoryValidationError, + MinerNotFound, ) +from app.inventory.models import InventoryAssetIDHistoryEntry def inventory_row(**overrides): @@ -121,6 +123,7 @@ class FakeInventoryRepository: "new_asset_id": new_asset_id, "changed_at": datetime.now(timezone.utc), "changed_by": changed_by, + "changed_by_display_name": "Test Actor", } self.asset_id_history.append(history) return history @@ -207,6 +210,60 @@ class InventoryServiceTests(unittest.TestCase): with self.assertRaises(InventoryValidationError): self.service.normalize_tags(["valid", "\t\n"]) + def test_asset_id_history_empty_for_known_miner(self): + row = inventory_row() + self.repository.rows[row["id"]] = row + + self.assertEqual( + self.service.get_asset_id_history(row["id"]), + [], + ) + + def test_asset_id_history_maps_populated_repository_results(self): + row = inventory_row() + self.repository.rows[row["id"]] = row + first_changed_at = datetime(2026, 7, 27, tzinfo=timezone.utc) + second_changed_at = datetime(2026, 7, 28, tzinfo=timezone.utc) + self.repository.asset_id_history = [ + { + "id": uuid4(), + "miner_id": row["id"], + "old_asset_id": "ASSET-FIRST", + "new_asset_id": "ASSET-SECOND", + "changed_at": first_changed_at, + "changed_by": self.actor_user_id, + "changed_by_display_name": "Test Actor", + }, + { + "id": uuid4(), + "miner_id": row["id"], + "old_asset_id": "ASSET-SECOND", + "new_asset_id": "ASSET-THIRD", + "changed_at": second_changed_at, + "changed_by": self.actor_user_id, + "changed_by_display_name": "Test Actor", + }, + ] + + history = self.service.get_asset_id_history(row["id"]) + + self.assertEqual(len(history), 2) + self.assertTrue( + all( + isinstance(entry, InventoryAssetIDHistoryEntry) + for entry in history + ) + ) + self.assertEqual( + [entry.new_asset_id for entry in history], + ["ASSET-SECOND", "ASSET-THIRD"], + ) + self.assertEqual(history[0].changed_by_display_name, "Test Actor") + + def test_asset_id_history_unknown_miner_raises_not_found(self): + with self.assertRaises(MinerNotFound): + self.service.get_asset_id_history(uuid4()) + def test_metadata_update_preserves_unspecified_fields(self): row = inventory_row( asset_tag="ASSET-001", @@ -472,6 +529,14 @@ class InventoryRepositoryIntegrationTests(unittest.TestCase): InventoryMetadataUpdateRequest(asset_id="ASSET-THIRD"), actor_user_id=self.actor_user_id, ) + other = service.create_miner( + InventoryCreateRequest(asset_id="OTHER-FIRST") + ) + service.update_metadata( + other.id, + InventoryMetadataUpdateRequest(asset_id="OTHER-SECOND"), + actor_user_id=self.actor_user_id, + ) self.assertEqual(created.id, second.id) self.assertEqual(second.id, third.id) @@ -491,6 +556,17 @@ class InventoryRepositoryIntegrationTests(unittest.TestCase): history[0]["changed_at"], history[1]["changed_at"], ) + self.assertTrue( + all( + row["miner_id"] == created.id + and row["changed_by_display_name"] + == "Inventory History Test" + for row in history + ) + ) + other_history = self.repository.list_asset_id_history(other.id) + self.assertEqual(len(other_history), 1) + self.assertEqual(other_history[0]["new_asset_id"], "OTHER-SECOND") def test_history_failure_rolls_back_asset_id_update(self): service = InventoryService(self.repository) diff --git a/backend/tests/test_inventory_router.py b/backend/tests/test_inventory_router.py index d6a9bf6..d4e2228 100644 --- a/backend/tests/test_inventory_router.py +++ b/backend/tests/test_inventory_router.py @@ -15,6 +15,7 @@ from app.inventory.exceptions import ( MinerNotFound, ) from app.inventory.models import ( + InventoryAssetIDHistoryEntry, InventoryCreateRequest, InventoryMetadataUpdateRequest, InventoryMinerDetail, @@ -39,6 +40,9 @@ class StubInventoryService: 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 def search_miners( self, @@ -82,6 +86,15 @@ class StubInventoryService: 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) @@ -106,6 +119,21 @@ def miner_detail(miner_id: UUID | None = None) -> InventoryMinerDetail: ) +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() @@ -350,11 +378,26 @@ class InventoryRouterTests(unittest.TestCase): 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()] @@ -457,6 +500,80 @@ class InventoryRouterTests(unittest.TestCase): 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"