251 lines
8.7 KiB
Python
251 lines
8.7 KiB
Python
"""Regression tests for weighted webapp report aggregation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from webapp.models import AdvisorComparison, AdvisorComparisonEntry
|
|
from webapp.services import question_history
|
|
from webapp.services import report_builder
|
|
from webapp.services.report_builder import build_report
|
|
from webapp.services.run_reader import _infer_metrics_from_scores, _read_weights_from_snapshot
|
|
|
|
|
|
def _write_run_artifacts(run_dir: Path) -> None:
|
|
"""Create a minimal run directory with weighted scores and a snapshot."""
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "scores.csv").write_text(
|
|
"\n".join(
|
|
[
|
|
"sample_id,doc_name,faithfulness,context_recall,weighted_score,sample_weight",
|
|
"s1,a.pdf,1.0,0.5,0.8333,3.0",
|
|
"s2,b.pdf,0.0,0.5,0.1667,1.0",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(run_dir / "summary.md").write_text("summary", encoding="utf-8")
|
|
(run_dir / "optimization_advice.md").write_text("advice", encoding="utf-8")
|
|
(run_dir / "scenario.snapshot.yaml").write_text(
|
|
"\n".join(
|
|
[
|
|
"metrics:",
|
|
" - faithfulness",
|
|
" - context_recall",
|
|
"metric_weights:",
|
|
" faithfulness: 2.0",
|
|
" context_recall: 1.0",
|
|
"doc_weights:",
|
|
" a.pdf: 3.0",
|
|
" b.pdf: 1.0",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_read_weights_from_snapshot_returns_metric_and_doc_weights(tmp_path: Path) -> None:
|
|
"""Snapshot weight reader returns both weight maps as plain float dicts."""
|
|
run_dir = tmp_path / "run"
|
|
_write_run_artifacts(run_dir)
|
|
|
|
metric_weights, doc_weights = _read_weights_from_snapshot(run_dir)
|
|
|
|
assert metric_weights == {"faithfulness": 2.0, "context_recall": 1.0}
|
|
assert doc_weights == {"a.pdf": 3.0, "b.pdf": 1.0}
|
|
|
|
|
|
def test_build_report_uses_weighted_means_and_exposes_snapshot_weights(tmp_path: Path) -> None:
|
|
"""Report aggregation uses weighted means and surfaces snapshot weights."""
|
|
run_dir = tmp_path / "run"
|
|
_write_run_artifacts(run_dir)
|
|
|
|
report = build_report(run_dir, ["faithfulness", "context_recall"])
|
|
|
|
assert report.metric_means == {
|
|
"faithfulness": pytest.approx(0.75, rel=1e-4),
|
|
"context_recall": pytest.approx(0.5, rel=1e-4),
|
|
}
|
|
# 综合加权得分已暂时禁用
|
|
assert report.weighted_score_mean is None
|
|
assert report.metric_weights == {"faithfulness": 2.0, "context_recall": 1.0}
|
|
assert report.doc_weights == {"a.pdf": 3.0, "b.pdf": 1.0}
|
|
assert report.summary_markdown == "summary"
|
|
assert report.advice_markdown == "advice"
|
|
|
|
|
|
def test_infer_metrics_excludes_weight_columns_without_snapshot(tmp_path: Path) -> None:
|
|
"""Metric inference excludes weighted helper columns from scores.csv."""
|
|
run_dir = tmp_path / "run"
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "scores.csv").write_text(
|
|
"\n".join(
|
|
[
|
|
"sample_id,doc_name,faithfulness,weighted_score,sample_weight",
|
|
"s1,a.pdf,0.8,0.8,2.0",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert _infer_metrics_from_scores(run_dir) == ["faithfulness"]
|
|
|
|
|
|
def test_build_report_ranks_noise_sensitivity_with_lower_values_as_better(tmp_path: Path) -> None:
|
|
"""Lowest-sample review should treat higher noise sensitivity as worse."""
|
|
run_dir = tmp_path / "run"
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "scores.csv").write_text(
|
|
"\n".join(
|
|
[
|
|
"sample_id,question,noise_sensitivity",
|
|
"s-good,q1,0.10",
|
|
"s-warn,q2,0.30",
|
|
"s-bad,q3,0.90",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(run_dir / "summary.md").write_text("summary", encoding="utf-8")
|
|
(run_dir / "optimization_advice.md").write_text("", encoding="utf-8")
|
|
|
|
report = build_report(run_dir, ["noise_sensitivity"])
|
|
|
|
assert [sample.sample_id for sample in report.lowest_samples[:3]] == [
|
|
"s-bad",
|
|
"s-warn",
|
|
"s-good",
|
|
]
|
|
|
|
|
|
def test_lowest_samples_attaches_history_from_index() -> None:
|
|
"""Surfaced samples are annotated with the same question's prior-run scores."""
|
|
frame = pd.DataFrame(
|
|
[
|
|
{"sample_id": "s1", "question": " How LONG is the tube? ", "faithfulness": 0.40},
|
|
{"sample_id": "s2", "question": "unrelated question", "faithfulness": 0.30},
|
|
]
|
|
)
|
|
history_index = {
|
|
question_history.normalize_question("How long is the tube?"): [
|
|
{
|
|
"run_id": "prev",
|
|
"scenario_name": "scn",
|
|
"finished_at": "2026-01-01T00:00:00",
|
|
"metrics": {"faithfulness": 0.90},
|
|
}
|
|
]
|
|
}
|
|
|
|
samples = report_builder._lowest_samples(frame, ["faithfulness"], history_index)
|
|
by_id = {s.sample_id: s for s in samples}
|
|
|
|
assert len(by_id["s1"].history) == 1
|
|
assert by_id["s1"].history[0].run_id == "prev"
|
|
assert by_id["s1"].history[0].metrics["faithfulness"] == 0.90
|
|
assert by_id["s2"].history == [] # no match → no history
|
|
|
|
|
|
def test_build_report_attaches_question_history(tmp_path: Path, monkeypatch) -> None:
|
|
"""build_report wires the question-history index into surfaced samples."""
|
|
run_dir = tmp_path / "run"
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "scores.csv").write_text(
|
|
"\n".join(
|
|
[
|
|
"sample_id,question,faithfulness",
|
|
"s1,How long is the tube?,0.40",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(run_dir / "summary.md").write_text("summary", encoding="utf-8")
|
|
(run_dir / "optimization_advice.md").write_text("", encoding="utf-8")
|
|
|
|
captured: dict = {}
|
|
|
|
def _fake_index(exclude_run_id=None, extra_roots=None):
|
|
captured["exclude_run_id"] = exclude_run_id
|
|
return {
|
|
question_history.normalize_question("How long is the tube?"): [
|
|
{
|
|
"run_id": "older",
|
|
"scenario_name": "scn",
|
|
"finished_at": "2026-01-01T00:00:00",
|
|
"metrics": {"faithfulness": 0.95},
|
|
}
|
|
]
|
|
}
|
|
|
|
monkeypatch.setattr(question_history, "build_question_history_index", _fake_index)
|
|
|
|
report = build_report(run_dir, ["faithfulness"])
|
|
|
|
assert captured["exclude_run_id"] == "run" # current run excluded from history
|
|
assert report.lowest_samples[0].history[0].run_id == "older"
|
|
assert report.lowest_samples[0].history[0].metrics["faithfulness"] == 0.95
|
|
|
|
|
|
def test_build_report_attaches_advisor_comparison(tmp_path: Path, monkeypatch) -> None:
|
|
"""build_report wires advisor_comparison.build_advisor_comparison() into ReportData."""
|
|
run_dir = tmp_path / "run"
|
|
_write_run_artifacts(run_dir)
|
|
(run_dir / "metadata.json").write_text(
|
|
json.dumps({"run_id": "run-1", "scenario_name": "my-scenario"}), encoding="utf-8"
|
|
)
|
|
|
|
fake_comparison = AdvisorComparison(
|
|
previous_run_id="prev-1",
|
|
previous_finished_at="2026-01-01T00:00:00+00:00",
|
|
entries=[
|
|
AdvisorComparisonEntry(
|
|
metric="faithfulness",
|
|
status="resolved",
|
|
previous_score=0.5,
|
|
previous_severity="warning",
|
|
current_score=None,
|
|
current_severity=None,
|
|
)
|
|
],
|
|
)
|
|
|
|
captured_args = {}
|
|
|
|
def _fake_build(run_dir_arg, scenario_name_arg, metrics_arg):
|
|
captured_args["scenario_name"] = scenario_name_arg
|
|
captured_args["metrics"] = metrics_arg
|
|
return fake_comparison
|
|
|
|
monkeypatch.setattr(
|
|
report_builder.advisor_comparison, "build_advisor_comparison", _fake_build
|
|
)
|
|
|
|
report = build_report(run_dir, ["faithfulness", "context_recall"])
|
|
|
|
assert report.advisor_comparison == fake_comparison
|
|
assert captured_args["scenario_name"] == "my-scenario"
|
|
assert captured_args["metrics"] == ["faithfulness", "context_recall"]
|
|
|
|
|
|
def test_build_report_advisor_comparison_none_when_no_previous_run(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
"""build_report leaves advisor_comparison as None when no predecessor exists."""
|
|
run_dir = tmp_path / "run"
|
|
_write_run_artifacts(run_dir)
|
|
(run_dir / "metadata.json").write_text(
|
|
json.dumps({"run_id": "run-1", "scenario_name": "my-scenario"}), encoding="utf-8"
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
report_builder.advisor_comparison, "build_advisor_comparison", lambda *a, **k: None
|
|
)
|
|
|
|
report = build_report(run_dir, ["faithfulness", "context_recall"])
|
|
|
|
assert report.advisor_comparison is None
|