Add inventory domain foundation
This commit is contained in:
lussierj committed
1 parent
d4df20b740
commit
d1d02962da
8 files changed
+1135
No files matched your search
@@ -0,0 +1,305 @@
|
||||
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()
|
||||
Reference in new issue
Block a user