siemens_ragas/tests/test_question_history.py

"""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") == []