dp-zp-agent/test/test_context_policy.py

192 lines
10 KiB
Python

import copy
import json
import pytest
from tools.answer_evaluation.context_policy import citation_check, answer_context_issues, joined_candidates
from tools.answer_evaluation.checked_generation import generate_checked_answer
from tools.answer_evaluation.evaluate_graph_answers import graph_context,grade_answer
@pytest.fixture
def context():
return {'status':'ok','reason':None,'complete':True,'truncated':False,'truncation_reasons':[],
'index_version':{'version':'generation-1'},'aggregation':None,
'sources':[{'source_id':'S1','document_path':'a.md','source_url':'https://example.test/a'}],
'facts':[{'person':'Eva Nová','title':'Vyhľadávanie informácií','year':2026,
'document_path':'a.md','source_url':'https://example.test/a'},
{'person':'Eva Nová','keyword':'RAG','document_path':'a.md','source_url':'https://example.test/a'}],
'answer_candidates':[]}
@pytest.fixture
def answer():
return {'status':'ok','answer':'Eva Nová má prácu Vyhľadávanie informácií z roku 2026. https://example.test/a',
'claims':[{'person':'Eva Nová','title':'Vyhľadávanie informácií','year':2026}],
'source_urls':['https://example.test/a'],'count':None}
def test_graph_context_forwards_exact_contract_without_inventing_version(context):
raw={**context,'constraints':{'mentions_title':True,'mentions_academic_year':False,'years':[],
'mentions_start_year':False},'documents':[]}
out=graph_context(raw)
for key in ['complete','truncated','truncation_reasons','index_version','sources']:
assert out[key]==context[key]
legacy=copy.deepcopy(raw);legacy.pop('index_version')
assert graph_context(legacy)['index_version'] is None
@pytest.mark.parametrize('location',['source_urls','answer'])
def test_unknown_citation_is_rejected_in_list_and_prose(answer,location):
if location=='source_urls':answer['source_urls'].append('https://invented.test/x')
else:answer['answer']+=' https://invented.test/x'
assert not citation_check(answer,{'https://example.test/a'})['citations_ok']
def test_legitimate_markdown_quoted_and_punctuated_urls_are_accepted(answer):
answer['answer']='[Zdroj](https://example.test/a), `https://example.test/a` a „https://example.test/a“. '
assert citation_check(answer,{'https://example.test/a'})['citations_ok']
@pytest.mark.parametrize('link', ['[zdroj](/invented)', 'www.invented.test/a', '[zdroj](mailto:fake@example.test)'])
def test_non_http_and_relative_citations_are_checked(answer, link):
answer['answer'] += ' ' + link
assert not citation_check(answer, {'https://example.test/a'})['citations_ok']
def test_rag_grader_rejects_extra_citation_despite_required_source_present():
from tools.rag.evaluation.rag_metrics import evaluate_answer
question = {'should_answer': True, 'expected_answer_contains': ['SQLite'],
'expected_source_urls': ['https://example.test/a']}
text = 'SQLite. https://example.test/a'
assert evaluate_answer(question, text, tool_called=True,
allowed_source_urls=['https://example.test/a'])['strict_pass']
result = evaluate_answer(question, text + ' https://invented.test', tool_called=True,
allowed_source_urls=['https://example.test/a'])
assert not result['strict_pass'] and result['unsupported_urls'] == ['https://invented.test']
@pytest.mark.parametrize('complete,reasons', [(None, []), (False, []), (False, ['source_limit'])])
def test_rag_prompt_metadata_matches_top_level_contract(monkeypatch, tmp_path, complete, reasons):
from tools.rag import rag_utils
version = {'version': 'test-generation'}
monkeypatch.setattr(rag_utils, 'search_database', lambda *a, **k: {
'results': [], 'engine': 'test', 'strategies': [], 'complete': complete,
'truncated': bool(reasons), 'truncation_reasons': reasons, 'index_version': version})
result = rag_utils.build_rag_context(tmp_path / 'test.sqlite', 'Neznáma osoba')
metadata = json.loads(result['context'].splitlines()[1])
assert metadata == {key: result[key] for key in
['complete', 'truncated', 'truncation_reasons', 'index_version']}
assert metadata['index_version'] == version
def test_known_wiki_url_outside_this_context_is_rejected(context,answer):
answer['source_urls'].append('https://example.test/other-person')
assert 'unsupported_citation' in answer_context_issues(answer,context)
def test_type_of_work_cannot_become_a_keyword(context,answer):
answer['claims'][0]['keyword']='diplomový projekt'
assert 'unsupported_keyword' in answer_context_issues(answer,context)
def test_title_and_year_must_share_one_actual_record(context,answer):
context['facts'].append({'person':'Eva Nová','title':'Iný názov','year':2025})
answer['claims'][0]['year']=2025
assert 'unsupported_academic_binding' in answer_context_issues(answer,context)
def test_person_join_supports_child_document_but_never_another_person(context,answer):
context['facts'][1]['document_path']='child.md'
intent={'mentions_title':True,'mentions_keyword':True}
candidates=joined_candidates(context['facts'],intent)
assert candidates[0]['evidence_documents']==['a.md','child.md']
context['answer_candidates']=candidates
answer['claims'][0]['keyword']='RAG'
assert answer_context_issues(answer,context)==[]
context['facts'][1]['person']='Boris Iný'
assert joined_candidates(context['facts'],intent)==[]
def test_year_and_keyword_join_does_not_require_a_title(context, answer):
context['facts'][0].pop('title')
context['facts'][1]['document_path'] = 'child.md'
candidates = joined_candidates(context['facts'], {'years': [2026], 'mentions_keyword': True})
assert candidates[0]['year'] == 2026 and 'title' not in candidates[0]
assert candidates[0]['evidence_documents'] == ['a.md', 'child.md']
context['answer_candidates'] = candidates
answer.update(status='no_answer', claims=[])
assert 'rejected_supported_person_join' in answer_context_issues(answer, context)
answer.update(status='ok', claims=[{'person': 'Eva Nová', 'year': 2026, 'keyword': 'RAG'}])
assert answer_context_issues(answer, context) == []
def test_known_join_cannot_be_rejected_or_split(context,answer):
context['answer_candidates']=joined_candidates(context['facts'],{'mentions_title':True,'mentions_keyword':True})
answer['status']='no_answer';answer['claims']=[]
assert 'rejected_supported_person_join' in answer_context_issues(answer,context)
answer['status']='ok';answer['claims']=[{'person':'Eva Nová','title':'Vyhľadávanie informácií','year':2026},
{'person':'Eva Nová','keyword':'RAG'}]
assert 'split_or_missing_person_join' in answer_context_issues(answer,context)
@pytest.mark.parametrize('complete,truncated',[(False,False),(True,True),(None,False)])
def test_incomplete_or_unknown_context_cannot_claim_full_list(context,answer,complete,truncated):
context.update(complete=complete,truncated=truncated)
answer['answer']='Toto je úplný zoznam: Eva Nová.'
assert 'unsupported_exhaustive_statement' in answer_context_issues(answer,context)
answer['answer']='Eva Nová. Výpis je neúplný.'
assert answer_context_issues(answer,context)==[]
def test_exact_count_survives_truncated_evidence_with_notice(context,answer):
context.update(complete=False,truncated=True,aggregation={'count':2,'complete':True})
answer.update(count=2,answer='Počet je úplný, dôkazy sú orezané.')
assert answer_context_issues(answer,context)==[]
answer['count']=3
assert 'incorrect_context_count' in answer_context_issues(answer,context)
context['aggregation']['complete']=False
assert 'unsupported_total_count' in answer_context_issues(answer,context)
def response(answer):
return {'choices':[{'message':{'content':answer if isinstance(answer,str) else json.dumps(answer)}}]}
def test_one_correction_for_bad_citation_preserves_both_attempts(context,answer):
bad=copy.deepcopy(answer);bad['source_urls'].append('https://invented.test/x')
replies=iter([response(bad),response(answer)]);messages=[]
def complete(rows):
messages.append(copy.deepcopy(rows));return next(replies)
out=generate_checked_answer('Otázka',context,'Pokyny',complete)
assert out['accepted_for_delivery'] and len(out['attempts'])==2
assert out['attempts'][0]['context_issues']==['unsupported_citation']
assert len(messages[0])==2 and len(messages[1])==4
assert 'expected_bindings' not in json.dumps(messages)
def test_one_total_correction_covers_format_then_context_failure(context,answer):
bad=copy.deepcopy(answer);bad['claims'][0]['year']=1901
replies=iter([response('{bad json'),response(bad)])
out=generate_checked_answer('Otázka',context,'Pokyny',lambda _:next(replies))
assert not out['accepted_for_delivery'] and len(out['attempts'])==2
assert out['attempts'][0]['format_error'] and out['context_rejected']
def test_valid_context_answer_never_retries_based_on_gold(context,answer):
calls=[]
def complete(rows):calls.append(rows);return response(answer)
out=generate_checked_answer('Otázka',context,'Pokyny',complete)
assert out['accepted_for_delivery'] and len(calls)==1
def test_grader_rejects_extra_url_without_rejecting_correct_answer(context,answer):
oracle={'people':{'Eva Nová':{'documents':['a.md'],'academics':[{'year':2026,'title':'Vyhľadávanie informácií'}]}},
'documents':{'a.md':{'source_url':'https://example.test/a','tags':[],'categories':[],'author':None}}}
case={'expected_answer_contains':['Eva Nová','2026'],'graph':{
'expected_bindings':[{'person':'Eva Nová','year':2026}],
'expected_evidence_documents':[{'any_of':[{'source_url':'https://example.test/a'}]}]}}
assert grade_answer(case,answer,oracle,context=context)['strict']
answer['answer']+=' https://invented.test/x'
assert not grade_answer(case,answer,oracle,context=context)['strict']