coulomb/resource-control
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(portfolio): complete RESOURCE-WP-0003 T06 optimization cases and T07 reporting

Browse source 
T06: optimization-case schema, fail-closed evaluator, cadence and decision
template. Every option including the baseline must present all ten decision
fields; one unknown blocks the comparison. Validated on the storage case
(Hetzner computes and loses to Scaleway by EUR 29.14/month on operator labour;
Host Europe blocks on four named gaps) and on the non-storage reef-railiance
k3s rightsizing case (low utilization is real, but nothing is costable while
the railiance01 price is unknown).

T07: portfolio report over coverage, lifecycle, utilization, cost, renewals,
risks, open cases, and next actions, derived only from committed evidence.
Portfolio spend is reported null rather than as a partial sum, unattributed
cost is a named list rather than a spread, and unmeasurable resources are
reported rather than dropped.

RESOURCE-WP-0003 is finished; both cases remain blocked_on_evidence against
live delegated records in other repositories. RESOURCE-WP-0002 is untouched.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-08-14 09:28:44 +02:00
parent 54dd45c926
commit 2c2a6073ff
63 changed files with 4662 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

54
tests/test_control_cycle.py Normal file
View file

@ -0,0 +1,54 @@
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from control_cycle import compare
def record(kind="forecast", resource_class="storage"):
result = {
"record_id": f"test:{kind}", "record_type": kind,
"resource_id": "resource:test", "resource_class": resource_class,
"period": "2026-09", "usage_proxies": {"stored_gb": {"value": 100, "unit": "GB-month"}},
"costs": {"currency": "EUR", "infrastructure": 10, "internal_labor": 60, "external_labor": 0, "total": 70},
"allocation": {"method": "direct", "driver": "stored_gb", "method_version": "1", "unattributed_eur": 0},
}
if kind == "actual":
result["forecast_ref"] = "test:forecast"
return result
class ControlCycleTest(unittest.TestCase):
def test_same_mechanism_supports_required_resource_classes(self):
for resource_class in ("storage", "cluster_compute", "shared_platform_service"):
forecast = record(resource_class=resource_class)
actual = record("actual", resource_class)
actual["usage_proxies"]["stored_gb"]["value"] = 120
result = compare(forecast, actual)
self.assertEqual(resource_class, result["resource_class"])
self.assertEqual("demand", result["usage_proxies"]["stored_gb"]["category"])
def test_explicit_attribution_and_cost_split_are_preserved(self):
forecast, actual = record(), record("actual")
actual["costs"].update(infrastructure=12, internal_labor=90, total=102)
actual["variance_attribution"] = {"costs.infrastructure": "model", "costs.internal_labor": "labor"}
result = compare(forecast, actual)
self.assertEqual("model", result["costs"]["infrastructure"]["category"])
self.assertEqual(30, result["costs"]["internal_labor"]["error"])
def test_rejects_wrong_forecast_reference(self):
forecast, actual = record(), record("actual")
actual["forecast_ref"] = "test:other"
with self.assertRaisesRegex(ValueError, "immutable forecast"):
compare(forecast, actual)
def test_missing_proxy_is_data_quality_error(self):
forecast, actual = record(), record("actual")
actual["usage_proxies"] = {"cpu_hours": {"value": 4, "unit": "vCPU-hour"}}
result = compare(forecast, actual)
self.assertEqual("data_quality", result["usage_proxies"]["stored_gb"]["category"])
if __name__ == "__main__":
unittest.main()

62
tests/test_cost_model.py Normal file
View file

