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 # deterministic given the seed train = a["split"]["train_indices"] test = a["split"]["eval_indices"] # disjoint and covering assert set(train).isdisjoint(test) assert sorted(train + test) == list(range(len(rows))) assert len(test) == round(40 * 0.15) + round(60 * 0.15) # per-stratum rounding 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"])) # ~half the held-out examples should carry a hallucinated token 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"]