project/tests/test_gate.py

"""Failure-oriented checks for the lab's evidence and decision behavior."""
import copy
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

import gate

ROOT = Path(__file__).resolve().parents[1]


class GateTests(unittest.TestCase):
    def setUp(self):
        self.cases = gate.read_json(ROOT / "fixtures/cases.json")[0]
        self.baseline = gate.read_json(ROOT / "fixtures/baseline.json")[0]
        self.regressed = gate.read_json(ROOT / "fixtures/candidate-regressed.json")[0]
        self.repaired = gate.read_json(ROOT / "fixtures/candidate-repaired.json")[0]
        self.policy = gate.read_json(ROOT / "policy.json")[0]

    def report(self, candidate=None):
        return gate.evaluate(self.cases, self.baseline, candidate or self.repaired, self.policy)

    def test_equal_aggregate_score_can_hide_blocking_regression(self):
        report = self.report(self.regressed)
        self.assertEqual(report["baseline"]["metrics"]["accuracy"], report["candidate"]["metrics"]["accuracy"])
        self.assertEqual(report["new_failure_ids"], ["boundary-02"])
        self.assertEqual(report["fixed_case_ids"], ["routine-04"])
        self.assertEqual(report["blocking_reasons"], ["critical_route_failures", "new_case_failures"])
        self.assertEqual(report["decision"], "BLOCK")

    def test_repaired_fixture_passes_without_authorizing_release(self):
        report = self.report()
        self.assertEqual(report["decision"], "PASS")
        self.assertEqual(report["candidate"]["metrics"]["passed"], 12)
        self.assertEqual(report["evidence_kind"], "synthetic_fixture_replay")
        self.assertIs(report["release_authorized"], False)

    def test_critical_failure_blocks_even_when_new_failures_are_allowed(self):
        self.policy["max_new_failures"] = 1
        self.assertEqual(self.report(self.regressed)["blocking_reasons"], ["critical_route_failures"])

    def test_regression_blocks_even_when_critical_failures_are_allowed(self):
        self.policy["max_critical_failures"] = 1
        self.assertEqual(self.report(self.regressed)["blocking_reasons"], ["new_case_failures"])

    def test_slice_floor_can_block_with_acceptable_overall_accuracy(self):
        self.policy.update(max_critical_failures=1, max_new_failures=1)
        self.policy["min_by_slice"]["boundary"] = 0.8
        self.assertEqual(self.report(self.regressed)["blocking_reasons"], ["slice_accuracy:boundary"])

    def test_overall_floor_has_independent_effect(self):
        self.repaired["results"][0]["route"] = "clarify"
        self.repaired["results"][4]["route"] = "answer"
        self.policy["max_new_failures"] = 2
        self.assertEqual(self.report()["blocking_reasons"], ["overall_accuracy"])

    def test_latency_budget_is_inclusive_and_uses_nearest_rank(self):
        self.repaired["results"][-1]["latency_ms"] = 700
        self.assertEqual(self.report()["decision"], "PASS")
        self.repaired["results"][-1]["latency_ms"] = 701
        self.assertEqual(self.report()["blocking_reasons"], ["p95_latency_ms"])
        self.assertEqual(gate.nearest_rank(list(range(1, 21)), .95), 19)
        self.assertEqual(gate.nearest_rank(list(range(1, 13)), .95), 12)

    def test_mean_cost_budget_is_inclusive(self):
        for row in self.repaired["results"]:
            row["cost_microusd"] = 2000
        self.assertEqual(self.report()["decision"], "PASS")
        self.repaired["results"][0]["cost_microusd"] += 1
        self.assertEqual(self.report()["blocking_reasons"], ["mean_cost_microusd"])

    def test_every_case_must_be_present(self):
        self.repaired["results"].pop()
        with self.assertRaisesRegex(gate.InputError, "every case"):
            self.report()

    def test_unknown_case_is_rejected(self):
        self.repaired["results"][0]["case_id"] = "not-a-case"
        with self.assertRaisesRegex(gate.InputError, "every case"):
            self.report()

    def test_duplicate_result_is_rejected(self):
        self.repaired["results"].append(copy.deepcopy(self.repaired["results"][0]))
        with self.assertRaisesRegex(gate.InputError, "Duplicate result"):
            self.report()

    def test_duplicate_case_is_rejected(self):
        self.cases["cases"].append(copy.deepcopy(self.cases["cases"][0]))
        with self.assertRaisesRegex(gate.InputError, "Duplicate case"):
            self.report()

    def test_dataset_mismatch_is_rejected(self):
        self.repaired["dataset_id"] = "a-different-test-set"
        with self.assertRaisesRegex(gate.InputError, "dataset_id"):
            self.report()

    def test_missing_slice_cannot_silently_disappear(self):
        del self.policy["min_by_slice"]["boundary"]
        with self.assertRaisesRegex(gate.InputError, "every dataset slice"):
            self.report()

    def test_invalid_route_is_not_treated_as_a_normal_miss(self):
        self.repaired["results"][0]["route"] = "refund-approved"
        with self.assertRaisesRegex(gate.InputError, "Unknown route"):
            self.report()

    def test_malformed_expected_route_produces_a_clear_input_error(self):
        self.cases["cases"][0]["expected_route"] = ["answer"]
        with self.assertRaisesRegex(gate.InputError, "Unknown expected route"):
            self.report()

    def test_negative_or_boolean_measurements_are_rejected(self):
        for value in (-1, True, 1.5):
            with self.subTest(value=value):
                self.repaired["results"][0]["latency_ms"] = value
                with self.assertRaisesRegex(gate.InputError, "nonnegative integer"):
                    self.report()

    def test_nonfinite_or_boolean_thresholds_are_rejected(self):
        for value in (float("nan"), float("inf"), True, -0.1, 1.1, 10 ** 400):
            with self.subTest(value=value):
                self.policy["min_overall"] = value
                with self.assertRaisesRegex(gate.InputError, "finite number"):
                    self.report()

    def test_schema_boolean_is_not_version_one(self):
        self.cases["schema_version"] = True
        with self.assertRaisesRegex(gate.InputError, "schema_version"):
            self.report()

    def test_model_runs_cannot_masquerade_as_fixture_evidence(self):
        self.repaired["system"]["model"] = "some-real-model"
        with self.assertRaisesRegex(gate.InputError, "fixtures, not model runs"):
            self.report()

    def test_unknown_fields_are_rejected(self):
        self.repaired["results"][0]["latency_seconds"] = 0.2
        with self.assertRaisesRegex(gate.InputError, "fields must be exactly"):
            self.report()

    def test_json_rejects_duplicate_keys_and_nonfinite_constants(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "invalid.json"
            for text in ('{"a": 1, "a": 2}', '{"a": NaN}', '{"a": Infinity}'):
                with self.subTest(text=text):
                    path.write_text(text)
                    with self.assertRaises(gate.InputError):
                        gate.read_json(path)

    def test_order_does_not_change_score_or_case_comparison(self):
        first = self.report(self.regressed)
        self.regressed["results"].reverse()
        self.cases["cases"].reverse()
        self.assertEqual(first, self.report(self.regressed))

    def test_cli_exit_codes_and_fingerprinted_report(self):
        with tempfile.TemporaryDirectory() as folder:
            report_path = Path(folder) / "report.json"
            for fixture, code in (("candidate-regressed", 1), ("candidate-repaired", 0)):
                with self.subTest(fixture=fixture):
                    result = subprocess.run([sys.executable, str(ROOT / "gate.py"), "--candidate",
                        str(ROOT / f"fixtures/{fixture}.json"), "--report", str(report_path)],
                        capture_output=True, text=True, cwd=folder)
                    self.assertEqual(result.returncode, code, result.stderr)
                    report = json.loads(report_path.read_text())
                    self.assertEqual(report["input_sha256"]["candidate"],
                        gate.read_json(ROOT / f"fixtures/{fixture}.json")[1])
                    self.assertIs(report["release_authorized"], False)

    def test_cli_input_error_has_distinct_exit_code(self):
        result = subprocess.run([sys.executable, str(ROOT / "gate.py"), "--candidate",
            str(ROOT / "fixtures/missing.json")], capture_output=True, text=True)
        self.assertEqual(result.returncode, 2)
        self.assertIn("INPUT ERROR", result.stderr)

    def test_cli_refuses_to_overwrite_input(self):
        path = ROOT / "fixtures/candidate-repaired.json"
        before = path.read_bytes()
        result = subprocess.run([sys.executable, str(ROOT / "gate.py"), "--candidate", str(path),
            "--report", str(path)], capture_output=True, text=True)
        self.assertEqual(result.returncode, 2)
        self.assertEqual(path.read_bytes(), before)

    def test_cli_refuses_input_alias_through_hard_link(self):
        with tempfile.TemporaryDirectory() as folder:
            candidate = Path(folder) / "candidate.json"
            candidate.write_bytes((ROOT / "fixtures/candidate-repaired.json").read_bytes())
            report = Path(folder) / "report.json"
            report.hardlink_to(candidate)
            before = candidate.read_bytes()
            result = subprocess.run([sys.executable, str(ROOT / "gate.py"), "--candidate", str(candidate),
                "--report", str(report)], capture_output=True, text=True)
            self.assertEqual(result.returncode, 2)
            self.assertEqual(candidate.read_bytes(), before)

    def test_limitation_is_count_neutral_for_twelve_default_cases(self):
        report = self.report()
        self.assertEqual(report["baseline"]["metrics"]["total"], 12)
        self.assertEqual(report["candidate"]["metrics"]["total"], 12)
        self.assertEqual(report["limitations"][1],
                         "These synthetic cases do not estimate production performance or safety.")
        self.assertIs(report["release_authorized"], False)

    def test_limitation_is_count_neutral_for_four_capstone_holdout_cases(self):
        import opslab
        report = opslab.capstone_evidence(opslab.load())["route"]
        self.assertEqual(report["baseline"]["metrics"]["total"], 4)
        self.assertEqual(report["candidate"]["metrics"]["total"], 4)
        self.assertEqual(report["limitations"][1],
                         "These synthetic cases do not estimate production performance or safety.")
        self.assertIs(report["release_authorized"], False)


if __name__ == "__main__":
    unittest.main()