@ -0,0 +1,62 @@
import json
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from cost_model import forecast
class CostModelTest(unittest.TestCase):
def setUp(self):
root = Path(__file__).parents[1]
self.demand = json.loads((root / "data/demand/platform-audit-storage.json").read_text())
self.providers = json.loads((root / "data/providers/object-storage.json").read_text())
def test_all_scenarios_and_providers_are_projected_for_12_months(self):
result = forecast(self.demand, self.providers)
for rows in result["scenarios"].values():
self.assertEqual(12 * len(self.providers["providers"]), len(rows))
def test_unknown_prices_never_become_zero(self):
result = forecast(self.demand, self.providers)
host_europe = [r for r in result["scenarios"]["base"] if r["provider_id"] == "host-europe-cloud-storage"]
self.assertTrue(all(r["recurring_total_eur"] is None for r in host_europe))
self.assertTrue(all(r["missing_price_fields"] for r in host_europe))
def test_storage_grows_month_over_month(self):
result = forecast(self.demand, self.providers)
rows = [r for r in result["scenarios"]["high"] if r["provider_id"] == "scaleway-standard-multi-az"]
self.assertGreater(rows[-1]["stored_gb"], rows[0]["stored_gb"])
def test_fixed_capacity_self_hosted_option_fails_closed_when_full(self):
result = forecast(self.demand, self.providers)
rows = [r for r in result["scenarios"]["base"] if r["provider_id"] == "hetzner-garage-3"]
self.assertIsNotNone(rows[0]["recurring_total_eur"])
self.assertIsNone(rows[-1]["recurring_total_eur"])
def test_managed_cloud_comparators_calculate(self):
result = forecast(self.demand, self.providers)
ids = {"aws-s3-standard", "azure-blob-hot-zrs", "gcp-cloud-storage-standard", "stackit-object-storage"}
rows = [r for r in result["scenarios"]["base"] if r["month"] == 1 and r["provider_id"] in ids]
self.assertEqual(ids, {r["provider_id"] for r in rows})
self.assertTrue(all(r["recurring_total_eur"] is not None for r in rows))
def test_recurring_total_is_infrastructure_plus_labor(self):
result = forecast(self.demand, self.providers)
row = next(r for r in result["comparison_320gb"] if r["provider_id"] == "stackit-object-storage")
self.assertAlmostEqual(row["recurring_total_eur"], row["monthly_infrastructure_eur"] + row["monthly_internal_labor_eur"])
def test_normalized_comparison_uses_320gb_for_every_provider(self):
result = forecast(self.demand, self.providers)
self.assertEqual(len(self.providers["providers"]), len(result["comparison_320gb"]))
self.assertTrue(all(r["stored_gb"] == 320 for r in result["comparison_320gb"]))
def test_garage_external_setup_services_are_unquoted(self):
result = forecast(self.demand, self.providers)
rows = [r for r in result["comparison_320gb"] if "garage" in r["provider_id"]]
self.assertTrue(all(r["setup_external_services_eur"] is None for r in rows))
if __name__ == "__main__":
unittest.main()

60
tests/test_financial_exchange.py Normal file
View file

