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']