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_excludes_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"]) # Host Europe never supplied written terms and was excluded at decision # time rather than left blocking the case indefinitely. self.assertEqual("excluded", 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) class ExcludedOptionTest(unittest.TestCase): """An alternative the deciding authority set aside must not block forever.""" def test_excluded_option_does_not_block_the_case(self): alternative = option("alt", one_time_eur=None, excluded=True, exclusion_reason="no written terms within the decision window") result = compare_option(option("baseline"), alternative) self.assertEqual("excluded", result["verdict"]) self.assertEqual([], result["blocking_evidence"]) def test_exclusion_reason_travels_with_the_verdict(self): alternative = option("alt", excluded=True, exclusion_reason="supplier never replied") self.assertEqual("supplier never replied", compare_option(option("baseline"), alternative)["exclusion_reason"]) def test_excluding_without_a_reason_is_rejected(self): record = case(alternatives=[option("alt", one_time_eur=None, excluded=True)]) with self.assertRaisesRegex(ValueError, "exclusion_reason"): validate_case(record) def test_case_with_only_excluded_alternatives_can_be_approved(self): record = case(alternatives=[option("alt", one_time_eur=None, excluded=True, exclusion_reason="unevaluable")]) record["decision"].update({"state": "approved", "recommended_option_id": "baseline", "approver": "founder", "approved_on": "2026-08-14"}) validate_case(record) def test_excluded_option_is_never_recommended(self): record = case(alternatives=[option("alt", recurring_infrastructure_eur_month=0, excluded=True, exclusion_reason="unevaluable")]) record["decision"].update({"state": "approved", "recommended_option_id": "baseline", "approver": "founder", "approved_on": "2026-08-14"}) self.assertIsNone(evaluate(record)["best_option_id"]) class ApprovedStorageCaseTest(unittest.TestCase): """The backup decision was made and executed on 2026-08-14.""" def setUp(self): path = CASE_DIR / "platform-audit-storage-2026-08.json" self.case = json.loads(path.read_text()) def test_case_is_approved_by_a_named_authority(self): decision = self.case["decision"] self.assertEqual("approved", decision["state"]) self.assertEqual("scaleway-standard-multi-az", decision["recommended_option_id"]) self.assertEqual("2026-08-14", decision["approved_on"]) self.assertTrue(decision["approver"]) def test_host_europe_is_excluded_with_a_reason_not_silently_dropped(self): host_europe = next(a for a in self.case["alternatives"] if a["option_id"] == "host-europe-cloud-storage") self.assertTrue(host_europe["excluded"]) self.assertIn("RESOURCE-WP-0002-T02", host_europe["exclusion_reason"]) self.assertTrue(host_europe["unknowns"], "the unknowns stay recorded after exclusion") def test_hetzner_stays_a_costed_rejection(self): verdicts = {c["option_id"]: c["verdict"] for c in evaluate(self.case)["comparisons"]} self.assertEqual("reject", verdicts["hetzner-object-storage"]) self.assertEqual("excluded", verdicts["host-europe-cloud-storage"]) def test_outcome_points_at_the_real_actuals(self): self.assertIn("data/actuals/2026-08.json", self.case["outcome"]["actual_refs"]) self.assertTrue(self.case["financial_handoff"]["sent"]) if __name__ == "__main__": unittest.main()