"""Tests for webapp.services.advisor_comparison: same-scenario run comparison.""" from __future__ import annotations import json from pathlib import Path import pandas as pd from webapp.services.advisor_comparison import build_advisor_comparison, find_previous_run def _write_fake_run( root: Path, run_id: str, scenario_name: str, finished_at: str, judge_model: str, rows: list[dict], ) -> Path: """Write a minimal run directory discoverable by run_reader.list_run_summaries().""" run_dir = root / run_id run_dir.mkdir(parents=True, exist_ok=True) pd.DataFrame(rows).to_csv(run_dir / "scores.csv", index=False) metadata = { "run_id": run_id, "scenario_name": scenario_name, "judge_model": judge_model, "embedding_model": "embed-model", "finished_at": finished_at, "started_at": finished_at, "valid_samples": len(rows), "invalid_samples": 0, } (run_dir / "metadata.json").write_text(json.dumps(metadata), encoding="utf-8") return run_dir class TestFindPreviousRun: def test_finds_most_recent_prior_run_with_same_scenario(self, tmp_path: Path) -> None: _write_fake_run(tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}]) _write_fake_run(tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.6}]) _write_fake_run(tmp_path, "r3", "scn-a", "2026-01-03T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.9}]) previous = find_previous_run( "scn-a", "r3", "2026-01-03T00:00:00+00:00", extra_roots=[tmp_path] ) assert previous is not None assert previous.run_id == "r2" def test_excludes_runs_with_different_scenario_name(self, tmp_path: Path) -> None: _write_fake_run(tmp_path, "r1", "scn-other", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}]) _write_fake_run(tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.6}]) previous = find_previous_run( "scn-a", "r2", "2026-01-02T00:00:00+00:00", extra_roots=[tmp_path] ) assert previous is None def test_returns_none_when_no_history(self, tmp_path: Path) -> None: _write_fake_run(tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}]) previous = find_previous_run( "scn-a", "r1", "2026-01-01T00:00:00+00:00", extra_roots=[tmp_path] ) assert previous is None def test_ignores_runs_at_or_after_current_time(self, tmp_path: Path) -> None: _write_fake_run(tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}]) _write_fake_run(tmp_path, "r2", "scn-a", "2026-01-05T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.6}]) # Current run finished at 2026-01-02, i.e. AFTER r1 but BEFORE r2. previous = find_previous_run( "scn-a", "r-current", "2026-01-02T00:00:00+00:00", extra_roots=[tmp_path] ) assert previous is not None assert previous.run_id == "r1" class TestBuildAdvisorComparison: def test_resolved_status_when_previously_triggered_now_healthy(self, tmp_path: Path) -> None: _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}, {"sample_id": "s2", "faithfulness": 0.5}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.95}, {"sample_id": "s2", "faithfulness": 0.95}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is not None assert len(comparison.entries) == 1 entry = comparison.entries[0] assert entry.metric == "faithfulness" assert entry.status == "resolved" assert entry.previous_score == 0.5 assert entry.current_score is None # not triggered now → no Diagnosis on current side def test_regressed_status_when_previously_healthy_now_triggered(self, tmp_path: Path) -> None: _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.95}, {"sample_id": "s2", "faithfulness": 0.95}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}, {"sample_id": "s2", "faithfulness": 0.5}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is not None assert len(comparison.entries) == 1 entry = comparison.entries[0] assert entry.status == "regressed" assert entry.previous_score is None assert entry.current_score == 0.5 def test_still_triggered_status_shows_score_and_severity_change(self, tmp_path: Path) -> None: _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.55}, {"sample_id": "s2", "faithfulness": 0.55}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.45}, {"sample_id": "s2", "faithfulness": 0.45}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is not None entry = comparison.entries[0] assert entry.status == "still_triggered" assert entry.previous_severity == "warning" assert entry.current_severity == "critical" assert entry.previous_score == 0.55 assert entry.current_score == 0.45 def test_new_metric_status_when_metric_not_measured_before(self, tmp_path: Path) -> None: # Previous run only measured context_recall (healthy); faithfulness wasn't tracked at all. _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "context_recall": 0.95}, {"sample_id": "s2", "context_recall": 0.95}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}, {"sample_id": "s2", "faithfulness": 0.5}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is not None assert len(comparison.entries) == 1 entry = comparison.entries[0] assert entry.metric == "faithfulness" assert entry.status == "new_metric" assert entry.previous_score is None assert entry.current_score == 0.5 def test_metrics_healthy_in_both_are_omitted(self, tmp_path: Path) -> None: _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.95}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.96}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is None # nothing to show → overall None def test_metric_dropped_from_current_scope_is_not_marked_resolved(self, tmp_path: Path) -> None: # Previous run triggered on context_precision, but current run doesn't # evaluate that metric at all — must NOT claim "resolved" without a # fair current-side measurement. _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "context_precision": 0.3}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.95}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is None def test_returns_none_when_no_previous_run(self, tmp_path: Path) -> None: current_dir = _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is None def test_judge_model_changed_flag_set_when_models_differ(self, tmp_path: Path) -> None: _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-4o", [{"sample_id": "s1", "faithfulness": 0.5}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is not None assert comparison.judge_model_changed is True assert comparison.previous_judge_model == "gpt-4o" assert comparison.current_judge_model == "gpt-5" def test_judge_model_changed_false_when_same(self, tmp_path: Path) -> None: _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is not None assert comparison.judge_model_changed is False def test_gracefully_returns_none_on_corrupt_previous_scores_csv(self, tmp_path: Path) -> None: previous_dir = _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}], ) # Corrupt the previous run's scores.csv with invalid-encoding bytes so # pd.read_csv raises UnicodeDecodeError (a ValueError subclass caught # by run_reader.read_scores_frame, which then returns an empty frame). (previous_dir / "scores.csv").write_bytes(b"\xff\xfe\x00\x01broken binary data \x00\x00") current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [{"sample_id": "s1", "faithfulness": 0.5}], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path] ) assert comparison is None def test_worse_statuses_sorted_before_resolved(self, tmp_path: Path) -> None: """regressed/still_triggered/new_metric surface above resolved for visibility.""" _write_fake_run( tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5", [ {"sample_id": "s1", "faithfulness": 0.5, "context_recall": 0.95}, ], ) current_dir = _write_fake_run( tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5", [ {"sample_id": "s1", "faithfulness": 0.95, "context_recall": 0.5}, ], ) comparison = build_advisor_comparison( current_dir, "scn-a", ["faithfulness", "context_recall"], extra_roots=[tmp_path] ) assert comparison is not None statuses = [entry.status for entry in comparison.entries] # context_recall regressed → must appear before faithfulness resolved. assert statuses.index("regressed") < statuses.index("resolved")