@ -0,0 +1,60 @@
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from financial_exchange import forecast_records, reconcile, validate_booked_cost
class FinancialExchangeTest(unittest.TestCase):
def setUp(self):
self.forecast = {
"schema_version": "0.1", "record_type": "forecast",
"workload": "platform-pg", "cost_attribution_key": "platform:audit-storage",
"provider_id": "scaleway-standard-multi-az", "created_at": "2026-08-10T17:10:00Z",
"scenario": "base", "forecast_ref": None,
"rows": [{"period": "2026-09", "database_gb": 5, "stored_gb": 180,
"wal_gb": 30, "restore_egress_gb": 5, "write_requests": 2500,
"read_requests": 1000, "infrastructure_eur": 2.89,
"internal_labor_hours": 1, "internal_labor_eur": 60,
"total_eur": 62.89}]
}
def test_forecast_export_separates_costs_and_is_stable(self):
first = forecast_records(self.forecast, resource_id="resource:platform_audit_storage", source_ref="forecast.json")
second = forecast_records(self.forecast, resource_id="resource:platform_audit_storage", source_ref="forecast.json")
self.assertEqual(first, second)
self.assertEqual("2.89", first[0]["costs"]["infrastructure"])
self.assertEqual("60.00", first[0]["costs"]["internal_labor"])
self.assertEqual("usage_observation" not in first[0]["record_type"], True)
def test_reconciles_only_attributable_booked_cost(self):
forecast = forecast_records(self.forecast, resource_id="resource:platform_audit_storage", source_ref="forecast.json")
booked = [{
"schema_version": "0.1", "record_type": "booked_cost", "financial_fact_id": "fact:1",
"correction_of": None, "adjustment_kind": "charge", "source_type": "provider_invoice",
"source_document_id": "invoice:1", "source_line_id": "invoice:1:1",
"content_fingerprint": "sha256:x", "provider": "scaleway",
"accounting_period": "2026-09", "currency": "EUR", "gross_amount": "3.10",
"adjustment_amount": "0.00", "effective_amount": "3.10", "tax_status": "unknown",
"tax_amount": None, "cost_attribution_key": "platform:audit-storage",
"source_evidence_ref": "invoice:1", "recorded_at": "2026-10-01T00:00:00Z"
}]
rows = reconcile(forecast, booked)
self.assertEqual("reconciled", rows[0]["status"])
self.assertEqual("0.21", rows[0]["variance"])
self.assertEqual(["fact:1"], rows[0]["financial_fact_ids"])
def test_rejects_invalid_booked_arithmetic(self):
with self.assertRaisesRegex(ValueError, "effective_amount"):
validate_booked_cost({
"schema_version": "0.1", "record_type": "booked_cost", "financial_fact_id": "fact:1",
"adjustment_kind": "charge", "source_document_id": "doc", "source_line_id": "line",
"content_fingerprint": "hash", "provider": "provider", "accounting_period": "2026-09",
"currency": "EUR", "gross_amount": "1.00", "adjustment_amount": "0.00",
"effective_amount": "2.00", "source_evidence_ref": "doc", "recorded_at": "2026-10-01T00:00:00Z"
})
if __name__ == "__main__":
unittest.main()

248
tests/test_optimization.py Normal file
View file

