124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tools.graphrag.evaluation.freeze_graphrag import make_freeze
|
|
from tools.graphrag.evaluation.reproducibility import source_corpus_fingerprint
|
|
|
|
|
|
def freeze_inputs():
|
|
questions = [
|
|
{
|
|
"id": "dev-a",
|
|
"split": "dev",
|
|
"graph": {"schema_version": 3, "source_profiles": ["pages/students/a/README.md"]},
|
|
},
|
|
{
|
|
"id": "test-b",
|
|
"split": "test",
|
|
"graph": {"schema_version": 3, "source_profiles": ["pages/students/b/README.md"]},
|
|
},
|
|
]
|
|
fingerprints = {
|
|
"implementation": {},
|
|
"dataset_sha256": "dataset",
|
|
"inputs": {},
|
|
"source_corpus": {"sha256": "raw-corpus"},
|
|
}
|
|
audit = {
|
|
"dataset_sha256": "dataset",
|
|
"inputs": {},
|
|
"source_corpus": {"sha256": "raw-corpus"},
|
|
"questions": [{"id": q["id"], "source_consistent": True} for q in questions],
|
|
"by_split": {},
|
|
}
|
|
dev = {
|
|
"configuration": {"split": "dev", "limit": 40, "max_entities": 8},
|
|
"questions": [{"id": "dev-a", "error": None}],
|
|
"summary": {"error_count": 0},
|
|
"run_metadata": {"fingerprints": fingerprints, "graph": {"sha256": "graph"}},
|
|
}
|
|
return dev, questions, fingerprints, audit
|
|
|
|
|
|
def test_primary_freeze_verifies_disjoint_profiles():
|
|
args = freeze_inputs()
|
|
manifest = make_freeze(*args)
|
|
assert manifest["partition"]["overlap"] == 0
|
|
assert manifest["partition"]["test_profile_count"] == 1
|
|
|
|
|
|
def test_primary_freeze_rejects_same_profile_across_splits():
|
|
dev, questions, fingerprints, audit = freeze_inputs()
|
|
questions[1]["graph"]["source_profiles"] = questions[0]["graph"]["source_profiles"]
|
|
with pytest.raises(RuntimeError, match="profiles overlap"):
|
|
make_freeze(dev, questions, fingerprints, audit)
|
|
|
|
|
|
def test_primary_freeze_rejects_unvalidated_source_expectation():
|
|
dev, questions, fingerprints, audit = freeze_inputs()
|
|
audit["questions"][1]["source_consistent"] = False
|
|
with pytest.raises(RuntimeError, match="source-inconsistent"):
|
|
make_freeze(dev, questions, fingerprints, audit)
|
|
|
|
|
|
def test_primary_freeze_rejects_stale_raw_markdown_audit():
|
|
dev, questions, fingerprints, audit = freeze_inputs()
|
|
audit["source_corpus"] = {"sha256": "stale"}
|
|
with pytest.raises(RuntimeError, match="original Markdown"):
|
|
make_freeze(dev, questions, fingerprints, audit)
|
|
|
|
|
|
def test_raw_source_fingerprint_detects_unindexed_change(tmp_path: Path, monkeypatch):
|
|
wiki = tmp_path / "wiki"
|
|
doc = wiki / "pages/students/example/README.md"
|
|
doc.parent.mkdir(parents=True)
|
|
doc.write_text("---\ntitle: Example\n---\nOriginal source\n")
|
|
monkeypatch.setenv("ZPWIKI_ROOT", str(wiki))
|
|
original = source_corpus_fingerprint()
|
|
doc.write_text("---\ntitle: Example\n---\nChanged source\n")
|
|
assert original["sha256"] != source_corpus_fingerprint()["sha256"]
|
|
|
|
|
|
def test_freeze_cli_does_not_overwrite_previous_manifest(tmp_path, monkeypatch):
|
|
from tools.graphrag.evaluation.freeze_graphrag import main
|
|
|
|
previous = tmp_path / "frozen_v1.json"
|
|
previous.write_text("historical manifest")
|
|
monkeypatch.setattr(
|
|
"sys.argv",
|
|
["freeze", "--dev-results", "unused", "--audit", "unused", "--output", str(previous)],
|
|
)
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
assert previous.read_text() == "historical manifest"
|
|
|
|
|
|
def test_dev_overwrite_cannot_replace_archived_results(tmp_path):
|
|
from argparse import Namespace
|
|
from tools.graphrag.evaluation.evaluate_graphrag import validate_output_destination
|
|
|
|
directory = tmp_path / "final_v1"
|
|
directory.mkdir()
|
|
previous = directory / "graphrag_dev.json"
|
|
previous.write_text("archived DEV")
|
|
args = Namespace(split="dev", ids="", max_questions=None, overwrite=True, output_dir=directory)
|
|
with pytest.raises(RuntimeError, match="immutable"):
|
|
validate_output_destination(args)
|
|
assert previous.read_text() == "archived DEV"
|
|
|
|
|
|
def test_reproduction_requires_fresh_results_directory(tmp_path):
|
|
from argparse import Namespace
|
|
from tools.graphrag.evaluation.evaluate_graphrag import validate_output_destination
|
|
|
|
previous = tmp_path / "graphrag_test.json"
|
|
previous.write_text("archived TEST")
|
|
args = Namespace(split="test", ids="", max_questions=None, overwrite=False, output_dir=tmp_path)
|
|
with pytest.raises(RuntimeError, match="already exist"):
|
|
validate_output_destination(args)
|
|
assert previous.read_text() == "archived TEST"
|