tegwick
2c2a6073ff
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>
248 lines
10 KiB
Python
248 lines
10 KiB
Python
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()
|