291 lines
11 KiB
Python
291 lines
11 KiB
Python
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"]
|