@ -0,0 +1,248 @@
import json
import sys
import unittest
from copy import deepcopy
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from optimization import (
compare_option,
evaluate,
recurring_total,
utilization_ratios,
validate_case,
)
CASE_DIR = Path(__file__).parents[1] / "data" / "optimization"
def option(option_id="alt", **overrides):
result = {
"option_id": option_id,
"label": f"option {option_id}",
"one_time_eur": 0,
"recurring_infrastructure_eur_month": 10,
"recurring_internal_labor_eur_month": 60,
"recurring_external_labor_eur_month": 0,
"utilization": {
"cpu": {
"provisioned": {"value": 4, "unit": "vCPU"},
"used": {"value": 1, "unit": "vCPU"},
}
},
"uncertainty": {"level": "low", "notes": []},
"service_constraints": {"nodes": {"value": 1, "unit": "count"}},
"failure_domains": ["provider:test"],
"exit_path": "cancel and erase",
"unknowns": [],
}
result.update(overrides)
return result
def case(**overrides):
result = {
"schema_version": "0.1",
"record_scope": "illustrative",
"case_id": "opt:test:2026-08",
"case_type": "rightsizing",
"trigger": "cadence",
"review_period": "2026-08",
"created_at": "2026-08-14T00:00:00Z",
"resource_ids": ["resource:test"],
"baseline": option("baseline"),
"alternatives": [option("alt", recurring_infrastructure_eur_month=0)],
"decision": {
"state": "proposed",
"recommended_option_id": "alt",
"rationale": "cheaper",
"approver": None,
"approved_on": None,
"delegated_to": [],
},
"evidence": [
{"kind": "telemetry", "ref": "test", "authority": "test", "observed_at": None}
],
}
result.update(overrides)
return result
class CostArithmeticTest(unittest.TestCase):
def test_recurring_total_sums_all_three_cost_components(self):
self.assertEqual(70, recurring_total(option()))
def test_recurring_total_is_unknown_when_any_component_is_unknown(self):
for field in (
"recurring_infrastructure_eur_month",
"recurring_internal_labor_eur_month",
"recurring_external_labor_eur_month",
):
self.assertIsNone(recurring_total(option(**{field: None})))
def test_utilization_ratio_is_unknown_rather_than_zero_when_unmeasured(self):
unknown = option(
utilization={
"cpu": {
"provisioned": {"value": None, "unit": "vCPU"},
"used": {"value": 1, "unit": "vCPU"},
}
}
)
self.assertIsNone(utilization_ratios(unknown)["cpu"])
self.assertEqual(0.25, utilization_ratios(option())["cpu"])
class ComparisonTest(unittest.TestCase):
def test_material_saving_is_recommended_with_computed_payback(self):
baseline = option("baseline")
alternative = option("alt", recurring_infrastructure_eur_month=0, one_time_eur=100)
result = compare_option(baseline, alternative)
self.assertEqual("recommend", result["verdict"])
self.assertEqual(10, result["monthly_saving_eur"])
self.assertEqual(10.0, result["payback_months"])
def test_more_expensive_alternative_is_rejected_and_never_pays_back(self):
result = compare_option(option("baseline"), option("alt", recurring_infrastructure_eur_month=100))
self.assertEqual("reject", result["verdict"])
self.assertIsNone(result["payback_months"])
self.assertIn("never", result["payback_note"])
def test_difference_below_materiality_threshold_is_not_a_change(self):
result = compare_option(option("baseline"), option("alt", recurring_infrastructure_eur_month=8))
self.assertEqual("no_material_change", result["verdict"])
def test_labor_increase_can_cancel_an_infrastructure_saving(self):
alternative = option(
"alt", recurring_infrastructure_eur_month=0, recurring_internal_labor_eur_month=90
)
result = compare_option(option("baseline"), alternative)
self.assertEqual(20, result["monthly_delta_eur"])
self.assertEqual("reject", result["verdict"])
def test_unknown_cost_blocks_the_comparison_instead_of_assuming_zero(self):
result = compare_option(option("baseline"), option("alt", recurring_infrastructure_eur_month=None))
self.assertEqual("blocked_on_evidence", result["verdict"])
self.assertIsNone(result["monthly_saving_eur"])
self.assertIn("alt.recurring_infrastructure_eur_month", result["blocking_evidence"])
def test_missing_exit_path_blocks_an_otherwise_cheaper_option(self):
alternative = option("alt", recurring_infrastructure_eur_month=0, exit_path=None)
result = compare_option(option("baseline"), alternative)
self.assertEqual("blocked_on_evidence", result["verdict"])
self.assertFalse(result["exit_path_known"])
def test_named_unknown_blocks_even_when_every_number_is_present(self):
alternative = option("alt", recurring_infrastructure_eur_month=0, unknowns=["price not confirmed"])
result = compare_option(option("baseline"), alternative)
self.assertEqual("blocked_on_evidence", result["verdict"])
self.assertIn("alt.unknown:price not confirmed", result["blocking_evidence"])
def test_unknown_baseline_blocks_every_alternative(self):
baseline = option("baseline", recurring_infrastructure_eur_month=None)
result = compare_option(baseline, option("alt", recurring_infrastructure_eur_month=0))
self.assertEqual("blocked_on_evidence", result["verdict"])
self.assertIn("baseline.recurring_infrastructure_eur_month", result["blocking_evidence"])
def test_failure_domain_changes_are_reported_in_both_directions(self):
baseline = option("baseline", failure_domains=["provider:test", "host:one"])
alternative = option("alt", failure_domains=["provider:test", "region:two"])
result = compare_option(baseline, alternative)
self.assertEqual(["host:one"], result["failure_domains_removed"])
self.assertEqual(["region:two"], result["failure_domains_added"])
class CaseValidationTest(unittest.TestCase):
def test_valid_case_evaluates_and_picks_the_best_option(self):
record = case()
validate_case(record)
self.assertEqual("alt", evaluate(record)["best_option_id"])
def test_blocked_case_cannot_be_proposed(self):
record = case(alternatives=[option("alt", one_time_eur=None)])
with self.assertRaises(ValueError):
validate_case(record)
def test_unblocked_case_cannot_claim_to_be_blocked(self):
record = case()
record["decision"]["state"] = "blocked_on_evidence"
with self.assertRaises(ValueError):
validate_case(record)
def test_approval_requires_a_named_authority_and_date(self):
record = case()
record["decision"]["state"] = "approved"
with self.assertRaises(ValueError):
validate_case(record)
record["decision"]["approver"] = "human financial authority"
record["decision"]["approved_on"] = "2026-08-14"
validate_case(record)
def test_blocked_case_cannot_be_approved(self):
record = case(alternatives=[option("alt", exit_path=None)])
record["decision"].update(
{"state": "approved", "approver": "human", "approved_on": "2026-08-14"}
)
with self.assertRaises(ValueError):
validate_case(record)
def test_duplicate_option_identifiers_are_rejected(self):
record = case(alternatives=[option("baseline", recurring_infrastructure_eur_month=0)])
with self.assertRaises(ValueError):
validate_case(record)
def test_resource_ids_must_be_portfolio_identifiers(self):
with self.assertRaises(ValueError):
validate_case(case(resource_ids=["railiance01"]))
class RegisteredCasesTest(unittest.TestCase):
"""The two cases required by RESOURCE-WP-0003-T06: storage and non-storage."""
def setUp(self):
self.cases = {
path.stem: json.loads(path.read_text()) for path in sorted(CASE_DIR.glob("*.json"))
}
def test_every_registered_case_validates(self):
self.assertTrue(self.cases)
for record in self.cases.values():
validate_case(record)
def test_process_is_validated_on_a_storage_and_a_non_storage_candidate(self):
types = {record["case_id"] for record in self.cases.values()}
self.assertIn("opt:platform-audit-storage:2026-08", types)
self.assertIn("opt:reef-railiance-k3s:2026-08", types)
def test_storage_case_computes_hetzner_and_blocks_host_europe(self):
report = evaluate(self.cases["platform-audit-storage-2026-08"])
verdicts = {c["option_id"]: c["verdict"] for c in report["comparisons"]}
self.assertEqual("reject", verdicts["hetzner-object-storage"])
self.assertEqual("blocked_on_evidence", verdicts["host-europe-cloud-storage"])
hetzner = next(c for c in report["comparisons"] if c["option_id"] == "hetzner-object-storage")
self.assertEqual(29.14, hetzner["monthly_delta_eur"])
def test_cluster_case_reports_low_utilization_but_refuses_to_recommend(self):
report = evaluate(self.cases["reef-railiance-k3s-2026-08"])
self.assertIsNone(report["best_option_id"])
self.assertEqual(0.141, report["baseline"]["utilization"]["cpu"])
self.assertIsNone(report["baseline"]["recurring_eur_month"])
for comparison in report["comparisons"]:
self.assertEqual("blocked_on_evidence", comparison["verdict"])
self.assertTrue(comparison["blocking_evidence"])
def test_blocking_evidence_names_an_owner_for_every_cluster_unknown(self):
record = self.cases["reef-railiance-k3s-2026-08"]
for option_record in [record["baseline"]] + record["alternatives"]:
for unknown in option_record["unknowns"]:
self.assertIn("owner:", unknown)
def test_registered_cases_are_not_silently_mutated_by_evaluation(self):
for name, record in self.cases.items():
before = deepcopy(record)
evaluate(record)
self.assertEqual(before, record, name)
if __name__ == "__main__":
unittest.main()

