| """Check benchmark integrity and prevent labels from entering inference requests.""" |
|
|
| import copy |
| import json |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from unittest.mock import patch |
|
|
| import httpx |
|
|
| from jev_adapter.benchmarks.data import ( |
| SuiteSpec, |
| normalize_record, |
| parse_partition, |
| population_counts, |
| sha256, |
| ) |
| from jev_adapter.benchmarks.prepare import prepare_suite |
|
|
|
|
| def fixture_record(name="example/1"): |
| return { |
| "state": {"message": "Needs a refund", "count": 4}, |
| "questions": { |
| "queue": { |
| "type": "choice", |
| "instructions": "Choose the queue.", |
| "criteria": {"other": None, "billing": "Refunds"}, |
| "label": "billing", |
| "src": "routing", |
| }, |
| "urgent": { |
| "type": "noul", |
| "instructions": "Is the message urgent?", |
| "label": False, |
| "src": "urgency", |
| }, |
| "priority": { |
| "type": "score", |
| "instructions": "Select priority.", |
| "criteria": ["low", "medium", "high"], |
| "label": 2, |
| "src": "priority", |
| }, |
| }, |
| "_meta": { |
| "id": name, |
| "source": "example", |
| "variant": "clean", |
| "group_id": "pair-1", |
| "pair_id": "pair-1", |
| "sibling": "a", |
| }, |
| } |
|
|
|
|
| def fake_source(records): |
| payload = ("".join(json.dumps(r) + "\n" for r in records)).encode() |
| manifest = json.dumps( |
| { |
| "files": { |
| name: { |
| "sha256": sha256(payload), |
| "records": len(records), |
| "questions": sum(len(r["questions"]) for r in records), |
| } |
| for name in ("development.jsonl", "test.jsonl") |
| }, |
| "dataset_revisions": {}, |
| } |
| ).encode() |
| spec = SuiteSpec("evals/example", sha256(manifest), "Test fixture") |
| return spec, manifest, payload |
|
|
|
|
| class TestBenchmarkNormalization(unittest.TestCase): |
| def test_all_question_types_preserve_order_and_hide_gold(self): |
| original = fixture_record() |
| before = copy.deepcopy(original) |
| row = normalize_record(original, "example", "development") |
| self.assertEqual(original, before) |
| self.assertEqual( |
| list(row["record"]["questions"]), ["queue", "urgent", "priority"] |
| ) |
| self.assertNotIn("model", row["record"]) |
| self.assertNotIn("_meta", row["record"]) |
| for q in row["record"]["questions"].values(): |
| self.assertNotIn("label", q) |
| self.assertNotIn("src", q) |
| self.assertEqual(row["expected"]["queue"]["labels"], ["other", "billing"]) |
| self.assertEqual(row["expected"]["queue"]["target"], [0, 1]) |
| self.assertEqual(row["expected"]["urgent"]["labels"], ["false", "true"]) |
| self.assertEqual(row["expected"]["urgent"]["target"], [1, 0]) |
| self.assertEqual(row["expected"]["priority"]["target"], [0, 0, 1]) |
| self.assertIsNone(row["record"]["questions"]["queue"]["criteria"]["other"]) |
| self.assertEqual(row["metadata"]["pair_id"], "pair-1") |
|
|
| def test_large_choice_space_is_not_truncated(self): |
| raw = fixture_record() |
| raw["questions"] = { |
| "intent": { |
| "type": "choice", |
| "instructions": "Which intent?", |
| "criteria": {f"intent_{i}": None for i in range(78)}, |
| "label": "intent_77", |
| } |
| } |
| row = normalize_record(raw, "example", "development") |
| self.assertEqual(len(row["expected"]["intent"]["labels"]), 78) |
| self.assertEqual(row["expected"]["intent"]["label"], 77) |
|
|
| def test_duplicate_ids_and_malformed_targets_fail(self): |
| raw = fixture_record() |
| payload = (json.dumps(raw) + "\n") * 2 |
| with self.assertRaisesRegex(ValueError, "duplicate record"): |
| parse_partition(payload.encode(), "example", "development") |
| for question, label in ( |
| ("queue", "absent"), |
| ("urgent", "false"), |
| ("priority", 9), |
| ): |
| modified = copy.deepcopy(raw) |
| modified["questions"][question]["label"] = label |
| with self.subTest(question=question), self.assertRaises(ValueError): |
| normalize_record(modified, "example", "development") |
|
|
| def test_population_separates_perturbations_and_unknowable(self): |
| raws = [fixture_record(str(i)) for i in range(3)] |
| raws[1]["_meta"]["variant"] = "permuted" |
| raws[2]["_meta"]["source"] = "unknowable" |
| counts = population_counts( |
| [normalize_record(raw, "example", "development") for raw in raws] |
| ) |
| self.assertEqual(counts["records"], 3) |
| self.assertEqual(counts["questions"], 9) |
| self.assertEqual(counts["clean_questions"], 6) |
| self.assertEqual(counts["headline_questions"], 3) |
|
|
|
|
| class TestBenchmarkPreparation(unittest.TestCase): |
| def test_verified_download_subset_and_idempotency(self): |
| spec, manifest, payload = fake_source( |
| [fixture_record("one"), fixture_record("two")] |
| ) |
| fetched = [] |
|
|
| def transport(request): |
| fetched.append(request.url.path) |
| return httpx.Response( |
| 200, |
| content=manifest |
| if request.url.path.endswith("manifest.json") |
| else payload, |
| ) |
|
|
| with ( |
| tempfile.TemporaryDirectory() as tmp, |
| patch.dict("jev_adapter.benchmarks.prepare.SUITES", {"example": spec}), |
| httpx.Client(transport=httpx.MockTransport(transport)) as client, |
| ): |
| out = Path(tmp) |
| result = prepare_suite( |
| "example", "development", out, client=client, limit=1 |
| ) |
| again = prepare_suite("example", "development", out, client=client, limit=1) |
| self.assertEqual(result, again) |
| self.assertFalse(result["selection"]["is_full_partition"]) |
| self.assertEqual(result["full_partition"]["questions"], 6) |
| self.assertEqual(result["selected"]["questions"], 3) |
| self.assertEqual( |
| result["data_sha256"], |
| sha256((out / "example/development.jsonl").read_bytes()), |
| ) |
| self.assertTrue(all("train" not in path for path in fetched)) |
| with self.assertRaises(FileExistsError): |
| prepare_suite("example", "development", out, client=client) |
|
|
| def test_tampered_partition_rejected_without_writing(self): |
| spec, manifest, payload = fake_source([fixture_record()]) |
|
|
| def transport(request): |
| return httpx.Response( |
| 200, |
| content=manifest |
| if request.url.path.endswith("manifest.json") |
| else payload + b" ", |
| ) |
|
|
| with ( |
| tempfile.TemporaryDirectory() as tmp, |
| patch.dict("jev_adapter.benchmarks.prepare.SUITES", {"example": spec}), |
| httpx.Client(transport=httpx.MockTransport(transport)) as client, |
| ): |
| out = Path(tmp) |
| with self.assertRaisesRegex(ValueError, "SHA256 mismatch"): |
| prepare_suite("example", "development", out, client=client) |
| self.assertFalse((out / "example").exists()) |
|
|
| def test_offline_source_has_same_integrity_and_test_gate(self): |
| spec, manifest, payload = fake_source([fixture_record()]) |
| with ( |
| tempfile.TemporaryDirectory() as tmp, |
| patch.dict("jev_adapter.benchmarks.prepare.SUITES", {"example": spec}), |
| ): |
| root = Path(tmp) / "source" |
| directory = root / "evals/example" |
| directory.mkdir(parents=True) |
| (directory / "manifest.json").write_bytes(manifest) |
| (directory / "test.jsonl").write_bytes(payload) |
| out = Path(tmp) / "out" |
| with self.assertRaisesRegex(ValueError, "locked test"): |
| prepare_suite("example", "test", out, source_root=root) |
| result = prepare_suite( |
| "example", "test", out, source_root=root, allow_test=True |
| ) |
| self.assertTrue(result["protocol"]["locked_test"]) |
| (directory / "manifest.json").write_bytes(manifest + b" ") |
| with self.assertRaisesRegex(ValueError, "SHA256 mismatch"): |
| prepare_suite("example", "test", out, source_root=root, allow_test=True) |
|
|
