690 lines
25 KiB
Python
690 lines
25 KiB
Python
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()
|