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