237 lines
8.1 KiB
Python
237 lines
8.1 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 (
|
|
InventoryValidationError,
|
|
MinerNotFound,
|
|
)
|
|
from app.inventory.models import (
|
|
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
|
|
|
|
def search_miners(
|
|
self,
|
|
filters: InventorySearchFilters,
|
|
) -> list[InventoryMinerSummary]:
|
|
self.search_filters = filters
|
|
if self.search_error is not None:
|
|
raise self.search_error
|
|
return self.search_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 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:
|
|
summary = miner_summary(miner_id)
|
|
return InventoryMinerDetail(
|
|
**summary.model_dump(),
|
|
miner_sn="MINER-001",
|
|
control_board_sn="CB-001",
|
|
psu_serial="PSU-001",
|
|
)
|
|
|
|
|
|
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_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_list_requires_miners_view_permission(self):
|
|
self.authenticate(permissions=())
|
|
response = self.client.get("/api/inventory/miners")
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
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},
|
|
)
|
|
self.assertIsNone(body["error"])
|
|
|
|
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"),
|
|
("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.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_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()
|