#!/usr/bin/env python3 """Tests for the contributor-held-out evaluation harness. Issue #29 asks for the authoritative comparison and release-gate harness: macro-F1, per-class precision/recall, a confusion matrix, calibration error, coverage, and precision-at-coverage, over splits that are contributor-held-out and reproducible, comparing candidate models. Every fixture in this file is synthetic: fictional pseudonymous subject keys and invented prediction scores. No real photo, label, or patient record is read or written. """ from __future__ import annotations import importlib.util import json import subprocess import sys import tempfile from pathlib import Path import unittest ROOT = Path(__file__).resolve().parents[1] HARNESS = ROOT / "scripts" / "calibration_report.py" def load_harness(): spec = importlib.util.spec_from_file_location("calibration_report", HARNESS) assert spec and spec.loader, "scripts/calibration_report.py must exist" module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module harness = load_harness() def record(subject_key: str, truth: int, predicted: int | None, confidence: float, *, split: str | None = None, model: str = "frozen-encoder") -> dict: return { "model": model, "subjectKey": subject_key, "split": split or harness.expected_split(subject_key), "truth": {"bristolType": truth}, "prediction": None if predicted is None else {"bristolType": predicted, "confidence": confidence}, } class ComparisonTests(unittest.TestCase): def models(self): return { "specialist-candidate": [ record("cd00000000000001", 4, 4, 1.0), record("ce00000000000002", 3, 3, 1.0), record("e600000000000001", 4, 4, 1.0), record("e700000000000002", 6, 6, 1.0), ], "vlm-bootstrap": [ record("d000000000000004", 4, 4, 1.0), record("e600000000000003", 4, 2, 1.0), record("e700000000000004", 6, 6, 1.0), ], } def test_measures_each_candidate_on_test_at_a_threshold_chosen_on_validation(self): verdict = harness.compare(self.models()) self.assertEqual([row["model"] for row in verdict["models"]], ["specialist-candidate", "vlm-bootstrap"]) specialist = verdict["models"][0] self.assertEqual(specialist["selectionSplit"], "validation") self.assertEqual(specialist["threshold"], 0.0) self.assertEqual(specialist["evaluatedSplit"], "test") self.assertAlmostEqual(specialist["macroF1"], 1.0, places=6) self.assertTrue(specialist["gate"]["passed"]) self.assertEqual(specialist["gate"]["failed"], []) self.assertEqual(verdict["recommended"], "specialist-candidate") self.assertEqual(verdict["verdict"], "GO") def test_a_candidate_that_is_only_half_right_fails_the_macro_f1_precision_and_calibration_gates(self): vlm = next(row for row in harness.compare(self.models())["models"] if row["model"] == "vlm-bootstrap") self.assertAlmostEqual(vlm["precisionAtCoverage"], 0.5, places=6) self.assertFalse(vlm["gate"]["passed"]) self.assertEqual(vlm["gate"]["failed"], ["macroF1", "precisionAtCoverage", "ece"]) def test_blocks_the_release_when_no_candidate_clears_every_gate(self): models = {"vlm-bootstrap": self.models()["vlm-bootstrap"]} verdict = harness.compare(models) self.assertEqual(verdict["verdict"], "BLOCKED") self.assertIsNone(verdict["recommended"]) self.assertEqual(verdict["gates"], {"macroF1": 0.80, "precisionAtCoverage": 0.90, "ece": 0.05}) def test_gate_limits_are_inclusive_so_a_candidate_exactly_on_the_line_passes(self): records = [ record("cd00000000000001", 4, 4, 0.9), record("e600000000000001", 4, 4, 0.9), record("e700000000000002", 3, 2, 0.9), ] verdict = harness.compare({"frozen-encoder": records}, gates={"macroF1": 0.0, "precisionAtCoverage": 0.5, "ece": 1.0}) row = verdict["models"][0] self.assertAlmostEqual(row["precisionAtCoverage"], 0.5, places=6) self.assertTrue(row["gate"]["passed"], row["gate"]) def test_refuses_a_candidate_with_no_held_out_predictions(self): with self.assertRaisesRegex(harness.EvaluationError, "no test-split predictions"): harness.compare({"frozen-encoder": [record("cd00000000000001", 4, 4, 1.0)]}) class CliTests(unittest.TestCase): def write_predictions(self, directory: Path, name: str, rows: list[dict]) -> Path: path = directory / name path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8") return path def run_cli(self, *args: str) -> subprocess.CompletedProcess: return subprocess.run( [sys.executable, str(HARNESS), *args], cwd=ROOT, capture_output=True, text=True ) def test_reports_a_go_with_the_candidate_table_and_writes_the_json_report(self): with tempfile.TemporaryDirectory() as raw: directory = Path(raw) predictions = self.write_predictions(directory, "predictions.jsonl", [ {**record("cd00000000000001", 4, 4, 1.0), "model": "specialist-candidate"}, {**record("e600000000000001", 4, 4, 1.0), "model": "specialist-candidate"}, {**record("e700000000000002", 6, 6, 1.0), "model": "specialist-candidate"}, ]) output = directory / "report.json" run = self.run_cli("--predictions", str(predictions), "--json", str(output), "--min-coverage", "0.5") self.assertEqual(run.returncode, 0, run.stderr) self.assertIn("verdict: GO", run.stdout) self.assertIn("specialist-candidate", run.stdout) self.assertIn("macroF1", run.stdout) report = json.loads(output.read_text(encoding="utf-8")) self.assertEqual(report["verdict"], "GO") self.assertEqual(report["recommended"], "specialist-candidate") self.assertEqual(report["models"][0]["evaluatedSplit"], "test") self.assertEqual(report["models"][0]["selectionSplit"], "validation") def test_exits_three_when_the_release_gate_blocks(self): with tempfile.TemporaryDirectory() as raw: directory = Path(raw) predictions = self.write_predictions(directory, "predictions.jsonl", [ {**record("cd00000000000001", 4, 3, 1.0), "model": "vlm-bootstrap"}, {**record("e600000000000001", 4, 2, 1.0), "model": "vlm-bootstrap"}, ]) run = self.run_cli("--predictions", str(predictions)) self.assertEqual(run.returncode, 3, run.stderr) self.assertIn("verdict: BLOCKED", run.stdout) def test_refuses_an_out_of_range_label_instead_of_scoring_it(self): with tempfile.TemporaryDirectory() as raw: directory = Path(raw) output = directory / "report.json" bad_truth = self.write_predictions(directory, "truth.jsonl", [ {**record("cd00000000000001", 9, 9, 1.0), "model": "frozen-encoder"}, ]) run = self.run_cli("--predictions", str(bad_truth), "--json", str(output)) self.assertEqual(run.returncode, 2) self.assertIn("outside Bristol 1-7", run.stderr) self.assertFalse(output.exists(), "a refused run must not leave a report behind") bad_prediction = self.write_predictions(directory, "prediction.jsonl", [ {**record("e600000000000001", 4, 9, 1.0), "model": "frozen-encoder"}, ]) run = self.run_cli("--predictions", str(bad_prediction)) self.assertEqual(run.returncode, 2) self.assertIn("predicted bristolType must be 1-7", run.stderr) def test_refuses_a_corpus_that_leaks_a_contributor_across_splits(self): with tempfile.TemporaryDirectory() as raw: directory = Path(raw) predictions = self.write_predictions(directory, "predictions.jsonl", [ {**record("cd00000000000001", 4, 4, 1.0, split="train"), "model": "frozen-encoder"}, {**record("cd00000000000001", 4, 4, 1.0, split="validation"), "model": "frozen-encoder"}, {**record("e600000000000001", 4, 4, 1.0), "model": "frozen-encoder"}, ]) run = self.run_cli("--predictions", str(predictions)) self.assertEqual(run.returncode, 2) self.assertIn("leaks across", run.stderr) class ThresholdSelectionTests(unittest.TestCase): def validation_records(self): # 0xcd–0xe5 bucket to validation under the ingestion rule. Confident answers are right and # the least confident one is wrong, which is what a threshold is supposed to exploit. return [ record("cd00000000000001", 4, 4, 0.90), record("ce00000000000002", 4, 4, 0.80), record("cf00000000000003", 3, 3, 0.60), record("d000000000000004", 7, 4, 0.40), ] def test_picks_the_most_precise_threshold_that_still_clears_the_coverage_floor(self): records = self.validation_records() # At a 0.5 floor, 0.60 and 0.80 are both perfect; the wider cut wins, so 0.60. floored = harness.select_threshold(records, min_coverage=0.5) self.assertEqual(floored["threshold"], 0.6) self.assertEqual(floored["counts"], {"total": 4, "accepted": 3, "abstained": 1}) self.assertAlmostEqual(floored["precisionAtCoverage"], 1.0, places=6) self.assertAlmostEqual(floored["coverage"], 0.75, places=6) self.assertEqual(floored["selectionSplit"], "validation") # A 0.8 floor forbids that cut, so the harness falls back to the widest honest threshold # instead of silently returning a threshold that violates the floor. strict = harness.select_threshold(records, min_coverage=0.8) self.assertEqual(strict["threshold"], 0.0) self.assertAlmostEqual(strict["precisionAtCoverage"], 0.75, places=6) def test_refuses_an_unreachable_coverage_floor(self): with self.assertRaisesRegex(harness.EvaluationError, "coverage floor"): harness.select_threshold(self.validation_records(), min_coverage=1.5) def test_refuses_to_choose_a_threshold_on_test_or_train_data(self): test_only = [record("e700000000000001", 4, 4, 0.9)] with self.assertRaisesRegex(harness.EvaluationError, "validation split only"): harness.select_threshold(test_only, min_coverage=0.5) train_mixed = [record("aa00000000000001", 4, 4, 0.9), record("cd00000000000001", 3, 3, 0.8)] with self.assertRaisesRegex(harness.EvaluationError, "validation split only"): harness.select_threshold(train_mixed, min_coverage=0.5) class ContributorHeldOutTests(unittest.TestCase): def test_refuses_a_split_that_cannot_be_derived_from_the_subject_key(self): # 0xaa = 170, below the train/validation edge, so this subject can only ever be train. records = [record("aa00000000000001", 4, 4, 0.9, split="test")] with self.assertRaisesRegex(harness.EvaluationError, "not reproducible"): harness.assert_contributor_held_out(records) def test_refuses_a_contributor_whose_photos_span_two_splits(self): records = [ record("aa00000000000001", 4, 4, 0.9, split="train"), record("aa00000000000001", 3, 3, 0.8, split="validation"), ] with self.assertRaisesRegex(harness.EvaluationError, "leaks across"): harness.assert_contributor_held_out(records) def test_allows_one_contributor_to_supply_several_photos_in_the_same_split(self): records = [ record("aa00000000000001", 4, 4, 0.9), record("aa00000000000001", 3, 3, 0.8), ] harness.assert_contributor_held_out(records) # several photos, one person, one split: fine class CalibrationTests(unittest.TestCase): def test_calibration_error_weights_each_confidence_bin_by_its_share_of_accepted_suggestions(self): records = [ record("aa00000000000001", 4, 4, 0.95), record("ab00000000000002", 4, 4, 0.90), record("ac00000000000003", 3, 3, 0.55), record("ad00000000000004", 7, 7, 0.70), ] # Bins [0.9,1.0) hold both 0.95 and 0.90: mean confidence 0.925, accuracy 1.0, gap 0.075. self.assertAlmostEqual(harness.evaluate(records, threshold=0.0)["ece"], 0.225, places=6) # Raising the threshold to 0.6 drops the 0.55 sample, so it can no longer pull calibration. self.assertAlmostEqual(harness.evaluate(records, threshold=0.6)["ece"], 0.15, places=6) class AbstentionTests(unittest.TestCase): def test_low_confidence_and_explicit_abstentions_shrink_coverage_but_never_count_as_errors(self): records = [ record("aa00000000000001", 4, 4, 0.90), record("ab00000000000002", 4, 3, 0.55), record("ac00000000000003", 3, 3, 0.40), record("ad00000000000004", 7, None, 0.0), ] report = harness.evaluate(records, threshold=0.6) self.assertEqual(report["counts"], {"total": 4, "accepted": 1, "abstained": 3}) self.assertAlmostEqual(report["coverage"], 0.25, places=6) self.assertAlmostEqual(report["precisionAtCoverage"], 1.0, places=6) # Only the accepted suggestion is scored; the class that abstained away is named, not hidden. self.assertEqual(sorted(report["perClass"]), [4]) self.assertAlmostEqual(report["macroF1"], 1.0, places=6) self.assertEqual(report["excludedClasses"], [3, 7]) self.assertEqual(report["abstentions"], {"low-confidence": 2, "no-suggestion": 1}) def test_a_threshold_above_every_confidence_abstains_on_everything_without_crashing(self): records = [ record("aa00000000000001", 4, 4, 0.30), record("ab00000000000002", 2, 2, 0.50), ] report = harness.evaluate(records, threshold=0.95) self.assertEqual(report["counts"], {"total": 2, "accepted": 0, "abstained": 2}) self.assertAlmostEqual(report["coverage"], 0.0, places=6) self.assertAlmostEqual(report["macroF1"], 0.0, places=6) self.assertEqual(report["perClass"], {}) class MacroF1Tests(unittest.TestCase): def test_scores_each_bristol_class_and_averages_only_the_classes_present_in_truth(self): records = [ record("aa00000000000001", 4, 4, 0.90), record("ab00000000000002", 4, 3, 0.80), record("ac00000000000003", 3, 3, 0.70), record("ad00000000000004", 7, 7, 0.60), ] report = harness.evaluate(records, threshold=0.0) self.assertEqual(sorted(report["perClass"]), [3, 4, 7]) self.assertAlmostEqual(report["perClass"][4]["precision"], 1.0, places=6) self.assertAlmostEqual(report["perClass"][4]["recall"], 0.5, places=6) self.assertAlmostEqual(report["perClass"][4]["f1"], 0.666667, places=6) self.assertAlmostEqual(report["perClass"][3]["precision"], 0.5, places=6) self.assertAlmostEqual(report["perClass"][3]["recall"], 1.0, places=6) self.assertAlmostEqual(report["perClass"][7]["f1"], 1.0, places=6) self.assertAlmostEqual(report["macroF1"], 0.777778, places=6) self.assertEqual(report["confusion"], {"4->4": 1, "4->3": 1, "3->3": 1, "7->7": 1}) if __name__ == "__main__": unittest.main()