| from __future__ import annotations |
|
|
| import json |
| import subprocess |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| class FailureReportTest(unittest.TestCase): |
| def test_report_hashes_identifiers_and_ranks_errors(self) -> None: |
| with tempfile.TemporaryDirectory() as directory: |
| root = Path(directory) |
| metrics = root / "metrics.json" |
| predictions = root / "metrics.predictions.jsonl" |
| output = root / "failures.json" |
| metrics.write_text( |
| json.dumps( |
| { |
| "threshold": 0.6, |
| "split": "validation", |
| "development_only": True, |
| "data_scope": "test", |
| } |
| ), |
| encoding="utf-8", |
| ) |
| rows = [ |
| {"record_id": "secret-a", "label": 0, "probability": 0.9}, |
| {"record_id": "secret-b", "label": 1, "probability": 0.1}, |
| {"record_id": "secret-c", "label": 1, "probability": 0.8}, |
| ] |
| predictions.write_text( |
| "".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8" |
| ) |
| completed = subprocess.run( |
| [ |
| sys.executable, |
| str(ROOT / "scripts/analyze_failures.py"), |
| "--metrics", |
| str(metrics), |
| "--output", |
| str(output), |
| "--min-slice-count", |
| "1", |
| ], |
| cwd=ROOT, |
| capture_output=True, |
| text=True, |
| check=False, |
| ) |
| self.assertEqual(completed.returncode, 0, completed.stderr) |
| report = json.loads(output.read_text(encoding="utf-8")) |
| serialized = json.dumps(report) |
| self.assertNotIn("secret-a", serialized) |
| self.assertNotIn("secret-b", serialized) |
| self.assertEqual(report["counts"]["false_interruptions"], 1) |
| self.assertEqual(report["counts"]["missed_ends"], 1) |
| self.assertTrue( |
| report["highest_confidence_false_interruptions"][0]["case_id"].startswith("case_") |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|