82 lines
3.6 KiB
Python
82 lines
3.6 KiB
Python
|
import json
|
||
|
|
import sys
|
||
|
|
import unittest
|
||
|
|
from copy import deepcopy
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT = Path(__file__).parents[1]
|
||
|
|
sys.path.insert(0, str(ROOT / "tools"))
|
||
|
|
from portfolio import validate_record, validate_transition
|
||
|
|
|
||
|
|
|
||
|
|
class PortfolioTest(unittest.TestCase):
|
||
|
|
def records(self):
|
||
|
|
paths = list((ROOT / "data/resources").glob("*.json"))
|
||
|
|
paths += list((ROOT / "examples/portfolio").glob("*.json"))
|
||
|
|
return [(path, json.loads(path.read_text())) for path in paths]
|
||
|
|
|
||
|
|
def test_inventory_and_examples_are_semantically_valid(self):
|
||
|
|
records = self.records()
|
||
|
|
self.assertGreaterEqual(len(records), 4)
|
||
|
|
for path, record in records:
|
||
|
|
with self.subTest(path=path):
|
||
|
|
validate_record(record)
|
||
|
|
|
||
|
|
def test_examples_cover_required_portfolio_shapes(self):
|
||
|
|
classes = {record["resource_class"] for _, record in self.records()}
|
||
|
|
self.assertTrue({"storage", "self_managed_service", "kubernetes_capacity", "shared_platform_service"} <= classes)
|
||
|
|
models = {record["management_model"] for _, record in self.records()}
|
||
|
|
self.assertEqual({"provider_managed", "self_managed", "shared_capacity"}, models)
|
||
|
|
|
||
|
|
def test_shared_resource_requires_allocation_driver(self):
|
||
|
|
record = deepcopy(next(r for _, r in self.records() if r["ownership"]["allocation"]["mode"] == "shared"))
|
||
|
|
record["ownership"]["allocation"]["driver"] = None
|
||
|
|
with self.assertRaisesRegex(ValueError, "shared resources"):
|
||
|
|
validate_record(record)
|
||
|
|
|
||
|
|
def test_unknown_commission_date_is_preserved(self):
|
||
|
|
record = deepcopy(next(r for _, r in self.records() if r["status"] == "active"))
|
||
|
|
record["lifecycle"]["commissioned_on"] = None
|
||
|
|
validate_record(record)
|
||
|
|
|
||
|
|
def test_lifecycle_dates_cannot_be_out_of_order(self):
|
||
|
|
record = deepcopy(self.records()[0][1])
|
||
|
|
record["lifecycle"]["ordered_on"] = "2026-08-12"
|
||
|
|
record["lifecycle"]["commissioned_on"] = "2026-08-11"
|
||
|
|
with self.assertRaisesRegex(ValueError, "out of order"):
|
||
|
|
validate_record(record)
|
||
|
|
|
||
|
|
def test_lifecycle_transition_rules(self):
|
||
|
|
validate_transition("proposed", "ordered")
|
||
|
|
validate_transition("active", "retiring")
|
||
|
|
validate_transition("suspended", "active")
|
||
|
|
with self.assertRaisesRegex(ValueError, "invalid lifecycle transition"):
|
||
|
|
validate_transition("proposed", "active")
|
||
|
|
with self.assertRaisesRegex(ValueError, "invalid lifecycle transition"):
|
||
|
|
validate_transition("retired", "active")
|
||
|
|
|
||
|
|
def test_inventory_relationships_resolve_to_inventory_records(self):
|
||
|
|
inventory = {r["id"]: r for _, r in self.records() if r["record_scope"] == "inventory"}
|
||
|
|
for record in inventory.values():
|
||
|
|
for relationship in record["relationships"]:
|
||
|
|
self.assertIn(relationship["resource_id"], inventory)
|
||
|
|
|
||
|
|
def test_initial_coverage_references_the_complete_inventory(self):
|
||
|
|
inventory = {r["id"] for _, r in self.records() if r["record_scope"] == "inventory"}
|
||
|
|
coverage = json.loads((ROOT / "data/portfolio-coverage-2026-08-11.json").read_text())
|
||
|
|
referenced = {
|
||
|
|
resource_id
|
||
|
|
for group in coverage["coverage"]
|
||
|
|
for resource_id in group["resource_ids"]
|
||
|
|
}
|
||
|
|
self.assertEqual(len(inventory), coverage["inventory_records"])
|
||
|
|
self.assertEqual(inventory, referenced)
|
||
|
|
self.assertEqual(
|
||
|
|
{"helix-forge", "coulomb-social", "shared-railiance", "representative-tenant"},
|
||
|
|
{group["group"] for group in coverage["coverage"]},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|