99 lines
4.4 KiB
Python
99 lines
4.4 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'
|