dp-zp-agent/tools/graphrag/tests/test_evaluator_connected_v2.py
2026-10-25 02:06:17 +01:00

227 lines
11 KiB
Python

from __future__ import annotations
import json
import pytest
from tools.graphrag.evaluation.graphrag_evaluator import evaluate_result, load_questions
def academic(person='Alena Nová', year=2024, title='Práca A', identifier='thesis-a', document='a.md', **extra):
return {'fact_type': 'person_thesis', 'person': person,
'thesis_id': identifier, 'thesis_type': 'bachelor_thesis',
'title': title, 'year': year, 'document_path': document,
'source_url': 'https://example.test/' + document, **extra}
def keyword(person='Alena Nová', value='RAG', document='a.md'):
return {'fact_type': 'person_keyword', 'person': person, 'keyword': value,
'document_path': document, 'source_url': 'https://example.test/' + document}
def question(bindings, **graph):
return {'id': 'synthetic-connected', 'split': 'dev', 'question': 'Overenie dôkazu.',
'graph': {'schema_version': 3, 'expected_bindings': bindings, **graph}}
def score(q, facts=(), **result):
result.setdefault('status', 'ok')
return evaluate_result(q, {'facts': list(facts), **result}, latency_seconds=0)
def test_same_person_keyword_and_academic_year_have_connected_proofs():
q = question([{'person': 'Alena Nová', 'academic_kind': 'Thesis', 'year': 2024, 'keyword': 'RAG'}])
row = score(q, [academic(), keyword()], status='ok')
assert row['strict'] and row['bindings_ok'] and row['paths_ok']
assert len(row['path_results']) == 2
assert all(path['witness'] for path in row['path_results'])
def test_disconnected_persons_cannot_satisfy_conjunction():
q = question([{'person': 'Alena Nová', 'year': 2024, 'keyword': 'RAG'}])
row = score(q, [academic(), keyword(person='Boris Iný')])
assert not row['strict'] and not row['bindings_ok']
def test_legacy_annotations_also_reject_disconnected_persons():
q = {'graph': {'schema_version': 2, 'expected_entities': [
{'type': 'person', 'value': 'Alena Nová'}, {'type': 'keyword', 'value': 'RAG'},
{'type': 'year', 'value': 2024}], 'expected_properties': ['Thesis.year'],
'expected_relations': ['AUTHORED', 'HAS_DOCUMENT', 'HAS_KEYWORD']}}
assert not score(q, [academic(), keyword(person='Boris Iný')])['strict']
def test_start_year_does_not_satisfy_academic_year():
q = question([{'person': 'Alena Nová', 'year': 2024}])
row = score(q, [academic(year=2025, start_year=2024)])
assert not row['strict']
def test_legacy_academic_year_entity_is_not_start_year():
q = {'expected_answer_contains': ['2024'], 'graph': {'expected_entities': [
{'type': 'year', 'value': '2024'}], 'expected_properties': ['Thesis.year']}}
row = score(q, [academic(year=2025, start_year=2024)])
assert not row['strict'] and row['entity_score'] == 0
def test_title_and_year_must_belong_to_the_same_work():
q = question([{'person': 'Alena Nová', 'title': 'Práca A', 'year': 2025}])
assert not score(q, [academic(), academic(year=2025, title='Práca B', identifier='thesis-b')])['strict']
def test_conflicting_same_node_properties_do_not_create_synthetic_work():
q = question([{'person': 'Alena Nová', 'title': 'Práca A', 'year': 2025}])
conflicting = academic(year=2025)
conflicting.pop('title')
assert not score(q, [academic(), conflicting])['strict']
def test_thesis_and_project_types_are_not_interchangeable():
q = question([{'person': 'Alena Nová', 'academic_kind': 'Work', 'year': 2024}])
assert not score(q, [academic()])['strict']
def test_project_proof_uses_has_work_not_authored():
q = question([{'person': 'Alena Nová', 'academic_kind': 'Work', 'academic_type': 'diploma_project', 'year': 2024}])
fact = academic()
fact.update(fact_type='person_work', work_id='project-a', work_type='diploma_project')
fact.pop('thesis_id'); fact.pop('thesis_type')
assert score(q, [fact])['strict']
def test_implicit_academic_proof_accepts_either_real_kind():
assert score(question([{'person': 'Alena Nová', 'year': 2024}]), [academic()])['strict']
def test_metadata_can_come_from_same_person_family_when_not_same_document_required():
q = question([{'person': 'Alena Nová', 'year': 2024, 'keyword': 'RAG'}])
assert score(q, [academic(), keyword(document='child.md')])['strict']
def test_same_document_constraint_rejects_family_only_join():
q = question([{'person': 'Alena Nová', 'year': 2024, 'keyword': 'RAG', 'same_document': True}])
assert not score(q, [academic(), keyword(document='child.md')])['strict']
def test_author_and_category_must_share_actual_document():
q = question([{'person': 'Alena Nová', 'category': 'bp2024', 'document_author': 'Autor Správny'}])
category = {'fact_type': 'person_category', 'person': 'Alena Nová', 'category': 'bp2024', 'document_path': 'a.md'}
author = {'fact_type': 'person_document_author', 'person': 'Alena Nová',
'document_author': 'Autor Správny', 'document_path': 'other.md'}
assert not score(q, [category, author])['strict']
author['document_path'] = 'a.md'
assert score(q, [category, author])['strict']
def test_author_category_query_fact_can_carry_same_document_category_proof():
q = question([{'person': 'Alena Nová', 'category': 'bp2024', 'document_author': 'Autor Správny'}])
author = {'fact_type': 'person_document_author', 'person': 'Alena Nová', 'category': 'bp2024',
'document_author': 'Autor Správny', 'document_path': 'a.md'}
assert score(q, [author])['strict']
def test_start_year_needs_direct_person_source_proof():
q = question([{'person': 'Alena Nová', 'start_year': 2020}])
assert not score(q, [academic(start_year=2020)])['strict']
direct = {'fact_type': 'person_start_year', 'person': 'Alena Nová', 'start_year': 2020, 'document_path': 'a.md'}
assert score(q, [direct])['strict']
direct.pop('document_path')
assert not score(q, [direct])['strict']
def test_conflicting_person_start_year_fails_closed():
q = question([{'person': 'Alena Nová', 'start_year': 2020}])
first = {'fact_type': 'person_start_year', 'person': 'Alena Nová', 'start_year': 2020, 'document_path': 'a.md'}
assert not score(q, [first, {**first, 'start_year': 2021}])['strict']
def test_explicit_path_pattern_is_bound_to_expected_subject():
q = question([{'person': 'Alena Nová', 'keyword': 'RAG'}], expected_path_patterns=[{
'binding_index': 0, 'node_kinds': ['Person', 'Document', 'Keyword'],
'relationships': ['HAS_DOCUMENT', 'HAS_KEYWORD']}])
assert not score(q, [keyword(person='Boris Iný')])['strict']
assert score(q, [keyword()])['strict']
def test_incompatible_ordered_path_does_not_forge_relationship():
q = question([{'person': 'Alena Nová', 'academic_kind': 'Work', 'year': 2024}])
path = {'nodes': [{'kind': 'Person', 'name': 'Alena Nová'},
{'kind': 'Work', 'id': 'work', 'year': 2024},
{'kind': 'Document', 'path': 'a.md'}],
'relationships': ['AUTHORED', 'EVIDENCED_BY']}
assert not score(q, paths=[path])['strict']
path['relationships'][0] = 'HAS_WORK'
assert score(q, paths=[path])['strict']
def test_unrecognized_query_entities_are_never_proof():
q = question([{'person': 'Alena Nová', 'year': 2024}])
assert not score(q, resolved_entities=[{'kind': 'Person', 'name': 'Alena Nová', 'year': 2024}])['strict']
def test_no_answer_is_a_success_only_when_no_positive_evidence_returned():
q = question([], expected_no_answer=True)
q['should_answer'] = False
assert score(q, status='no_answer', reason='unknown_person')['strict']
assert not score(q, [academic()], status='no_answer', reason='unknown_person')['strict']
assert not score(q, status='no_answer')['strict']
assert not score(q, status='ok', reason='unknown_person')['strict']
def test_no_answer_requires_expected_ambiguity_when_specified():
q = question([], expected_no_answer=True, expected_status='ambiguous')
assert score(q, status='ambiguous', reason='multiple_people')['strict']
assert not score(q, status='no_answer', reason='unknown_person')['strict']
def test_false_comparison_can_be_no_answer_evidence_without_positive_claim():
q = question([], expected_no_answer=True)
fact = {'fact_type': 'academic_year_comparison', 'shared_years': [], 'same_academic_year': False}
assert score(q, [fact], status='no_answer', reason='no_shared_year')['strict']
def test_empty_positive_contract_does_not_pass_vacuously():
assert not score(question([]), status='ok')['strict']
assert not score(question([{}]), status='ok')['strict']
def test_zero_count_is_a_grounded_answer_not_a_no_answer():
q = question([], expected_count=0, expected_entity_set=[], entity_set_mode='exact', entity_set_field='person')
result = {'aggregation': {'count': 0, 'people': [], 'complete': True}}
assert score(q, status='ok', **result)['strict']
assert not score(q, status='no_answer', reason='no_rows', **result)['strict']
def test_schema3_loader_rejects_unbound_path_pattern(tmp_path):
q = question([{'person': 'Alena Nová', 'keyword': 'RAG'}], expected_path_patterns=[{
'binding_index': 2, 'node_kinds': ['Person', 'Document'], 'relationships': ['HAS_DOCUMENT']}])
path = tmp_path / 'questions.json'; path.write_text(json.dumps([q]))
with pytest.raises(ValueError, match='invalid connected path pattern'):
load_questions(path)
def test_declared_count_and_people_must_match_actual_answer_evidence():
q = question([{'person': 'Alena Nová', 'year': 2024}], expected_count=1,
expected_entity_set=['Alena Nová'], entity_set_mode='exact', entity_set_field='person')
assert score(q, [academic()], aggregation={'count': 1, 'people': ['Alena Nová'], 'complete': True})['strict']
assert not score(q, [academic()], aggregation={'count': 1, 'people': ['Boris Iný'], 'complete': True})['strict']
assert not score(q, [academic()], aggregation={'count': True, 'people': ['Alena Nová'], 'complete': True})['strict']
def test_no_answer_cannot_hide_a_positive_comparison_claim():
q = question([], expected_no_answer=True)
fact = {'fact_type': 'academic_year_comparison', 'shared_years': [], 'same_academic_year': True}
assert not score(q, [fact], status='no_answer', reason='no_shared_year')['strict']
def test_evidence_accepts_one_of_valid_source_alternatives_not_every_duplicate():
q = question([{'person': 'Alena Nová', 'keyword': 'RAG'}], expected_evidence_documents=[{
'binding_index': 0, 'role': 'keyword_document', 'any_of': [
{'document_path': 'a.md', 'source_url': 'https://example.test/a.md'},
{'document_path': 'child.md', 'source_url': 'https://example.test/child.md'}]}])
assert score(q, [keyword(document='a.md')])['strict']
assert score(q, [keyword(document='child.md')])['strict']
assert not score(q, [keyword(document='unrelated.md')])['strict']
fact = keyword(document='a.md'); fact['source_url'] = 'https://wrong.test/'
assert not score(q, [fact])['strict']