stack/tests/lead/test_gates.py

"""Promotion gates, the failure record, and drift windows."""

from __future__ import annotations

import pytest

from lead.gates import (
    GateInput,
    GateResult,
    events_in_window,
    promotion_gates,
    record_outcome,
)
from lead.ledger.store import open_ledger


def _passing(**overrides: object) -> GateInput:
    base: dict[str, object] = {
        "beats_automl": True,
        "beats_naive": True,
        "train_through": 2026,
        "test_year": 2027,
        "calibration_error": 0.01,
        "calibration_tolerance": 0.05,
        "conformal_coverage": 0.91,
        "coverage_target": 0.90,
        "subgroup_pass": True,
        "feature_claim_ids": ("claim-1",),
        "known_claim_ids": frozenset({"claim-1"}),
    }
    base.update(overrides)
    return GateInput(**base)  # type: ignore[arg-type]


def test_a_candidate_that_clears_every_gate_is_promoted():
    result = promotion_gates(_passing())
    assert result == GateResult(promoted=True, reasons=())


@pytest.mark.parametrize(
    ("overrides", "reason"),
    [
        ({"beats_automl": False}, "baseline"),
        ({"beats_naive": False}, "baseline"),
        ({"test_year": 2026}, "out_of_time"),
        ({"test_year": 2028}, "out_of_time"),
        ({"calibration_error": 0.06}, "calibration"),
        ({"conformal_coverage": 0.89}, "coverage"),
        ({"subgroup_pass": False}, "subgroup"),
        ({"feature_claim_ids": ()}, "claim"),
        ({"feature_claim_ids": ("missing",)}, "claim"),
    ],
)
def test_each_failed_gate_is_one_reason(overrides, reason):
    result = promotion_gates(_passing(**overrides))
    assert result.promoted is False
    assert result.reasons == (reason,)


def test_a_nan_metric_does_not_promote():
    calibrated = promotion_gates(_passing(calibration_error=float("nan")))
    covered = promotion_gates(_passing(conformal_coverage=float("nan")))
    assert calibrated.promoted is False
    assert calibrated.reasons == ("calibration",)
    assert covered.promoted is False
    assert covered.reasons == ("coverage",)


def test_a_discarded_run_is_recorded_once(tmp_path):
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    gate = GateResult(promoted=False, reasons=("baseline",))
    try:
        assert record_outcome(
            con,
            actor="tester",
            target_id="Q.ACR",
            model="glm",
            where_ran="local",
            finished_on="2026-10-31",
            metrics="mae=1",
            gate=gate,
            relation="discarded",
        ) == (True, True)
        assert record_outcome(
            con,
            actor="other",
            target_id="Q.ACR",
            model="glm",
            where_ran="local",
            finished_on="2026-10-31",
            metrics="mae=1",
            gate=gate,
            relation="discarded",
        ) == (False, False)
        runs = con.execute("SELECT count(*) FROM lead_evidence.run").fetchone()[0]
        findings = con.execute("SELECT count(*) FROM lead_evidence.finding").fetchone()[
            0
        ]
        assert runs == 1
        assert findings == 1
    finally:
        con.close()


def test_a_promoted_run_is_recorded(tmp_path):
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        wrote = record_outcome(
            con,
            actor="tester",
            target_id="Q.ACR",
            model="glm",
            where_ran="local",
            finished_on="2026-10-31",
            metrics="mae=1",
            gate=promotion_gates(_passing()),
            relation="promoted",
            claim_id="claim-1",
        )
        assert wrote == (True, True)
    finally:
        con.close()


def test_an_unknown_relation_writes_nothing(tmp_path):
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        with pytest.raises(ValueError):
            record_outcome(
                con,
                actor="tester",
                target_id="Q.ACR",
                model="glm",
                where_ran="local",
                finished_on="2026-10-31",
                metrics="mae=1",
                gate=GateResult(promoted=False, reasons=("baseline",)),
                relation="other",
            )
        assert con.execute("SELECT count(*) FROM lead_evidence.run").fetchone()[0] == 0
        assert (
            con.execute("SELECT count(*) FROM lead_evidence.finding").fetchone()[0] == 0
        )
    finally:
        con.close()


def test_drift_window_includes_both_ends():
    events = (
        {"event_on": "2026-10-01", "citation": "start"},
        {"event_on": "2026-10-31", "citation": "alarm"},
        {"event_on": "2026-09-30", "citation": "before"},
        {"event_on": "2026-11-01", "citation": "after"},
    )
    assert events_in_window(
        window_start="2026-10-01", alarm_on="2026-10-31", events=events
    ) == ("start", "alarm")