test: add scenario eval runner, recorded fixtures, and baseline comparison

eval/run_eval.py

@@ -0,0 +1,98 @@
+"""Command-line entry point for live and baseline evaluation."""
+
+import argparse
+import asyncio
+import os
+from pathlib import Path
+
+from baseline import record_baselines, write_comparison
+from live import LiveLimits, ScenarioResult, run_live_scenarios
+from scenario import SCENARIO_PATH, load_scenarios
+
+EVAL_ROOT = Path(__file__).parent
+
+
+def main() -> int:
+    """Run the selected evaluation arms and return a process status."""
+    arguments = _parse_arguments()
+    scenarios = load_scenarios(arguments.scenarios)
+    exit_code = 0
+    if arguments.base_url is None:
+        raise SystemExit("--base-url is required")
+    limits = LiveLimits(
+        minimum_tracks=arguments.min_tracks,
+        maximum_tracks=arguments.max_tracks,
+        minimum_artists=arguments.min_artists,
+        first_track_budget_ms=arguments.first_track_budget_ms,
+        total_budget_ms=arguments.total_budget_ms,
+    )
+    results = asyncio.run(
+        run_live_scenarios(scenarios, arguments.base_url, arguments.report_dir, limits)
+    )
+    _print_summary(results)
+    if not all(result.passed for result in results):
+        exit_code = 1
+
+    if arguments.baseline:
+        client_id = os.environ.get("SPOTIFY_CLIENT_ID")
+        client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET")
+        if not client_id or not client_secret:
+            print("Spotify client credentials are absent; skipping baseline.")
+        else:
+            baseline_root = arguments.snapshot_dir / "baseline"
+            asyncio.run(record_baselines(scenarios, client_id, client_secret, baseline_root))
+        comparison_path = write_comparison(
+            scenarios,
+            arguments.snapshot_dir / "baseline",
+            arguments.report_dir,
+            arguments.comparison_top_n,
+        )
+        print(f"Comparison: {comparison_path}")
+    return exit_code
+
+
+def _parse_arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base-url")
+    parser.add_argument("--baseline", action="store_true")
+    parser.add_argument("--scenarios", type=Path, default=SCENARIO_PATH)
+    parser.add_argument("--snapshot-dir", type=Path, default=EVAL_ROOT / "snapshots")
+    parser.add_argument("--report-dir", type=Path, default=EVAL_ROOT / "reports")
+    parser.add_argument("--min-tracks", type=int, default=8)
+    parser.add_argument("--max-tracks", type=int, default=15)
+    parser.add_argument("--min-artists", type=int, default=5)
+    parser.add_argument("--first-track-budget-ms", type=int, default=20_000)
+    parser.add_argument("--total-budget-ms", type=int, default=30_000)
+    parser.add_argument("--comparison-top-n", type=int, default=10)
+    arguments = parser.parse_args()
+    if arguments.min_tracks < 0 or arguments.max_tracks < arguments.min_tracks:
+        parser.error("track bounds are invalid")
+    if min(arguments.min_artists, arguments.first_track_budget_ms, arguments.total_budget_ms) < 0:
+        parser.error("artist and latency limits must be non-negative")
+    if arguments.comparison_top_n < 1:
+        parser.error("comparison top N must be positive")
+    return arguments
+
+
+def _print_summary(results: tuple[ScenarioResult, ...]) -> None:
+    print("| Scenario | Status | Tracks | Artists | First track | Total |")
+    print("| --- | --- | ---: | ---: | ---: | ---: |")
+    for result in results:
+        artists = {str(artist).casefold() for event in result.tracks for artist in _artists(event)}
+        first = f"{result.first_track_ms} ms" if result.first_track_ms is not None else "n/a"
+        total = f"{result.total_ms} ms" if result.total_ms is not None else "n/a"
+        status = "PASS" if result.passed else "FAIL"
+        print(
+            f"| {result.scenario.key} | {status} | {len(result.tracks)} | "
+            f"{len(artists)} | {first} | {total} |"
+        )
+
+
+def _artists(event: dict[str, object]) -> list[object]:
+    track = event.get("track")
+    artists = track.get("artists") if isinstance(track, dict) else None
+    return artists if isinstance(artists, list) else []
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())