81
tests/test_portfolio.py Normal file
View file

@ -0,0 +1,81 @@
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()

163
tests/test_portfolio_report.py Normal file
View file

@ -0,0 +1,163 @@
import sys
import unittest
from datetime import date
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from portfolio_report import (
build,
cost_section,
lifecycle_section,
renewals_section,
risk_section,
utilization_section,
)
ROOT = Path(__file__).parents[1]
TODAY = date(2026, 8, 14)
def resource(rid="resource:test", **overrides):
result = {
"id": rid,
"status": "active",
"location": {"failure_domains": ["provider:test"]},
"capacity": [
{"metric": "cpu", "value": 4, "unit": "vCPU", "kind": "usable", "observed_at": "2026-08-11"},
{"metric": "cpu_usage", "value": 1, "unit": "vCPU", "kind": "observed", "observed_at": "2026-08-11"},
],
"ownership": {"owner": "test-owner", "allocation": {"mode": "dedicated"}},
"cost": {"price_evidence": "quote#1", "billing_model": "fixed"},
"lifecycle": {"renews_on": None, "cancel_by": None},
}
result.update(overrides)
return result
class UtilizationTest(unittest.TestCase):
def test_paired_metrics_produce_a_ratio_and_a_signal(self):
section = utilization_section([resource()])
metric = section["measured"][0]["metrics"][0]
self.assertEqual(0.25, metric["ratio"])
self.assertEqual("idle", metric["signal"])
self.assertEqual(["resource:test"], section["idle"])
def test_saturated_capacity_is_flagged_separately(self):
record = resource()
record["capacity"][1]["value"] = 3.8
section = utilization_section([record])
self.assertEqual(["resource:test"], section["saturated"])
self.assertEqual([], section["idle"])
def test_unpaired_capacity_is_reported_as_unmeasured_not_omitted(self):
record = resource(capacity=[{"metric": "cpu", "value": 4, "unit": "vCPU", "kind": "usable"}])
section = utilization_section([record])
self.assertEqual([], section["measured"])
self.assertEqual("resource:test", section["unmeasured"][0]["resource_id"])
def test_zero_provisioned_capacity_does_not_divide_by_zero(self):
record = resource()
record["capacity"][0]["value"] = 0
self.assertEqual([], utilization_section([record])["measured"])
class CostTest(unittest.TestCase):
def test_portfolio_spend_is_unknown_rather_than_a_partial_sum(self):
section = cost_section([resource(), resource("resource:b", cost={"price_evidence": None, "billing_model": "unknown"})])
self.assertIsNone(section["known_monthly_spend_eur"])
self.assertEqual(1, len(section["unpriced"]))
self.assertIn("not computable", section["spend_note"])
def test_unattributed_resources_are_named_not_spread(self):
record = resource("resource:shared", ownership={"owner": "o", "allocation": {"mode": "unattributed"}})
section = cost_section([record])
self.assertEqual(["resource:shared"], [r["resource_id"] for r in section["unattributed_allocation"]])
class RenewalTest(unittest.TestCase):
def test_dates_inside_the_horizon_are_reported_with_days_remaining(self):
record = resource(lifecycle={"renews_on": "2026-09-01", "cancel_by": None})
section = renewals_section([record], TODAY)
self.assertEqual(18, section["approaching"][0]["days_remaining"])
def test_dates_beyond_the_horizon_are_not_reported(self):
record = resource(lifecycle={"renews_on": "2027-09-01", "cancel_by": None})
self.assertEqual([], renewals_section([record], TODAY)["approaching"])
def test_active_resource_without_dates_is_a_named_gap(self):
section = renewals_section([resource()], TODAY)
self.assertEqual("resource:test", section["undated"][0]["resource_id"])
def test_retired_resource_without_dates_is_not_a_gap(self):
self.assertEqual([], renewals_section([resource(status="retired")], TODAY)["undated"])
class RiskTest(unittest.TestCase):
def test_shared_failure_domain_is_flagged_above_two_resources(self):
records = [resource(f"resource:{i}") for i in range(3)]
risks = risk_section(records, utilization_section(records), cost_section(records))
kinds = {risk["kind"] for risk in risks}
self.assertIn("concentrated_failure_domain", kinds)
def test_two_resources_sharing_a_domain_is_not_yet_a_concentration_risk(self):
records = [resource(f"resource:{i}") for i in range(2)]
risks = risk_section(records, utilization_section(records), cost_section(records))
self.assertNotIn("concentrated_failure_domain", {risk["kind"] for risk in risks})
class LifecycleTest(unittest.TestCase):
def test_states_are_counted_and_unowned_resources_named(self):
records = [resource(), resource("resource:b", status="proposed",
ownership={"owner": None, "allocation": {"mode": "dedicated"}})]
section = lifecycle_section(records)
self.assertEqual({"active": 1, "proposed": 1}, section["by_status"])
self.assertEqual(["resource:b"], section["without_owner"])
class LivePortfolioTest(unittest.TestCase):
"""The report must render from the real committed portfolio, not fixtures."""
@classmethod
def setUpClass(cls):
cls.report = build(ROOT, TODAY)
def test_report_covers_every_registered_resource(self):
self.assertEqual(7, self.report["resource_count"])
self.assertEqual([], self.report["lifecycle"]["without_owner"])
def test_every_service_group_from_discovery_is_present(self):
groups = {row["group"] for row in self.report["coverage"]["groups"]}
self.assertEqual(
{"helix-forge", "coulomb-social", "shared-railiance", "representative-tenant"}, groups
)
def test_spend_is_reported_unknown_while_no_booked_cost_exists(self):
self.assertIsNone(self.report["cost"]["known_monthly_spend_eur"])
self.assertTrue(self.report["cost"]["unpriced"])
def test_host_concentration_is_surfaced_as_a_risk(self):
concentrations = [r for r in self.report["risks"] if r["kind"] == "concentrated_failure_domain"]
self.assertIn("host:railiance01", " ".join(r["detail"] for r in concentrations))
def test_idle_cluster_capacity_is_surfaced(self):
self.assertIn("resource:railiance:reef-railiance:k3s", self.report["utilization"]["idle"])
def test_missing_contract_dates_are_surfaced_rather_than_read_as_no_commitments(self):
self.assertTrue(self.report["renewals"]["undated"])
def test_open_optimization_cases_are_listed_with_their_blockers(self):
self.assertEqual(2, len(self.report["optimization"]["cases"]))
self.assertEqual(2, len(self.report["optimization"]["undecided"]))
self.assertTrue(self.report["optimization"]["blocked_on_evidence"])
def test_next_actions_name_an_owning_repository_for_each_gap(self):
self.assertTrue(self.report["next_actions"])
for action in self.report["next_actions"]:
self.assertIn(":", action)
def test_report_is_deterministic_for_a_fixed_date(self):
self.assertEqual(self.report, build(ROOT, TODAY))
if __name__ == "__main__":
unittest.main()

