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_ordered_resource_requires_approved_decision(self): record = deepcopy(next( r for _, r in self.records() if r["id"] == "resource:platform:audit-storage" )) record["status"] = "ordered" record["decision"]["status"] = "draft" with self.assertRaisesRegex(ValueError, "approved decision"): validate_record(record) record["decision"]["status"] = "approved" record["decision"]["approved_by"] = "human" record["decision"]["approved_on"] = "2026-08-14" validate_record(record) def test_v03_requires_five_facets(self): record = deepcopy(next( r for _, r in self.records() if r["id"] == "resource:platform:audit-storage" )) self.assertEqual(record["schema_version"], "0.3") del record["consumers"] with self.assertRaisesRegex(ValueError, "five facets"): validate_record(record) def test_inline_endpoint_is_rejected(self): record = deepcopy(next( r for _, r in self.records() if r["id"] == "resource:platform:audit-storage" )) record["description"] += " https://s3.nl-ams.scw.cloud" with self.assertRaisesRegex(ValueError, "inline provider endpoint"): validate_record(record) def test_inline_secret_is_rejected(self): record = deepcopy(next( r for _, r in self.records() if r["id"] == "resource:platform:audit-storage" )) record["description"] += " AGE-SECRET-KEY-1TEST" with self.assertRaisesRegex(ValueError, "inline secret"): validate_record(record) def test_v02_records_remain_valid(self): v02 = [r for _, r in self.records() if r.get("schema_version") == "0.2"] self.assertGreaterEqual(len(v02), 1) for record in v02: 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"]}, ) class ReefViewTest(unittest.TestCase): def test_reef_railiance_view_cites_only_inventory_and_excludes_s3(self): view = json.loads((ROOT / "data/reefs/reef-railiance.json").read_text()) inventory = { json.loads(path.read_text())["id"] for path in (ROOT / "data/resources").glob("*.json") } self.assertEqual("reef-railiance", view["reef_id"]) self.assertTrue(view["declaration_ref"].startswith("reef:")) self.assertEqual("compute_substrate", view["role"]) for row in view["resources"]: self.assertIn(row["resource_id"], inventory) self.assertNotIn("resource:platform:audit-storage", {row["resource_id"] for row in view["resources"]}) self.assertTrue(any("reef-storage" in note for note in view["notes"])) def test_reef_storage_view_is_delegated_object_store(self): view = json.loads((ROOT / "data/reefs/reef-storage.json").read_text()) self.assertEqual("storage_substrate", view["role"]) self.assertEqual( ["resource:platform:audit-storage"], [row["resource_id"] for row in view["resources"]], ) self.assertEqual(["rapp-postgres"], view["consumers_potential"]) # rapp-postgres became an actual consumer on 2026-08-14: continuous WAL # archiving and a proven restore make it a realised consumer, not a # potential one. self.assertEqual(["rapp-postgres"], view["consumers_actual"]) if __name__ == "__main__": unittest.main()