feat: add asset ID history API

This commit is contained in:
lussierj committed 2026-07-28 19:48:57 +00:00
1 parent a87ab2709f
commit d738b8e27e
6 files changed
+254 -9

No files matched your search

+76
View File
@@ -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)
+117
View File
@@ -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"