File size: 2,378 Bytes
35d483e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | 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()
|