resource-control/tests/test_portfolio_report.py

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