resource-control/tests/test_optimization.py

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()