| """Check JevBench conversion, reference alignment and frozen source integrity.""" |
|
|
| import copy |
| import json |
| import tempfile |
| import unittest |
| from dataclasses import replace |
| from pathlib import Path |
| from unittest.mock import patch |
|
|
| import httpx |
|
|
| from jev_adapter.benchmarks.data import sha256 |
| from jev_adapter.benchmarks.jevbench import ( |
| JEVBENCH_COMMIT, |
| JEVBENCH_SUITES, |
| JevBenchSpec, |
| normalize_jevbench_record, |
| parse_jevbench, |
| prepare_jevbench, |
| ) |
|
|
|
|
| def fixture(kind="choice", name="sample"): |
| questions = { |
| "choice": { |
| "type": "choice", |
| "instructions": "Choose a queue.", |
| "criteria": {"support": "Support", "billing": "Billing"}, |
| }, |
| "noul": { |
| "type": "noul", |
| "instructions": "Is a refund requested?", |
| "criteria": {"true": "Refund requested", "false": "No refund request"}, |
| }, |
| "score": { |
| "type": "score", |
| "instructions": "How urgent?", |
| "criteria": ["low", "medium", "high"], |
| }, |
| } |
| return { |
| "id": name, |
| "family": "routing", |
| "group": None, |
| "split": "public", |
| "state": {"message": "Please refund my bill."}, |
| "question": questions[kind], |
| "labels": { |
| "choice": ["billing", "support"], |
| "noul": ["no", "yes"], |
| "score": ["0", "1", "2"], |
| }[kind], |
| "expected": {"choice": "billing", "noul": "yes", "score": 2}[kind], |
| "provenance": { |
| "source": "Unit test fixture", |
| "license": "MIT", |
| "rationale": "SECRET_RATIONALE", |
| "surface_answer": "SECRET_SURFACE_ANSWER", |
| "label_basis": "Authored rubric", |
| }, |
| } |
|
|
|
|
| def source_fixture(records): |
| payload = "".join(json.dumps(row) + "\n" for row in records).encode() |
| notices = {"LICENSE": b"Fixture MIT notice", "THIRD-PARTY.md": b"Fixture notice"} |
| spec = JevBenchSpec( |
| "datasets/public/fixture.jsonl", sha256(payload), len(records), "Fixture" |
| ) |
| return ( |
| spec, |
| {spec.path: payload, **notices}, |
| {path: sha256(content) for path, content in notices.items()}, |
| ) |
|
|
|
|
| class TestJevBenchNormalization(unittest.TestCase): |
| def test_native_question_and_canonical_label_orders_preserved_without_gold(self): |
| raw = fixture() |
| raw["provenance"]["gold_probs"] = {"support": 0.25, "billing": 0.75} |
| original = copy.deepcopy(raw) |
| row = normalize_jevbench_record(raw, "jevbench-hard") |
| self.assertEqual(raw, original) |
| self.assertEqual( |
| row["record"], |
| {"state": raw["state"], "questions": {"decision": raw["question"]}}, |
| ) |
| expected = row["expected"]["decision"] |
| self.assertEqual(expected["labels"], ["billing", "support"]) |
| self.assertEqual( |
| list(row["record"]["questions"]["decision"]["criteria"]), |
| ["support", "billing"], |
| ) |
| self.assertEqual(expected["target"], [1, 0]) |
| self.assertEqual(expected["reference_probs"], [0.75, 0.25]) |
| request_text = json.dumps(row["record"]) |
| for secret in ( |
| "SECRET_", |
| "gold_probs", |
| "expected", |
| "provenance", |
| "label_basis", |
| ): |
| self.assertNotIn(secret, request_text) |
| self.assertEqual(row["metadata"]["provenance"], raw["provenance"]) |
| self.assertEqual(row["metadata"]["group_id"], raw["id"]) |
| self.assertEqual(row["split"], "public") |
|
|
| def test_noul_reference_mapping_and_score_hard_label(self): |
| raw = fixture("noul") |
| raw["provenance"]["gold_probs"] = {"yes": 0.8, "no": 0.2} |
| noul = normalize_jevbench_record(raw, "jevbench-hard")["expected"]["decision"] |
| self.assertEqual(noul["labels"], ["false", "true"]) |
| self.assertEqual(noul["target"], [0, 1]) |
| self.assertEqual(noul["reference_probs"], [0.2, 0.8]) |
| score = normalize_jevbench_record(fixture("score"), "jevbench-original") |
| self.assertEqual(score["expected"]["decision"]["label"], 2) |
| self.assertNotIn("reference_probs", score["expected"]["decision"]) |
|
|
| def test_paraphrase_group_preserved_and_null_groups_independent(self): |
| first, second = fixture(name="one"), fixture(name="two") |
| a = normalize_jevbench_record(first, "jevbench-easy") |
| b = normalize_jevbench_record(second, "jevbench-easy") |
| self.assertNotEqual(a["metadata"]["group_id"], b["metadata"]["group_id"]) |
| first["group"] = second["group"] = "pair" |
| self.assertEqual( |
| normalize_jevbench_record(first, "jevbench-original")["metadata"][ |
| "group_id" |
| ], |
| normalize_jevbench_record(second, "jevbench-original")["metadata"][ |
| "group_id" |
| ], |
| ) |
|
|
| def test_bad_labels_private_rows_and_malformed_references_rejected(self): |
| changes = [ |
| {"split": "private"}, |
| {"labels": ["billing", "absent"]}, |
| {"expected": "absent"}, |
| {"provenance": {"exclude_reason": "ambiguous"}}, |
| ] |
| for change in changes: |
| with self.subTest(change=change), self.assertRaises(ValueError): |
| normalize_jevbench_record({**fixture(), **change}, "jevbench-hard") |
| for probs in ( |
| {"billing": 0.5}, |
| {"billing": 0.8, "support": 0.3}, |
| {"billing": float("nan"), "support": 0.3}, |
| {"billing": True, "support": 0.0}, |
| ): |
| raw = fixture() |
| raw["provenance"]["gold_probs"] = probs |
| with self.subTest(probs=probs), self.assertRaises(ValueError): |
| normalize_jevbench_record(raw, "jevbench-hard") |
| for kind, gold in (("score", True), ("score", 3), ("noul", True)): |
| raw = fixture(kind) |
| raw["expected"] = gold |
| with self.subTest(kind=kind, gold=gold), self.assertRaises(ValueError): |
| normalize_jevbench_record(raw, "jevbench-hard") |
|
|
| def test_duplicate_and_blank_rows_rejected(self): |
| row = json.dumps(fixture()) + "\n" |
| with self.assertRaisesRegex(ValueError, "duplicate"): |
| parse_jevbench((row * 2).encode(), "jevbench-original") |
| with self.assertRaisesRegex(ValueError, "blank"): |
| parse_jevbench((row + "\n").encode(), "jevbench-original") |
|
|
|
|
| class TestJevBenchPreparation(unittest.TestCase): |
| def test_pinned_download_preserves_notices_and_is_idempotent(self): |
| spec, sources, notices = source_fixture( |
| [fixture(name="one"), fixture(name="two")] |
| ) |
| requests = [] |
|
|
| def transport(request): |
| requests.append(request.url.path) |
| prefix = f"/fstandhartinger/jevbench/{JEVBENCH_COMMIT}/" |
| self.assertTrue(request.url.path.startswith(prefix)) |
| return httpx.Response( |
| 200, content=sources[request.url.path.removeprefix(prefix)] |
| ) |
|
|
| with ( |
| tempfile.TemporaryDirectory() as tmp, |
| patch.dict(JEVBENCH_SUITES, {"jevbench-fixture": spec}), |
| patch("jev_adapter.benchmarks.jevbench.JEVBENCH_NOTICES", notices), |
| httpx.Client(transport=httpx.MockTransport(transport)) as client, |
| ): |
| output = Path(tmp) |
| result = prepare_jevbench( |
| "jevbench-fixture", output, client=client, limit=1 |
| ) |
| self.assertEqual( |
| result, |
| prepare_jevbench("jevbench-fixture", output, client=client, limit=1), |
| ) |
| self.assertEqual(result["full_partition"]["questions"], 2) |
| self.assertEqual(result["selected"]["questions"], 1) |
| self.assertFalse(result["selection"]["is_full_partition"]) |
| directory = output / "jevbench-fixture" |
| self.assertEqual( |
| result["data_sha256"], sha256((directory / "public.jsonl").read_bytes()) |
| ) |
| for name in notices: |
| self.assertEqual((directory / name).read_bytes(), sources[name]) |
| self.assertTrue( |
| all("private" not in path and "train" not in path for path in requests) |
| ) |
| with self.assertRaises(FileExistsError): |
| prepare_jevbench("jevbench-fixture", output, client=client) |
| self.assertEqual( |
| result["data_sha256"], sha256((directory / "public.jsonl").read_bytes()) |
| ) |
|
|
| def test_bad_data_notice_or_count_fails_before_writing(self): |
| spec, sources, notices = source_fixture([fixture()]) |
| for corrupt in (spec.path, "LICENSE", "THIRD-PARTY.md", "count"): |
| with ( |
| self.subTest(corrupt=corrupt), |
| tempfile.TemporaryDirectory() as tmp, |
| patch.dict( |
| JEVBENCH_SUITES, |
| { |
| "jevbench-fixture": replace(spec, records=2) |
| if corrupt == "count" |
| else spec |
| }, |
| ), |
| patch("jev_adapter.benchmarks.jevbench.JEVBENCH_NOTICES", notices), |
| ): |
| root, output = Path(tmp) / "source", Path(tmp) / "out" |
| for name, content in sources.items(): |
| path = root / name |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_bytes(content + (b" " if name == corrupt else b"")) |
| with self.assertRaisesRegex( |
| ValueError, "SHA256 mismatch|count mismatch" |
| ): |
| prepare_jevbench( |
| "jevbench-fixture", output, source_root=root, limit=1 |
| ) |
| self.assertFalse(output.exists()) |
|
|
| def test_changed_notice_is_not_partially_overwritten(self): |
| spec, sources, notices = source_fixture([fixture()]) |
| with ( |
| tempfile.TemporaryDirectory() as tmp, |
| patch.dict(JEVBENCH_SUITES, {"jevbench-fixture": spec}), |
| patch("jev_adapter.benchmarks.jevbench.JEVBENCH_NOTICES", notices), |
| ): |
| root, output = Path(tmp) / "source", Path(tmp) / "out" |
| for name, content in sources.items(): |
| path = root / name |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_bytes(content) |
| directory = output / "jevbench-fixture" |
| directory.mkdir(parents=True) |
| (directory / "LICENSE").write_bytes(b"Existing different license") |
| with self.assertRaises(FileExistsError): |
| prepare_jevbench("jevbench-fixture", output, source_root=root) |
| self.assertFalse((directory / "public.jsonl").exists()) |
| self.assertEqual( |
| (directory / "LICENSE").read_bytes(), b"Existing different license" |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
