tegwick
dafc812daa
All seven data/resources records now carry description, decision, reef/secret refs, and potential vs actual consumers. Discovered resources use decision.status: gap. Validators require live inventory to be 0.3. WP-0006 T03 done; workplan finished.
179 lines
7.9 KiB
Python
179 lines
7.9 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_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)
|
|
self.assertTrue(all(r["record_scope"] == "example" for r in v02))
|
|
for record in v02:
|
|
validate_record(record)
|
|
|
|
def test_live_inventory_is_v03_with_consumers(self):
|
|
live = [
|
|
(path, record)
|
|
for path, record in self.records()
|
|
if record.get("record_scope") == "inventory"
|
|
]
|
|
self.assertGreaterEqual(len(live), 7)
|
|
for path, record in live:
|
|
with self.subTest(path=path):
|
|
self.assertEqual(record["schema_version"], "0.3")
|
|
for facet in (
|
|
"description",
|
|
"decision",
|
|
"operational_refs",
|
|
"credential_handles",
|
|
"consumers",
|
|
):
|
|
self.assertIn(facet, record)
|
|
consumers = record["consumers"]
|
|
self.assertIn("potential", consumers)
|
|
self.assertIn("actual", consumers)
|
|
self.assertTrue(consumers["potential"])
|
|
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()
|