| import copy |
| import sys |
| from pathlib import Path |
|
|
| import pytest |
|
|
| REPO_ROOT = Path(__file__).resolve().parents[1] |
| if str(REPO_ROOT) not in sys.path: |
| sys.path.insert(0, str(REPO_ROOT)) |
|
|
| from scripts.make_eval_split import ( |
| build_split, |
| dataset_binding, |
| has_hallucinated, |
| row_split_group, |
| validate_split_summary, |
| ) |
|
|
|
|
| def make_rows(n_pos: int, n_neg: int) -> list[dict]: |
| rows = [ |
| {"id": f"positive-{index}", "prompt": f"positive {index}", "labels": [0, 1, 0]} |
| for index in range(n_pos) |
| ] |
| rows += [ |
| {"id": f"negative-{index}", "prompt": f"negative {index}", "labels": [0, 0]} |
| for index in range(n_neg) |
| ] |
| return rows |
|
|
|
|
| def test_has_hallucinated(): |
| assert has_hallucinated([0, 2, 1]) is True |
| assert has_hallucinated(["G", "H", "U"]) is True |
| assert has_hallucinated([0, 2, 0]) is False |
| assert has_hallucinated(None) is False |
|
|
|
|
| def test_build_split_is_deterministic_and_partitions(): |
| rows = make_rows(40, 60) |
| a = build_split(rows, test_fraction=0.15, seed=42) |
| b = build_split(rows, test_fraction=0.15, seed=42) |
| assert a == b |
| train = a["split"]["train_indices"] |
| test = a["split"]["eval_indices"] |
| |
| assert set(train).isdisjoint(test) |
| assert sorted(train + test) == list(range(len(rows))) |
| assert len(test) == round(40 * 0.15) + round(60 * 0.15) |
| assert a["dataset_binding"] == dataset_binding(rows) |
| assert a["test_fraction"] == 0.15 |
| assert validate_split_summary(a, rows) == a["split"] |
|
|
|
|
| def test_build_split_stratifies_positive_rate(): |
| rows = make_rows(50, 50) |
| split = build_split(rows, test_fraction=0.2, seed=7) |
| test = split["split"]["eval_indices"] |
| positives = sum(1 for i in test if has_hallucinated(rows[i]["labels"])) |
| |
| assert positives == pytest.approx(len(test) / 2, abs=1) |
|
|
|
|
| def test_build_split_can_emit_train_validation_test_partitions(): |
| rows = make_rows(50, 50) |
| result = build_split( |
| rows, |
| test_fraction=0.10, |
| validation_fraction=0.15, |
| train_fraction=0.75, |
| seed=11, |
| ) |
|
|
| train = result["split"]["train_indices"] |
| validation = result["split"]["validation_indices"] |
| test = result["split"]["eval_indices"] |
|
|
| assert set(train).isdisjoint(validation) |
| assert set(train).isdisjoint(test) |
| assert set(validation).isdisjoint(test) |
| assert sorted(train + validation + test) == list(range(len(rows))) |
| assert len(train) == 75 |
| assert len(validation) == 15 |
| assert len(test) == 10 |
| assert result["scheme"] == "grouped_stratified_train_validation_test" |
| for partition in (train, validation, test): |
| positives = sum(1 for i in partition if has_hallucinated(rows[i]["labels"])) |
| assert positives == pytest.approx(len(partition) / 2, abs=1) |
|
|
|
|
| def test_build_split_keeps_duplicate_prompts_in_one_partition(): |
| rows = [ |
| { |
| "id": f"row-{index}", |
| "prompt": f"question {index}", |
| "labels": [0, index % 2], |
| "metadata": {"source": "source_a" if index % 2 else "source_b"}, |
| } |
| for index in range(40) |
| ] |
| rows.append( |
| { |
| "id": "row-duplicate-question-3", |
| "prompt": " QUESTION 3 ", |
| "labels": [0, 1], |
| "metadata": {"source": "source_a"}, |
| } |
| ) |
|
|
| result = build_split( |
| rows, |
| test_fraction=0.10, |
| validation_fraction=0.15, |
| train_fraction=0.75, |
| seed=9, |
| ) |
| partitions = { |
| index: name |
| for name, indices in ( |
| ("train", result["split"]["train_indices"]), |
| ("validation", result["split"]["validation_indices"]), |
| ("test", result["split"]["eval_indices"]), |
| ) |
| for index in indices |
| } |
|
|
| assert partitions[3] == partitions[40] |
| assert result["n_groups"] == 40 |
| assert result["grouped_by"] == "v5_1.split_group or normalized prompt" |
|
|
|
|
| def test_build_split_rejects_bad_fraction(): |
| with pytest.raises(ValueError): |
| build_split(make_rows(5, 5), test_fraction=0.0, seed=1) |
|
|
|
|
| def test_duplicate_prompts_never_cross_train_test(): |
| rows = make_rows(8, 8) |
| rows[0]["prompt"] = "Duplicate question?" |
| rows[1]["prompt"] = " DUPLICATE question? " |
| for index, row in enumerate(rows[2:], 2): |
| row["prompt"] = f"unique-{index}" |
|
|
| split = build_split(rows, test_fraction=0.25, seed=7) |
| train = set(split["split"]["train_indices"]) |
| test = set(split["split"]["eval_indices"]) |
| assert ({0, 1} <= train) or ({0, 1} <= test) |
| assert split["groups"] == 15 |
|
|
|
|
| def test_explicit_v5_1_group_takes_precedence(): |
| first = { |
| "prompt": "one", |
| "metadata": {"extra": {"v5_1": {"split_group": "shared"}}}, |
| } |
| second = { |
| "prompt": "two", |
| "metadata": {"extra": {"v5_1": {"split_group": "shared"}}}, |
| } |
| assert row_split_group(first, 0) == row_split_group(second, 1) |
|
|
|
|
| def test_dataset_binding_rejects_missing_and_duplicate_ids(): |
| with pytest.raises(ValueError, match="non-empty string id"): |
| dataset_binding([{"prompt": "missing"}]) |
| with pytest.raises(ValueError, match="duplicate row id"): |
| dataset_binding([{"id": "same"}, {"id": "same"}]) |
|
|
|
|
| def test_dataset_binding_changes_for_edit_or_reorder(): |
| rows = make_rows(2, 2) |
| original = dataset_binding(rows) |
| edited = [dict(row) for row in rows] |
| edited[1]["labels"] = [1] |
| assert dataset_binding(edited)["ordered_rows_sha256"] != original["ordered_rows_sha256"] |
| assert dataset_binding(list(reversed(rows)))["ordered_ids_sha256"] != original[ |
| "ordered_ids_sha256" |
| ] |
|
|
|
|
| def test_validator_rejects_group_safe_but_non_deterministic_partition(): |
| rows = make_rows(40, 60) |
| summary = build_split(rows, test_fraction=0.15, seed=42) |
| broken = copy.deepcopy(summary) |
| train = broken["split"]["train_indices"] |
| test = broken["split"]["eval_indices"] |
| train[0], test[0] = test[0], train[0] |
|
|
| with pytest.raises(ValueError, match="do not match the deterministic"): |
| validate_split_summary(broken, rows) |
|
|
|
|
| def test_validator_rejects_empty_partition_before_recomputation(): |
| rows = make_rows(5, 5) |
| summary = build_split(rows, test_fraction=0.2, seed=42) |
| broken = copy.deepcopy(summary) |
| broken["split"]["train_indices"] = list(range(len(rows))) |
| broken["split"]["eval_indices"] = [] |
|
|
| with pytest.raises(ValueError, match="must both be non-empty"): |
| validate_split_summary(broken, rows) |
|
|
|
|
| def test_validator_accepts_legacy_fraction_encoded_in_scheme(): |
| rows = make_rows(10, 10) |
| summary = build_split(rows, test_fraction=0.2, seed=42) |
| summary.pop("test_fraction") |
|
|
| assert validate_split_summary(summary, rows) == summary["split"] |
|
|
