import copy import json import pytest from tools.answers.context_policy import citation_check, answer_context_issues, joined_candidates from tools.answers.checked_generation import generate_checked_answer from tools.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"]