132 lines
3.7 KiB
Python
132 lines
3.7 KiB
Python
"""Tests for the cross-run question-history index used by the report page."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from webapp.services.question_history import (
|
|
build_question_history_index,
|
|
lookup,
|
|
normalize_question,
|
|
)
|
|
|
|
|
|
def _write_run(
|
|
run_dir: Path,
|
|
*,
|
|
run_id: str,
|
|
scenario_name: str,
|
|
finished_at: str,
|
|
rows: list[tuple[str, float]],
|
|
metric: str = "faithfulness",
|
|
) -> None:
|
|
"""Create a minimal run directory (metadata.json + scores.csv)."""
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
import json
|
|
|
|
(run_dir / "metadata.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"run_id": run_id,
|
|
"scenario_name": scenario_name,
|
|
"finished_at": finished_at,
|
|
"valid_samples": len(rows),
|
|
"invalid_samples": 0,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
lines = [f"sample_id,question,{metric}"]
|
|
for i, (question, score) in enumerate(rows):
|
|
lines.append(f"s{i},{question},{score}")
|
|
(run_dir / "scores.csv").write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
# A question unlikely to collide with anything under the real outputs/ tree.
|
|
_Q = "UNIQTESTQ ball tube lifetime evaluation method 9f3a"
|
|
|
|
|
|
def test_normalize_question_is_case_and_whitespace_insensitive() -> None:
|
|
assert normalize_question(" Hello World ") == normalize_question("hello world")
|
|
|
|
|
|
def test_index_matches_question_across_runs_newest_first(tmp_path: Path) -> None:
|
|
_write_run(
|
|
tmp_path / "runA",
|
|
run_id="runA",
|
|
scenario_name="scnA",
|
|
finished_at="2026-01-01T00:00:00",
|
|
rows=[(_Q, 0.40)],
|
|
)
|
|
_write_run(
|
|
tmp_path / "runB",
|
|
run_id="runB",
|
|
scenario_name="scnB",
|
|
finished_at="2026-02-01T00:00:00",
|
|
rows=[(_Q, 0.80)],
|
|
)
|
|
|
|
index = build_question_history_index(extra_roots=[tmp_path])
|
|
entries = lookup(index, _Q)
|
|
|
|
assert [e["run_id"] for e in entries] == ["runB", "runA"] # newest first
|
|
assert entries[0]["metrics"]["faithfulness"] == 0.80
|
|
assert entries[1]["scenario_name"] == "scnA"
|
|
|
|
|
|
def test_index_excludes_current_run(tmp_path: Path) -> None:
|
|
_write_run(
|
|
tmp_path / "cur",
|
|
run_id="cur",
|
|
scenario_name="scn",
|
|
finished_at="2026-03-01T00:00:00",
|
|
rows=[(_Q, 0.50)],
|
|
)
|
|
_write_run(
|
|
tmp_path / "prev",
|
|
run_id="prev",
|
|
scenario_name="scn",
|
|
finished_at="2026-01-01T00:00:00",
|
|
rows=[(_Q, 0.60)],
|
|
)
|
|
|
|
index = build_question_history_index(exclude_run_id="cur", extra_roots=[tmp_path])
|
|
entries = lookup(index, _Q)
|
|
|
|
assert [e["run_id"] for e in entries] == ["prev"]
|
|
|
|
|
|
def test_index_keeps_latest_occurrence_within_a_run(tmp_path: Path) -> None:
|
|
_write_run(
|
|
tmp_path / "run1",
|
|
run_id="run1",
|
|
scenario_name="scn",
|
|
finished_at="2026-01-01T00:00:00",
|
|
rows=[(_Q, 0.30), (_Q, 0.70)], # same question twice
|
|
)
|
|
|
|
index = build_question_history_index(extra_roots=[tmp_path])
|
|
entries = lookup(index, _Q)
|
|
|
|
assert len(entries) == 1
|
|
assert entries[0]["metrics"]["faithfulness"] == 0.70 # last occurrence wins
|
|
|
|
|
|
def test_lookup_caps_at_limit(tmp_path: Path) -> None:
|
|
for i in range(7):
|
|
_write_run(
|
|
tmp_path / f"r{i}",
|
|
run_id=f"r{i}",
|
|
scenario_name="scn",
|
|
finished_at=f"2026-01-0{i + 1}T00:00:00",
|
|
rows=[(_Q, 0.1 * i)],
|
|
)
|
|
|
|
index = build_question_history_index(extra_roots=[tmp_path])
|
|
assert len(lookup(index, _Q, limit=3)) == 3
|
|
|
|
|
|
def test_lookup_unknown_question_returns_empty(tmp_path: Path) -> None:
|
|
index = build_question_history_index(extra_roots=[tmp_path])
|
|
assert lookup(index, "NO SUCH QUESTION zzz 0000") == []
|