25
tests/test_variance.py Normal file
View file

@ -0,0 +1,25 @@
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from variance import compare
class VarianceTest(unittest.TestCase):
def test_reports_signed_error_and_absolute_percentage_error(self):
base = {"period":"2026-09","database_gb":5,"stored_gb":100,"wal_gb":10,"restore_egress_gb":5,"write_requests":100,"read_requests":50,"infrastructure_eur":10,"internal_labor_hours":1,"internal_labor_eur":60,"total_eur":70}
actual_row = dict(base, stored_gb=120, infrastructure_eur=12, total_eur=72)
result = compare({"created_at":"2026-08-10T00:00:00Z","rows":[base]}, {"provider_id":"test","rows":[actual_row]})
stored = result["rows"][0]["metrics"]["stored_gb"]
self.assertEqual(20, stored["error"])
self.assertEqual(20, stored["absolute_percentage_error"])
def test_zero_forecast_has_no_percentage_error(self):
base = {"period":"2026-09","database_gb":0,"stored_gb":0,"wal_gb":0,"restore_egress_gb":0,"write_requests":0,"read_requests":0,"infrastructure_eur":0,"internal_labor_hours":0,"internal_labor_eur":0,"total_eur":0}
result = compare({"created_at":"x","rows":[base]}, {"provider_id":"test","rows":[dict(base, wal_gb=1)]})
self.assertIsNone(result["rows"][0]["metrics"]["wal_gb"]["absolute_percentage_error"])
if __name__ == "__main__":
unittest.main()