581 lines
11 KiB
Python
581 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import evaluation.evaluate_rag_answers as evaluator
|
|
|
|
|
|
def sample_question(
|
|
question_id: str = "q0001",
|
|
*,
|
|
question: str = (
|
|
"Aká je téma práce?"
|
|
),
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": question_id,
|
|
"split": "dev",
|
|
"category": "specific_fact",
|
|
"difficulty": "medium",
|
|
"question": question,
|
|
"expected_documents": [
|
|
"pages/test/README.md"
|
|
],
|
|
"expected_source_urls": [
|
|
"https://example.test/source"
|
|
],
|
|
"expected_answer_contains": [
|
|
"test"
|
|
],
|
|
"should_answer": True,
|
|
"note": None,
|
|
}
|
|
|
|
|
|
def sample_configuration() -> dict[
|
|
str,
|
|
Any,
|
|
]:
|
|
return {
|
|
"schema_version": (
|
|
evaluator.PARTIAL_SCHEMA_VERSION
|
|
),
|
|
"questions_file": (
|
|
"/tmp/questions.json"
|
|
),
|
|
"overrides_file": (
|
|
"/tmp/overrides.json"
|
|
),
|
|
"questions_fingerprint": (
|
|
"abc123"
|
|
),
|
|
"selected_question_ids": [
|
|
"q0001",
|
|
"q0002",
|
|
],
|
|
"applied_override_ids": [],
|
|
"split": "dev",
|
|
"selected_question_count": 2,
|
|
"requested_model": (
|
|
"model120-fast"
|
|
),
|
|
"operation_id": (
|
|
"retrieve_zpwiki_context"
|
|
),
|
|
"openwebui_url": (
|
|
"https://ui.example/api"
|
|
),
|
|
"rag_url": (
|
|
"http://localhost:8000/rag"
|
|
),
|
|
"timeout": 180,
|
|
"max_attempts": 4,
|
|
"backoff_base": 1.0,
|
|
"backoff_max": 8.0,
|
|
"delay": 0.5,
|
|
}
|
|
|
|
|
|
def test_save_json_results_is_atomic_and_valid(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
path = (
|
|
tmp_path
|
|
/ "result.partial.json"
|
|
)
|
|
|
|
payload = {
|
|
"status": "partial",
|
|
"results": [
|
|
{
|
|
"id": "q0001"
|
|
}
|
|
],
|
|
}
|
|
|
|
evaluator.save_json_results(
|
|
path,
|
|
payload,
|
|
)
|
|
|
|
loaded = json.loads(
|
|
path.read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
|
|
assert (
|
|
loaded
|
|
== payload
|
|
)
|
|
|
|
assert (
|
|
path.read_bytes()
|
|
.endswith(
|
|
b"\n"
|
|
)
|
|
)
|
|
|
|
assert (
|
|
list(
|
|
tmp_path.glob(
|
|
".*.tmp"
|
|
)
|
|
)
|
|
== []
|
|
)
|
|
|
|
|
|
def test_questions_fingerprint_is_stable_and_detects_change() -> None:
|
|
first = [
|
|
sample_question()
|
|
]
|
|
|
|
same = [
|
|
sample_question()
|
|
]
|
|
|
|
changed = [
|
|
sample_question(
|
|
question=(
|
|
"Iná otázka"
|
|
)
|
|
)
|
|
]
|
|
|
|
assert (
|
|
evaluator.questions_fingerprint(
|
|
first
|
|
)
|
|
== evaluator.questions_fingerprint(
|
|
same
|
|
)
|
|
)
|
|
|
|
assert (
|
|
evaluator.questions_fingerprint(
|
|
first
|
|
)
|
|
!= evaluator.questions_fingerprint(
|
|
changed
|
|
)
|
|
)
|
|
|
|
|
|
def test_validate_answer_questions_accepts_valid_question() -> None:
|
|
evaluator.validate_answer_questions(
|
|
[
|
|
sample_question()
|
|
]
|
|
)
|
|
|
|
|
|
def test_validate_answer_questions_rejects_invalid_types() -> None:
|
|
question = (
|
|
sample_question()
|
|
)
|
|
|
|
question[
|
|
"should_answer"
|
|
] = "yes"
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="should_answer",
|
|
):
|
|
evaluator.validate_answer_questions(
|
|
[
|
|
question
|
|
]
|
|
)
|
|
|
|
question = (
|
|
sample_question()
|
|
)
|
|
|
|
question[
|
|
"expected_source_urls"
|
|
] = (
|
|
"https://example.test/source"
|
|
)
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="expected_source_urls",
|
|
):
|
|
evaluator.validate_answer_questions(
|
|
[
|
|
question
|
|
]
|
|
)
|
|
|
|
|
|
def test_validate_resume_compatibility_accepts_matching_configuration() -> None:
|
|
configuration = (
|
|
sample_configuration()
|
|
)
|
|
|
|
evaluator.validate_resume_compatibility(
|
|
configuration,
|
|
dict(
|
|
configuration
|
|
),
|
|
)
|
|
|
|
|
|
def test_validate_resume_compatibility_rejects_model_mismatch() -> None:
|
|
partial_configuration = (
|
|
sample_configuration()
|
|
)
|
|
|
|
current_configuration = (
|
|
sample_configuration()
|
|
)
|
|
|
|
current_configuration[
|
|
"requested_model"
|
|
] = "model2"
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="requested_model",
|
|
):
|
|
evaluator.validate_resume_compatibility(
|
|
partial_configuration,
|
|
current_configuration,
|
|
)
|
|
|
|
|
|
def test_extract_resume_state_skips_success_and_retries_errors() -> None:
|
|
partial_payload = {
|
|
"results": [
|
|
{
|
|
"id": "q0001",
|
|
"answer": "ok",
|
|
"error": None,
|
|
},
|
|
{
|
|
"id": "q0002",
|
|
"answer": "",
|
|
"error": "timeout",
|
|
},
|
|
]
|
|
}
|
|
|
|
(
|
|
successful,
|
|
failed_ids,
|
|
) = evaluator.extract_resume_state(
|
|
partial_payload,
|
|
selected_question_ids={
|
|
"q0001",
|
|
"q0002",
|
|
},
|
|
)
|
|
|
|
assert (
|
|
set(
|
|
successful
|
|
)
|
|
== {
|
|
"q0001"
|
|
}
|
|
)
|
|
|
|
assert (
|
|
failed_ids
|
|
== [
|
|
"q0002"
|
|
]
|
|
)
|
|
|
|
|
|
def test_extract_resume_state_rejects_duplicate_ids() -> None:
|
|
partial_payload = {
|
|
"results": [
|
|
{
|
|
"id": "q0001",
|
|
"error": None,
|
|
},
|
|
{
|
|
"id": "q0001",
|
|
"error": None,
|
|
},
|
|
]
|
|
}
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="duplicitné",
|
|
):
|
|
evaluator.extract_resume_state(
|
|
partial_payload,
|
|
selected_question_ids={
|
|
"q0001"
|
|
},
|
|
)
|
|
|
|
|
|
def test_determine_run_status_distinguishes_states() -> None:
|
|
assert (
|
|
evaluator.determine_run_status(
|
|
[],
|
|
expected_total=2,
|
|
)
|
|
== "partial"
|
|
)
|
|
|
|
assert (
|
|
evaluator.determine_run_status(
|
|
[
|
|
{
|
|
"id": "q0001",
|
|
"error": None,
|
|
}
|
|
],
|
|
expected_total=2,
|
|
)
|
|
== "partial"
|
|
)
|
|
|
|
assert (
|
|
evaluator.determine_run_status(
|
|
[
|
|
{
|
|
"id": "q0001",
|
|
"error": None,
|
|
},
|
|
{
|
|
"id": "q0002",
|
|
"error": "timeout",
|
|
},
|
|
],
|
|
expected_total=2,
|
|
)
|
|
== "complete_with_errors"
|
|
)
|
|
|
|
assert (
|
|
evaluator.determine_run_status(
|
|
[
|
|
{
|
|
"id": "q0001",
|
|
"error": None,
|
|
},
|
|
{
|
|
"id": "q0002",
|
|
"error": None,
|
|
},
|
|
],
|
|
expected_total=2,
|
|
)
|
|
== "complete"
|
|
)
|
|
|
|
|
|
def test_prepare_output_state_requires_resume_or_overwrite_for_partial(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
json_path = (
|
|
tmp_path
|
|
/ "results.json"
|
|
)
|
|
|
|
csv_path = (
|
|
tmp_path
|
|
/ "results.csv"
|
|
)
|
|
|
|
partial_path = (
|
|
tmp_path
|
|
/ "results.partial.json"
|
|
)
|
|
|
|
partial_path.write_text(
|
|
"{}\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(
|
|
FileExistsError,
|
|
match="--resume",
|
|
):
|
|
evaluator.prepare_output_state(
|
|
json_path=(
|
|
json_path
|
|
),
|
|
csv_path=(
|
|
csv_path
|
|
),
|
|
partial_path=(
|
|
partial_path
|
|
),
|
|
resume=False,
|
|
overwrite=False,
|
|
)
|
|
|
|
evaluator.prepare_output_state(
|
|
json_path=(
|
|
json_path
|
|
),
|
|
csv_path=(
|
|
csv_path
|
|
),
|
|
partial_path=(
|
|
partial_path
|
|
),
|
|
resume=False,
|
|
overwrite=True,
|
|
)
|
|
|
|
assert (
|
|
not partial_path.exists()
|
|
)
|
|
|
|
|
|
def test_prepare_output_state_requires_existing_partial_for_resume(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
with pytest.raises(
|
|
FileNotFoundError,
|
|
match="partial neexistuje",
|
|
):
|
|
evaluator.prepare_output_state(
|
|
json_path=(
|
|
tmp_path
|
|
/ "results.json"
|
|
),
|
|
csv_path=(
|
|
tmp_path
|
|
/ "results.csv"
|
|
),
|
|
partial_path=(
|
|
tmp_path
|
|
/ "results.partial.json"
|
|
),
|
|
resume=True,
|
|
overwrite=False,
|
|
)
|
|
|
|
|
|
def test_build_run_configuration_contains_fingerprint(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
questions_path = (
|
|
tmp_path
|
|
/ "questions.json"
|
|
)
|
|
|
|
overrides_path = (
|
|
tmp_path
|
|
/ "overrides.json"
|
|
)
|
|
|
|
questions_path.write_text(
|
|
"[]\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
overrides_path.write_text(
|
|
"{}\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
args = SimpleNamespace(
|
|
questions=(
|
|
questions_path
|
|
),
|
|
overrides=(
|
|
overrides_path
|
|
),
|
|
split="dev",
|
|
model="model120-fast",
|
|
timeout=180,
|
|
max_attempts=4,
|
|
backoff_base=1.0,
|
|
backoff_max=8.0,
|
|
delay=0.5,
|
|
)
|
|
|
|
questions = [
|
|
sample_question()
|
|
]
|
|
|
|
configuration = (
|
|
evaluator.build_run_configuration(
|
|
args=args,
|
|
questions=(
|
|
questions
|
|
),
|
|
selected_override_ids=[],
|
|
operation_id=(
|
|
"retrieve_zpwiki_context"
|
|
),
|
|
)
|
|
)
|
|
|
|
assert (
|
|
configuration[
|
|
"schema_version"
|
|
]
|
|
== evaluator.PARTIAL_SCHEMA_VERSION
|
|
)
|
|
|
|
assert (
|
|
configuration[
|
|
"selected_question_ids"
|
|
]
|
|
== [
|
|
"q0001"
|
|
]
|
|
)
|
|
|
|
assert (
|
|
configuration[
|
|
"questions_fingerprint"
|
|
]
|
|
== evaluator.questions_fingerprint(
|
|
questions
|
|
)
|
|
)
|
|
|
|
|
|
def test_apply_question_overrides_rejects_empty_question(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
questions = [
|
|
sample_question()
|
|
]
|
|
|
|
overrides_path = (
|
|
tmp_path
|
|
/ "overrides.json"
|
|
)
|
|
|
|
overrides_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"q0001": {
|
|
"question": " "
|
|
}
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="override question",
|
|
):
|
|
evaluator.apply_question_overrides(
|
|
questions,
|
|
overrides_path,
|
|
)
|