306 lines
10 KiB
Python
306 lines
10 KiB
Python
import ast
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import UUID, uuid4
|
|
|
|
from app.inventory import (
|
|
DuplicateAssetID,
|
|
InventoryCreateRequest,
|
|
InventoryRepository,
|
|
InventorySearchFilters,
|
|
InventoryService,
|
|
InventoryUpdateRequest,
|
|
InventoryValidationError,
|
|
)
|
|
|
|
|
|
def inventory_row(**overrides):
|
|
now = datetime.now(timezone.utc)
|
|
row = {
|
|
"id": uuid4(),
|
|
"site_id": None,
|
|
"room_id": None,
|
|
"asset_tag": "ASSET-001",
|
|
"current_mac": None,
|
|
"current_ip": None,
|
|
"hostname": None,
|
|
"miner_sn": None,
|
|
"control_board_sn": None,
|
|
"power_sn": None,
|
|
"model": None,
|
|
"firmware_version": None,
|
|
"lifecycle_status": "active",
|
|
"status": "unknown",
|
|
"notes": None,
|
|
"first_seen_at": None,
|
|
"last_seen_at": None,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"tags": [],
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
class FakeInventoryRepository:
|
|
def __init__(self):
|
|
self.rows = {}
|
|
self.created_values = None
|
|
|
|
@contextmanager
|
|
def transaction(self):
|
|
yield self
|
|
|
|
def get_miner_by_id(self, miner_id):
|
|
return self.rows.get(miner_id)
|
|
|
|
def get_miner_by_asset_id(self, asset_id):
|
|
key = asset_id.strip().casefold()
|
|
return next(
|
|
(
|
|
row
|
|
for row in self.rows.values()
|
|
if str(row.get("asset_tag") or "").strip().casefold() == key
|
|
),
|
|
None,
|
|
)
|
|
|
|
def list_miners(self, *, limit=100, offset=0):
|
|
return list(self.rows.values())[offset : offset + limit]
|
|
|
|
def search_miners(self, **kwargs):
|
|
rows = list(self.rows.values())
|
|
lifecycle = kwargs.get("lifecycle_status")
|
|
if lifecycle:
|
|
rows = [
|
|
row for row in rows if row["lifecycle_status"] == lifecycle
|
|
]
|
|
return rows[kwargs.get("offset", 0) : kwargs.get("limit", 100)]
|
|
|
|
def create_miner(self, values):
|
|
self.created_values = dict(values)
|
|
row = inventory_row(**values)
|
|
self.rows[row["id"]] = row
|
|
return row
|
|
|
|
def update_miner(self, miner_id, values):
|
|
row = self.rows.get(miner_id)
|
|
if not row:
|
|
return None
|
|
row.update(values)
|
|
row["updated_at"] = datetime.now(timezone.utc)
|
|
return row
|
|
|
|
def add_tag(self, miner_id, tag_name):
|
|
now = datetime.now(timezone.utc)
|
|
tag = {
|
|
"id": uuid4(),
|
|
"name": tag_name,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
existing = {
|
|
item["name"].strip().casefold()
|
|
for item in self.rows[miner_id]["tags"]
|
|
}
|
|
if tag_name.casefold() not in existing:
|
|
self.rows[miner_id]["tags"].append(tag)
|
|
return tag
|
|
|
|
def remove_tag(self, miner_id, tag_id):
|
|
before = len(self.rows[miner_id]["tags"])
|
|
self.rows[miner_id]["tags"] = [
|
|
tag for tag in self.rows[miner_id]["tags"] if tag["id"] != tag_id
|
|
]
|
|
return len(self.rows[miner_id]["tags"]) != before
|
|
|
|
|
|
class InventoryServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.repository = FakeInventoryRepository()
|
|
self.service = InventoryService(self.repository)
|
|
|
|
def test_create_normalizes_asset_id_tags_and_never_seen(self):
|
|
created = self.service.create_miner(
|
|
InventoryCreateRequest(
|
|
asset_id=" Mine-001 ",
|
|
psu_serial=" PSU-9 ",
|
|
tags=[" Priority ", "priority", "\tRepair\n"],
|
|
)
|
|
)
|
|
|
|
self.assertEqual(created.asset_id, "Mine-001")
|
|
self.assertEqual(created.psu_serial, "PSU-9")
|
|
self.assertEqual([tag.name for tag in created.tags], ["Priority", "Repair"])
|
|
self.assertIsNone(created.first_seen_at)
|
|
self.assertIsNone(created.last_seen_at)
|
|
self.assertIsNone(self.repository.created_values["first_seen_at"])
|
|
self.assertIsNone(self.repository.created_values["last_seen_at"])
|
|
|
|
def test_duplicate_asset_id_is_case_and_whitespace_insensitive(self):
|
|
row = inventory_row(asset_tag="Mine-001")
|
|
self.repository.rows[row["id"]] = row
|
|
|
|
with self.assertRaises(DuplicateAssetID):
|
|
self.service.create_miner(
|
|
InventoryCreateRequest(asset_id=" mine-001 ")
|
|
)
|
|
|
|
def test_lifecycle_validation_is_owned_by_service(self):
|
|
self.assertEqual(
|
|
self.service.validate_lifecycle_status(" SPARE "),
|
|
"spare",
|
|
)
|
|
with self.assertRaises(InventoryValidationError):
|
|
self.service.create_miner(
|
|
InventoryCreateRequest(
|
|
asset_id="ASSET-BAD",
|
|
lifecycle_status="warehouse",
|
|
)
|
|
)
|
|
|
|
def test_blank_asset_and_tags_are_rejected(self):
|
|
with self.assertRaises(InventoryValidationError):
|
|
self.service.normalize_asset_id(" ")
|
|
with self.assertRaises(InventoryValidationError):
|
|
self.service.normalize_tags(["valid", "\t\n"])
|
|
|
|
def test_update_maps_product_names_without_repository_rules(self):
|
|
row = inventory_row(asset_tag="ASSET-OLD", power_sn="PSU-OLD")
|
|
self.repository.rows[row["id"]] = row
|
|
|
|
updated = self.service.update_miner(
|
|
row["id"],
|
|
InventoryUpdateRequest(
|
|
asset_id=" ASSET-NEW ",
|
|
psu_serial=" PSU-NEW ",
|
|
lifecycle_status="retired",
|
|
),
|
|
)
|
|
|
|
self.assertEqual(updated.asset_id, "ASSET-NEW")
|
|
self.assertEqual(updated.psu_serial, "PSU-NEW")
|
|
self.assertEqual(updated.lifecycle_status, "retired")
|
|
|
|
def test_search_validation_precedes_repository(self):
|
|
with self.assertRaises(InventoryValidationError):
|
|
self.service.search_miners(
|
|
InventorySearchFilters(lifecycle_status="broken")
|
|
)
|
|
|
|
|
|
class InventoryRepositoryIntegrationTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.repository = InventoryRepository()
|
|
|
|
def setUp(self):
|
|
with self.repository.transaction() as repository:
|
|
with repository._connection.cursor() as cur:
|
|
cur.execute("DELETE FROM miner_tags")
|
|
cur.execute("DELETE FROM tags")
|
|
cur.execute("DELETE FROM miners")
|
|
|
|
def test_repository_crud_search_and_tags(self):
|
|
created = self.repository.create_miner(
|
|
{
|
|
"asset_tag": "REPO-001",
|
|
"current_ip": "192.0.2.10",
|
|
"model": "M50",
|
|
"miner_sn": "MINER-001",
|
|
"control_board_sn": "CB-001",
|
|
"power_sn": "PSU-001",
|
|
"lifecycle_status": "active",
|
|
"status": "unknown",
|
|
"first_seen_at": None,
|
|
"last_seen_at": None,
|
|
}
|
|
)
|
|
miner_id = created["id"]
|
|
|
|
self.assertIsInstance(miner_id, UUID)
|
|
self.assertIsNone(created["first_seen_at"])
|
|
self.assertIsNone(created["last_seen_at"])
|
|
self.assertEqual(
|
|
self.repository.get_miner_by_asset_id(" repo-001 ")["id"],
|
|
miner_id,
|
|
)
|
|
self.assertEqual(self.repository.get_miner_by_id(miner_id)["model"], "M50")
|
|
self.assertEqual(len(self.repository.list_miners()), 1)
|
|
self.assertEqual(
|
|
self.repository.search_miners(query="CB-001")[0]["id"],
|
|
miner_id,
|
|
)
|
|
|
|
updated = self.repository.update_miner(
|
|
miner_id,
|
|
{"lifecycle_status": "spare", "notes": "Bench stock"},
|
|
)
|
|
self.assertEqual(updated["lifecycle_status"], "spare")
|
|
self.assertEqual(
|
|
self.repository.search_miners(lifecycle_status="spare")[0]["id"],
|
|
miner_id,
|
|
)
|
|
|
|
tag = self.repository.add_tag(miner_id, "Priority")
|
|
repeated = self.repository.add_tag(miner_id, " priority ")
|
|
self.assertEqual(tag["id"], repeated["id"])
|
|
self.assertEqual(len(self.repository.list_tags()), 1)
|
|
self.assertEqual(
|
|
self.repository.search_miners(tag_names=["priority"])[0]["id"],
|
|
miner_id,
|
|
)
|
|
self.assertEqual(
|
|
len(self.repository.get_miner_by_id(miner_id)["tags"]),
|
|
1,
|
|
)
|
|
self.assertTrue(self.repository.remove_tag(miner_id, tag["id"]))
|
|
self.assertEqual(
|
|
self.repository.get_miner_by_id(miner_id)["tags"],
|
|
[],
|
|
)
|
|
|
|
def test_transaction_rolls_back_repository_writes(self):
|
|
with self.assertRaises(RuntimeError):
|
|
with self.repository.transaction() as repository:
|
|
repository.create_miner(
|
|
{
|
|
"asset_tag": "ROLLBACK-001",
|
|
"status": "unknown",
|
|
"first_seen_at": None,
|
|
"last_seen_at": None,
|
|
}
|
|
)
|
|
raise RuntimeError("rollback")
|
|
|
|
self.assertIsNone(
|
|
self.repository.get_miner_by_asset_id("ROLLBACK-001")
|
|
)
|
|
|
|
|
|
class InventoryArchitectureTests(unittest.TestCase):
|
|
def test_inventory_package_has_no_http_layer_or_router(self):
|
|
import app.inventory as inventory
|
|
|
|
inventory_path = Path(__file__).parents[1] / "app" / "inventory"
|
|
for source_path in inventory_path.glob("*.py"):
|
|
tree = ast.parse(source_path.read_text(encoding="utf-8"))
|
|
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)
|
|
self.assertFalse(
|
|
any(module.startswith("fastapi") for module in imported_modules),
|
|
source_path,
|
|
)
|
|
|
|
self.assertFalse(hasattr(inventory, "router"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|