| """Tests for the canonical Sage CLI.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import subprocess |
| import sys |
| from pathlib import Path |
| from unittest.mock import MagicMock |
|
|
| import pytest |
|
|
| from sage.cli import build_parser, main |
| import sage.cli.evaluation_support.boundary as evaluation_boundary |
| import sage.cli.evaluation_support.readiness as evaluation_readiness |
| import sage.cli.evaluation as evaluation_cli |
| from sage.cli.evaluation_support.boundary import ensure_boundary_guardrail_passed |
| from sage.cli.evaluation_support.readiness import ( |
| ensure_boundary_eval_query_bank_ready, |
| ensure_eval_query_bank_ready, |
| ensure_faithfulness_cases_ready, |
| ) |
| from sage.cli.evaluation import ( |
| command_eval_dev, |
| command_eval_boundary, |
| ) |
| from sage.cli.state import ( |
| command_health, |
| command_qdrant_stamp_anchor, |
| command_reset_eval_dev, |
| command_reset_artifacts, |
| command_reset_experiments, |
| command_reset_baseline, |
| ) |
| from sage.cli.shared import ( |
| DEFAULT_DEV_RAGAS_SAMPLES, |
| DEFAULT_DEV_REQUESTS, |
| DEFAULT_DEV_SAMPLES, |
| DEFAULT_REQUESTS, |
| remove_path, |
| ) |
| from sage.data.query_bank.sources.esci._config import ( |
| DEFAULT_RETRIEVAL_FINAL_REPORT_SUBSET_TAG, |
| ) |
| from sage.data.query_bank import build_query_bank_identity, load_query_bank_subset |
|
|
|
|
| def _stub_boundary_corpus_alignment( |
| monkeypatch: pytest.MonkeyPatch, |
| *, |
| fingerprint: str = "corpus-fingerprint", |
| ) -> None: |
| monkeypatch.setattr( |
| evaluation_boundary, |
| "assert_corpus_alignment", |
| lambda: {"corpus_fingerprint": fingerprint}, |
| ) |
|
|
|
|
| def _write_jsonl(path: Path, rows: list[dict]) -> None: |
| with open(path, "w", encoding="utf-8") as f: |
| for row in rows: |
| f.write(json.dumps(row) + "\n") |
|
|
|
|
| def _boundary_artifact_payload( |
| query_bank_path: Path, |
| *, |
| status: str, |
| violations: list[dict] | None = None, |
| sample_limited: bool = False, |
| requested_query_limit: int | None = None, |
| available_query_count: int = 1, |
| evaluated_query_count: int = 1, |
| subset_tag: str = "boundary_eval", |
| artifact_scope: str = "canonical", |
| corpus_fingerprint: str = "corpus-fingerprint", |
| available_query_ids: list[str] | None = None, |
| evaluated_query_ids: list[str] | None = None, |
| ) -> dict: |
| subset_entries = load_query_bank_subset( |
| subset_tag, |
| path=query_bank_path, |
| require_nonempty=True, |
| ) |
| canonical_query_ids = sorted(entry.query_id for entry in subset_entries) |
| if available_query_ids is None: |
| available_query_ids = canonical_query_ids |
| if evaluated_query_ids is None: |
| evaluated_query_ids = canonical_query_ids |
| return { |
| "query_bank_identity": build_query_bank_identity(query_bank_path), |
| "corpus_alignment": { |
| "corpus_fingerprint": corpus_fingerprint, |
| }, |
| "dataset_summary": { |
| "available_query_count": available_query_count, |
| "evaluated_query_count": evaluated_query_count, |
| "requested_query_limit": requested_query_limit, |
| "sample_limited": sample_limited, |
| "full_subset_evaluated": available_query_count == evaluated_query_count, |
| "artifact_scope": artifact_scope, |
| "available_query_ids": available_query_ids, |
| "evaluated_query_ids": evaluated_query_ids, |
| }, |
| "methodology": { |
| "subset_tag": subset_tag, |
| "artifact_scope": artifact_scope, |
| }, |
| "boundary_guardrail": { |
| "status": status, |
| "violations": violations or [], |
| }, |
| } |
|
|
|
|
| def test_top_level_help_lists_canonical_commands(capsys): |
| with pytest.raises(SystemExit) as excinfo: |
| main(["--help"]) |
|
|
| assert excinfo.value.code == 0 |
| output = capsys.readouterr().out |
| assert "python -m sage.cli" in output |
| assert "eval" in output |
| assert "health" in output |
| assert "stage" in output |
| assert "qdrant" in output |
| assert "reset" in output |
| assert "use the CLI directly" in output |
| assert "make ci-fresh" in output |
| assert "stage experiments full" in output |
| assert "--retrieval-decision baseline-retained" in output |
| assert "eval dev" in output |
| assert "reset eval-dev" in output |
| assert "setup" not in output |
| assert "serve-dev" not in output |
|
|
|
|
| def test_eval_run_parser_defaults(): |
| parser = build_parser() |
| args = parser.parse_args(["eval", "run"]) |
|
|
| assert args.command == "eval" |
| assert args.samples is None |
| assert args.ragas_samples is None |
| assert args.requests == DEFAULT_REQUESTS |
|
|
|
|
| def test_eval_run_parser_accepts_all_and_explicit_ragas_limit(): |
| parser = build_parser() |
| args = parser.parse_args( |
| ["eval", "run", "--samples", "all", "--ragas-samples", "25"] |
| ) |
|
|
| assert args.command == "eval" |
| assert args.samples is None |
| assert args.ragas_samples == 25 |
|
|
|
|
| def test_eval_dev_parser_defaults(): |
| parser = build_parser() |
| args = parser.parse_args(["eval", "dev"]) |
|
|
| assert args.command == "eval" |
| assert args.eval_command == "dev" |
| assert args.samples == DEFAULT_DEV_SAMPLES |
| assert args.ragas_samples == DEFAULT_DEV_RAGAS_SAMPLES |
| assert args.requests == DEFAULT_DEV_REQUESTS |
|
|
|
|
| def test_command_eval_dev_delegates_to_eval_with_dev_defaults(monkeypatch): |
| recorded = {} |
|
|
| def fake_run_eval(*, samples, ragas_samples, url, requests, enforce_gate): |
| recorded["samples"] = samples |
| recorded["ragas_samples"] = ragas_samples |
| recorded["url"] = url |
| recorded["requests"] = requests |
| recorded["enforce_gate"] = enforce_gate |
|
|
| monkeypatch.setattr(evaluation_cli, "_run_eval", fake_run_eval) |
|
|
| command_eval_dev( |
| argparse.Namespace( |
| samples=DEFAULT_DEV_SAMPLES, |
| ragas_samples=DEFAULT_DEV_RAGAS_SAMPLES, |
| url="https://example.com", |
| requests=DEFAULT_DEV_REQUESTS, |
| ) |
| ) |
|
|
| assert recorded == { |
| "samples": DEFAULT_DEV_SAMPLES, |
| "ragas_samples": DEFAULT_DEV_RAGAS_SAMPLES, |
| "url": "https://example.com", |
| "requests": DEFAULT_DEV_REQUESTS, |
| "enforce_gate": False, |
| } |
|
|
|
|
| def test_eval_workflow_passes_full_scope_defaults_to_faithfulness_script( |
| monkeypatch, capsys |
| ): |
| recorded_steps = [] |
|
|
| monkeypatch.setattr( |
| evaluation_cli, |
| "run_steps", |
| lambda _title, steps, _footer: recorded_steps.extend(steps), |
| ) |
| monkeypatch.setattr(evaluation_cli, "capture_output", lambda _command: "summary") |
| monkeypatch.setattr( |
| evaluation_boundary, |
| "ensure_boundary_guardrail_passed", |
| lambda: {"status": "pass"}, |
| ) |
| monkeypatch.setattr( |
| evaluation_cli, |
| "build_eval_status", |
| lambda **_kwargs: { |
| "execution_complete": True, |
| "reportable_green": False, |
| "reportable_reasons": ["Load-test artifact is present but not reportable."], |
| "safety_green": True, |
| }, |
| ) |
|
|
| evaluation_cli._run_full_eval_workflow( |
| samples=None, |
| ragas_samples=None, |
| url="https://example.com", |
| requests=25, |
| enforce_gate=False, |
| ) |
|
|
| assert len(recorded_steps) == 8 |
| faithfulness_command = list(recorded_steps[4].commands[0]) |
| assert faithfulness_command[:2] == [sys.executable, "scripts/faithfulness.py"] |
| assert "--samples" in faithfulness_command |
| assert faithfulness_command[faithfulness_command.index("--samples") + 1] == "all" |
| assert "--ragas" in faithfulness_command |
| assert "--ragas-samples" in faithfulness_command |
| assert ( |
| faithfulness_command[faithfulness_command.index("--ragas-samples") + 1] == "all" |
| ) |
| assert recorded_steps[5].title == "[6/8] Boundary behavior" |
| all_commands = [ |
| str(item) |
| for step in recorded_steps |
| for command in step.commands |
| for item in command |
| ] |
| assert "--delta" not in all_commands |
|
|
| output = capsys.readouterr().out |
| assert "grounding_delta_latest.json" not in output |
| assert "adjusted_faithfulness_latest.json" in output |
| assert "execution_complete: True" in output |
| assert "safety_green: True" in output |
| assert "reportable_green: False" in output |
|
|
|
|
| def test_eval_workflow_rejects_incomplete_current_cycle_artifacts(monkeypatch): |
| monkeypatch.setattr( |
| evaluation_cli, |
| "run_steps", |
| lambda _title, _steps, _footer: None, |
| ) |
| monkeypatch.setattr( |
| evaluation_cli, |
| "build_eval_status", |
| lambda **_kwargs: { |
| "execution_complete": False, |
| "execution_reasons": [ |
| "Artifact `load_test_latest.json` was not refreshed during the current evaluation run." |
| ], |
| "reportable_green": False, |
| "reportable_reasons": [], |
| "safety_green": False, |
| }, |
| ) |
| monkeypatch.setattr( |
| evaluation_boundary, |
| "ensure_boundary_guardrail_passed", |
| lambda: pytest.fail( |
| "boundary safety check should not run when execution is incomplete" |
| ), |
| ) |
|
|
| with pytest.raises( |
| SystemExit, match="did not finish with a complete current-cycle artifact set" |
| ): |
| evaluation_cli._run_full_eval_workflow( |
| samples=None, |
| ragas_samples=None, |
| url="https://example.com", |
| requests=25, |
| enforce_gate=True, |
| ) |
|
|
|
|
| def test_health_parser_is_available(): |
| parser = build_parser() |
| args = parser.parse_args(["health"]) |
|
|
| assert args.command == "health" |
|
|
|
|
| def test_health_command_unhealthy_without_qdrant(monkeypatch, capsys): |
| monkeypatch.setattr("sage.cli.state.load_dotenv_if_available", lambda: None) |
| monkeypatch.delenv("QDRANT_URL", raising=False) |
| monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) |
| monkeypatch.delenv("OPENAI_API_KEY", raising=False) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| command_health(argparse.Namespace()) |
|
|
| assert excinfo.value.code == 1 |
| output = capsys.readouterr().out |
| assert "status: unhealthy" in output |
| assert "qdrant_url_configured: False" in output |
|
|
|
|
| def test_health_command_reports_healthy(monkeypatch, capsys): |
| monkeypatch.setattr("sage.cli.state.load_dotenv_if_available", lambda: None) |
| monkeypatch.setenv("QDRANT_URL", "https://example.qdrant.io") |
| monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key") |
|
|
| from sage.adapters import vector_store |
|
|
| monkeypatch.setattr(vector_store, "get_client", lambda: MagicMock()) |
| monkeypatch.setattr( |
| vector_store, |
| "get_collection_info", |
| lambda _client: { |
| "name": "sage_reviews", |
| "points_count": 423165, |
| "status": "green", |
| }, |
| ) |
|
|
| command_health(argparse.Namespace()) |
|
|
| output = capsys.readouterr().out |
| assert "status: healthy" in output |
| assert "qdrant_connected: True" in output |
| assert "llm_credentials_configured: True" in output |
| assert "name: sage_reviews" in output |
|
|
|
|
| def test_eval_summary_parser_is_available(): |
| parser = build_parser() |
| args = parser.parse_args(["eval", "summary"]) |
|
|
| assert args.command == "eval" |
| assert args.eval_command == "summary" |
|
|
|
|
| def test_reset_eval_dev_parser_supports_dry_run(): |
| parser = build_parser() |
| args = parser.parse_args(["reset", "eval-dev", "--dry-run"]) |
|
|
| assert args.command == "reset" |
| assert args.reset_command == "eval-dev" |
| assert args.dry_run is True |
|
|
|
|
| def test_eval_boundary_parser_is_available(): |
| parser = build_parser() |
| args = parser.parse_args(["eval", "boundary", "--query-limit", "5"]) |
|
|
| assert args.command == "eval" |
| assert args.eval_command == "boundary" |
| assert args.query_limit == 5 |
| assert args.subset_tag == "boundary_eval" |
| assert args.min_rating is None |
|
|
|
|
| def test_eval_boundary_parser_accepts_none_min_rating(): |
| parser = build_parser() |
| args = parser.parse_args(["eval", "boundary", "--min-rating", "none"]) |
|
|
| assert args.command == "eval" |
| assert args.eval_command == "boundary" |
| assert args.min_rating is None |
|
|
|
|
| def test_boundary_guardrail_check_accepts_pass_artifact(monkeypatch, tmp_path: Path): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| query_bank_path, |
| status="pass", |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| guardrail = ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| assert guardrail["status"] == "pass" |
|
|
|
|
| def test_boundary_guardrail_check_rejects_fail_artifact(monkeypatch, tmp_path: Path): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| query_bank_path, |
| status="fail", |
| violations=[ |
| { |
| "metric": "ambiguous_clarify_rate", |
| "message": "Ambiguous queries are not clarifying.", |
| } |
| ], |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| message = str(excinfo.value) |
| assert "Evaluation boundary guardrail did not pass" in message |
| assert "Status: fail" in message |
| assert "ambiguous_clarify_rate" in message |
|
|
|
|
| def test_boundary_guardrail_check_rejects_insufficient_coverage( |
| monkeypatch, tmp_path: Path |
| ): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| query_bank_path, |
| status="insufficient_coverage", |
| violations=[ |
| { |
| "metric": "total_queries", |
| "message": "Boundary benchmark has too few queries.", |
| } |
| ], |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| assert "Status: insufficient_coverage" in str(excinfo.value) |
|
|
|
|
| def test_boundary_guardrail_check_rejects_legacy_artifact(monkeypatch, tmp_path: Path): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text(json.dumps({"summary": {}}), encoding="utf-8") |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| assert "boundary guardrail result is missing" in str(excinfo.value) |
|
|
|
|
| def test_boundary_guardrail_check_rejects_query_limited_artifact( |
| monkeypatch, tmp_path: Path |
| ): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| query_bank_path, |
| status="pass", |
| sample_limited=True, |
| requested_query_limit=10, |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| assert "query-limited dry run" in str(excinfo.value) |
|
|
|
|
| def test_boundary_guardrail_check_rejects_mismatched_query_bank( |
| monkeypatch, tmp_path: Path |
| ): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| other_query_bank_path = tmp_path / "other_query_bank.jsonl" |
| _write_jsonl( |
| other_query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| }, |
| { |
| "query_id": "qb_002", |
| "text": "current speaker generation", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| }, |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| other_query_bank_path, |
| status="pass", |
| available_query_count=2, |
| evaluated_query_count=2, |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| assert "different canonical query bank" in str(excinfo.value) |
|
|
|
|
| def test_boundary_guardrail_check_rejects_mismatched_row_set( |
| monkeypatch, tmp_path: Path |
| ): |
| _stub_boundary_corpus_alignment(monkeypatch) |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| }, |
| { |
| "query_id": "qb_002", |
| "text": "current speaker generation", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| }, |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| query_bank_path, |
| status="pass", |
| available_query_count=2, |
| evaluated_query_count=2, |
| available_query_ids=["qb_001", "qb_999"], |
| evaluated_query_ids=["qb_001", "qb_999"], |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| message = str(excinfo.value) |
| assert "does not match the current boundary_eval row set" in message |
| assert "qb_002" in message |
| assert "qb_999" in message |
|
|
|
|
| def test_boundary_guardrail_check_rejects_mismatched_corpus_fingerprint( |
| monkeypatch, tmp_path: Path |
| ): |
| _stub_boundary_corpus_alignment(monkeypatch, fingerprint="current-corpus") |
| query_bank_path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| query_bank_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
| path = tmp_path / "boundary_behavior_latest.json" |
| path.write_text( |
| json.dumps( |
| _boundary_artifact_payload( |
| query_bank_path, |
| status="pass", |
| corpus_fingerprint="stale-corpus", |
| ) |
| ), |
| encoding="utf-8", |
| ) |
|
|
| with pytest.raises(SystemExit) as excinfo: |
| ensure_boundary_guardrail_passed(path, query_bank_path=query_bank_path) |
|
|
| message = str(excinfo.value) |
| assert "different served corpus snapshot" in message |
| assert "stale-corpus" in message |
| assert "current-corpus" in message |
|
|
|
|
| def test_qdrant_stamp_anchor_parser_is_available(): |
| parser = build_parser() |
| args = parser.parse_args(["qdrant", "stamp-anchor", "--force"]) |
|
|
| assert args.command == "qdrant" |
| assert args.qdrant_command == "stamp-anchor" |
| assert args.force is True |
|
|
|
|
| def test_qdrant_stamp_anchor_prints_result(monkeypatch, capsys): |
| monkeypatch.setattr("sage.cli.state.load_dotenv_if_available", lambda: None) |
|
|
| import sage.services.corpus_alignment as corpus_alignment |
|
|
| monkeypatch.setattr( |
| corpus_alignment, |
| "stamp_corpus_anchor", |
| lambda **_kwargs: { |
| "status": "stamped", |
| "collection_name": "sage_reviews", |
| "corpus_fingerprint": "abc123", |
| }, |
| ) |
|
|
| command_qdrant_stamp_anchor( |
| argparse.Namespace(anchor=Path("data/indexed_product_ids.json"), force=False) |
| ) |
|
|
| output = capsys.readouterr().out |
| assert "status: stamped" in output |
| assert "corpus_fingerprint: abc123" in output |
|
|
|
|
| def test_reset_experiments_parser_supports_dry_run(): |
| parser = build_parser() |
| args = parser.parse_args(["reset", "experiments", "--dry-run"]) |
|
|
| assert args.command == "reset" |
| assert args.reset_command == "experiments" |
| assert args.dry_run is True |
|
|
|
|
| def test_reset_eval_dev_dry_run_matches_eval_dev_language( |
| monkeypatch, tmp_path: Path, capsys |
| ): |
| data_root = tmp_path / "data" |
| (data_root / "eval_results").mkdir(parents=True) |
| (data_root / "eval_results" / "faithfulness_latest.json").write_text( |
| "{}", |
| encoding="utf-8", |
| ) |
|
|
| monkeypatch.setenv("SAGE_DATA_DIR", str(data_root)) |
| monkeypatch.setattr("sage.cli.state.PROJECT_ROOT", tmp_path) |
|
|
| command_reset_eval_dev(argparse.Namespace(dry_run=True)) |
|
|
| output = capsys.readouterr().out |
| assert "Would clear rerunnable evaluation dev artifacts" in output |
| assert "Dry run only; no files were removed." in output |
|
|
|
|
| def test_reset_baseline_parser_supports_dry_run(): |
| parser = build_parser() |
| args = parser.parse_args(["reset", "baseline", "--dry-run"]) |
|
|
| assert args.command == "reset" |
| assert args.reset_command == "baseline" |
| assert args.dry_run is True |
|
|
|
|
| def test_python_module_help_smoke(): |
| repo_root = Path(__file__).resolve().parents[1] |
| result = subprocess.run( |
| [sys.executable, "-m", "sage.cli", "--help"], |
| cwd=repo_root, |
| text=True, |
| capture_output=True, |
| check=False, |
| ) |
|
|
| assert result.returncode == 0 |
| assert "python -m sage.cli" in result.stdout |
| assert "--retrieval-decision baseline-retained" in result.stdout |
| assert result.stderr == "" |
| assert "INFO:" not in result.stdout |
| assert "WARNING:" not in result.stdout |
|
|
|
|
| @pytest.mark.parametrize( |
| "argv", |
| [ |
| ["data", "build", "--subset-size", "0"], |
| ["eval", "run", "--requests", "0"], |
| ["eval", "dev", "--requests", "-1"], |
| ["eval", "boundary", "--max-evidence", "0"], |
| ["demo", "--top-k", "0"], |
| ["serve", "--port", "0"], |
| ["serve", "--port", "65536"], |
| ], |
| ) |
| def test_parser_rejects_invalid_numeric_boundaries(argv): |
| parser = build_parser() |
|
|
| with pytest.raises(SystemExit): |
| parser.parse_args(argv) |
|
|
|
|
| def test_remove_path_deletes_broken_symlink(tmp_path: Path): |
| missing_target = tmp_path / "missing.json" |
| broken_link = tmp_path / "eda_stats_latest.json" |
| broken_link.symlink_to(missing_target) |
|
|
| assert broken_link.is_symlink() |
| assert not broken_link.exists() |
|
|
| remove_path(broken_link) |
|
|
| assert not broken_link.exists() |
| assert not broken_link.is_symlink() |
|
|
|
|
| def test_remove_path_ignores_disappearing_directory(monkeypatch, tmp_path: Path): |
| path = tmp_path / "eval_results" |
| path.mkdir() |
|
|
| from sage.cli import shared as cli_shared |
|
|
| real_rmtree = cli_shared.shutil.rmtree |
|
|
| def flaky_rmtree(target: Path) -> None: |
| real_rmtree(target) |
| raise FileNotFoundError(target) |
|
|
| monkeypatch.setattr(cli_shared.shutil, "rmtree", flaky_rmtree) |
|
|
| remove_path(path) |
|
|
| assert not path.exists() |
|
|
|
|
| def test_reset_artifacts_dry_run_leaves_files_in_place( |
| monkeypatch, tmp_path: Path, capsys |
| ): |
| data_root = tmp_path / "data" |
| eval_results = data_root / "eval_results" |
| eval_results.mkdir(parents=True) |
| eval_file = eval_results / "faithfulness_latest.json" |
| eval_file.write_text("{}", encoding="utf-8") |
| eda_file = data_root / "eda_stats_latest.json" |
| eda_file.write_text("{}", encoding="utf-8") |
|
|
| assets_root = tmp_path / "assets" |
| assets_root.mkdir() |
| figure = assets_root / "retrieval.png" |
| figure.write_text("png", encoding="utf-8") |
|
|
| monkeypatch.setenv("SAGE_DATA_DIR", str(data_root)) |
| monkeypatch.setattr("sage.cli.state.PROJECT_ROOT", tmp_path) |
|
|
| command_reset_artifacts(argparse.Namespace(dry_run=True)) |
|
|
| output = capsys.readouterr().out |
| assert "Would clear rerunnable evaluation artifacts" in output |
| assert "Dry run only; no files were removed." in output |
| assert eval_file.exists() |
| assert eda_file.exists() |
| assert figure.exists() |
|
|
|
|
| def test_reset_experiments_clears_generated_outputs_but_preserves_foundations( |
| monkeypatch, tmp_path: Path, capsys |
| ): |
| data_root = tmp_path / "data" |
| (data_root / "eval_results").mkdir(parents=True) |
| (data_root / "calibration").mkdir() |
| (data_root / "explanations").mkdir() |
| (data_root / "figures").mkdir() |
| (data_root / "query_bank").mkdir() |
|
|
| (data_root / "eval_results" / "faithfulness_latest.json").write_text( |
| "{}", encoding="utf-8" |
| ) |
| (data_root / "calibration" / "evidence_gate_calibration.json").write_text( |
| "{}", encoding="utf-8" |
| ) |
| (data_root / "explanations" / "sample.txt").write_text("x", encoding="utf-8") |
| (data_root / "figures" / "curve.png").write_text("x", encoding="utf-8") |
| (data_root / "query_bank" / "query_bank.jsonl").write_text( |
| '{"query_id":"qb_001","text":"example"}\n', |
| encoding="utf-8", |
| ) |
| (data_root / "query_bank" / "manifest.json").write_text("{}", encoding="utf-8") |
| (data_root / "query_bank" / "query_candidates.jsonl").write_text( |
| '{"query":"example"}\n', |
| encoding="utf-8", |
| ) |
| (data_root / "indexed_product_ids.json").write_text("{}", encoding="utf-8") |
|
|
| assets_root = tmp_path / "assets" |
| assets_root.mkdir() |
| figure = assets_root / "experiment.png" |
| figure.write_text("png", encoding="utf-8") |
|
|
| monkeypatch.setenv("SAGE_DATA_DIR", str(data_root)) |
| monkeypatch.setattr("sage.cli.state.PROJECT_ROOT", tmp_path) |
|
|
| command_reset_experiments(argparse.Namespace(dry_run=False)) |
|
|
| output = capsys.readouterr().out |
| assert "Clearing rerunnable evaluation and experimentation artifacts" in output |
| assert not (data_root / "eval_results").exists() |
| assert not (data_root / "calibration").exists() |
| assert not (data_root / "explanations").exists() |
| assert not (data_root / "figures").exists() |
| assert not figure.exists() |
| assert (data_root / "query_bank" / "query_bank.jsonl").exists() |
| assert (data_root / "query_bank" / "manifest.json").exists() |
| assert (data_root / "query_bank" / "query_candidates.jsonl").exists() |
| assert (data_root / "indexed_product_ids.json").exists() |
|
|
|
|
| def test_reset_baseline_restores_scaffold_and_clears_local_stage_outputs( |
| monkeypatch, tmp_path: Path, capsys |
| ): |
| data_root = tmp_path / "data" |
| (data_root / "eval_results").mkdir(parents=True) |
| (data_root / "calibration").mkdir() |
| (data_root / "explanations").mkdir() |
| (data_root / "figures").mkdir() |
| (data_root / "query_bank" / "sources" / "esci-data").mkdir(parents=True) |
|
|
| (data_root / "eval_results" / "faithfulness_latest.json").write_text( |
| "{}", |
| encoding="utf-8", |
| ) |
| (data_root / "calibration" / "curve.json").write_text("{}", encoding="utf-8") |
| (data_root / "explanations" / "sample.txt").write_text("x", encoding="utf-8") |
| (data_root / "figures" / "plot.png").write_text("x", encoding="utf-8") |
| (data_root / "indexed_product_ids.json").write_text("{}", encoding="utf-8") |
| (data_root / "chunks_418824.jsonl").write_text("{}", encoding="utf-8") |
| (data_root / "sage-stage-data.log").write_text("log", encoding="utf-8") |
| (data_root / "query_bank" / "query_bank.jsonl").write_text( |
| '{"query_id":"qb_001","text":"example"}\n', |
| encoding="utf-8", |
| ) |
| (data_root / "query_bank" / "manifest.json").write_text("{}", encoding="utf-8") |
| (data_root / "query_bank" / "query_candidates.jsonl").write_text( |
| '{"query":"example"}\n', |
| encoding="utf-8", |
| ) |
| (data_root / "query_bank" / "split_leakage_audit.json").write_text( |
| "{}", |
| encoding="utf-8", |
| ) |
| (data_root / "query_bank" / "README.md").write_text("query docs", encoding="utf-8") |
| (data_root / "query_bank" / "sources" / "README.md").write_text( |
| "source docs", |
| encoding="utf-8", |
| ) |
| ( |
| data_root / "query_bank" / "sources" / "manual_boundary_queries_v2.jsonl" |
| ).write_text( |
| '{"manual_id":"bq_001","text":"x"}\n', |
| encoding="utf-8", |
| ) |
| (data_root / "README.md").write_text("root docs", encoding="utf-8") |
| (data_root / ".gitignore").write_text("*\n", encoding="utf-8") |
|
|
| monkeypatch.setenv("SAGE_DATA_DIR", str(data_root)) |
| monkeypatch.setattr("sage.cli.state.PROJECT_ROOT", tmp_path) |
|
|
| command_reset_baseline(argparse.Namespace(dry_run=False)) |
|
|
| output = capsys.readouterr().out |
| assert "baseline scaffold contract" in output |
| assert not (data_root / "indexed_product_ids.json").exists() |
| assert not (data_root / "chunks_418824.jsonl").exists() |
| assert not (data_root / "sage-stage-data.log").exists() |
| assert not (data_root / "query_bank" / "query_bank.jsonl").exists() |
| assert not (data_root / "query_bank" / "manifest.json").exists() |
| assert not (data_root / "query_bank" / "query_candidates.jsonl").exists() |
| assert not (data_root / "query_bank" / "split_leakage_audit.json").exists() |
| assert not (data_root / "query_bank" / "sources" / "esci-data").exists() |
| assert ( |
| data_root / "query_bank" / "sources" / "manual_boundary_queries_v2.jsonl" |
| ).exists() |
| assert (data_root / "calibration" / ".gitkeep").exists() |
| assert (data_root / "eval_results" / ".gitkeep").exists() |
| assert (data_root / "explanations" / ".gitkeep").exists() |
| assert (data_root / "figures" / ".gitkeep").exists() |
| assert (data_root / "query_bank" / ".gitkeep").exists() |
| assert (data_root / "query_bank" / "sources" / ".gitkeep").exists() |
| assert (data_root / "query_bank" / "README.md").exists() |
| assert (data_root / "query_bank" / "sources" / "README.md").exists() |
|
|
|
|
| def test_reset_baseline_dry_run_is_preview_only_and_preserves_checked_in_sources( |
| monkeypatch, tmp_path: Path, capsys |
| ): |
| data_root = tmp_path / "data" |
| (data_root / "query_bank" / "sources" / "esci-data").mkdir(parents=True) |
| ( |
| data_root / "query_bank" / "sources" / "manual_boundary_queries_v2.jsonl" |
| ).write_text( |
| '{"manual_id":"bq_001","text":"x"}\n', |
| encoding="utf-8", |
| ) |
| (data_root / "query_bank" / "sources" / "README.md").write_text( |
| "source docs", |
| encoding="utf-8", |
| ) |
| (data_root / "README.md").write_text("root docs", encoding="utf-8") |
| (data_root / ".gitignore").write_text("*\n", encoding="utf-8") |
|
|
| monkeypatch.setenv("SAGE_DATA_DIR", str(data_root)) |
| monkeypatch.setattr("sage.cli.state.PROJECT_ROOT", tmp_path) |
|
|
| command_reset_baseline(argparse.Namespace(dry_run=True)) |
|
|
| output = capsys.readouterr().out |
| assert "Dry run only; no files were removed." in output |
| assert "Preview only; rerun without --dry-run to apply these changes." in output |
| assert ( |
| "Done. Local data state is back at the baseline scaffold boundary." |
| not in output |
| ) |
| assert " - data/query_bank/sources/manual_boundary_queries_v2.jsonl" not in output |
| assert ( |
| data_root / "query_bank" / "sources" / "manual_boundary_queries_v2.jsonl" |
| ).exists() |
|
|
|
|
| def test_eval_run_preflight_rejects_empty_query_bank(tmp_path: Path): |
| path = tmp_path / "query_bank.jsonl" |
| path.write_text("", encoding="utf-8") |
|
|
| with pytest.raises(SystemExit, match="required query-bank subsets are empty"): |
| ensure_eval_query_bank_ready(path) |
|
|
|
|
| def test_eval_run_preflight_accepts_required_query_subsets(tmp_path: Path): |
| path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "best travel keyboard", |
| "source_type": "amazon_esci", |
| "subset_tags": [DEFAULT_RETRIEVAL_FINAL_REPORT_SUBSET_TAG], |
| "relevant_items": {"ASIN1": 3.0}, |
| }, |
| { |
| "query_id": "qb_004", |
| "text": "best newest earbuds", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| }, |
| ], |
| ) |
|
|
| ensure_eval_query_bank_ready(path) |
|
|
|
|
| def test_boundary_eval_preflight_rejects_empty_boundary_subset(tmp_path: Path): |
| path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "best travel keyboard", |
| "source_type": "amazon_esci", |
| "subset_tags": [DEFAULT_RETRIEVAL_FINAL_REPORT_SUBSET_TAG], |
| "relevant_items": {"ASIN1": 3.0}, |
| } |
| ], |
| ) |
|
|
| with pytest.raises(SystemExit, match="sage eval boundary"): |
| ensure_boundary_eval_query_bank_ready(path=path) |
|
|
|
|
| def test_command_eval_boundary_runs_script_with_forwarded_args( |
| monkeypatch, tmp_path: Path |
| ): |
| path = tmp_path / "query_bank.jsonl" |
| _write_jsonl( |
| path, |
| [ |
| { |
| "query_id": "qb_001", |
| "text": "latest headphones to avoid", |
| "source_type": "manual_boundary", |
| "subset_tags": ["boundary_eval"], |
| } |
| ], |
| ) |
|
|
| commands: list[list[str]] = [] |
|
|
| monkeypatch.setattr("sage.cli.evaluation.ensure_env", lambda: None) |
| monkeypatch.setattr( |
| "sage.cli.evaluation.run_command", |
| lambda command, **_kwargs: commands.append(list(command)), |
| ) |
|
|
| command_eval_boundary( |
| argparse.Namespace( |
| query_bank_path=path, |
| subset_tag="boundary_eval", |
| query_limit=5, |
| top_k=2, |
| min_rating=None, |
| aggregation="weighted_mean", |
| max_evidence=2, |
| ) |
| ) |
|
|
| assert commands |
| command = commands[0] |
| assert command[:2] == [sys.executable, "scripts/evaluate_boundary_behavior.py"] |
| assert "--query-bank-path" in command |
| assert str(path) in command |
| assert "--query-limit" in command |
| assert "--aggregation" in command |
| assert "weighted_mean" in command |
| assert "--artifact-scope" in command |
| assert "dev" in command |
| assert "--min-rating" not in command |
|
|
|
|
| def test_eval_run_preflight_rejects_missing_faithfulness_cases(tmp_path: Path): |
| path = tmp_path / "faithfulness_cases.jsonl" |
|
|
| with pytest.raises( |
| SystemExit, |
| match="freeze-time manifest are not ready", |
| ): |
| ensure_faithfulness_cases_ready(path) |
|
|
|
|
| def test_eval_run_preflight_rejects_missing_faithfulness_outcomes(tmp_path: Path): |
| path = tmp_path / "faithfulness_cases.jsonl" |
| _write_jsonl( |
| path, |
| [ |
| { |
| "case_id": "fc_001", |
| "query_id": "qb_001", |
| "query": "speaker with clear vocals", |
| "source_subset": "faithfulness_seed", |
| "source_type": "manual_seed", |
| "product_id": "ASIN1", |
| "product_score": 0.91, |
| "product_rank": 1, |
| "avg_rating": 4.6, |
| "aggregation": "max", |
| "evidence": [ |
| { |
| "text": "Very clear vocals and speech.", |
| "score": 0.91, |
| "product_id": "ASIN1", |
| "rating": 5.0, |
| "review_id": "review_1", |
| } |
| ], |
| } |
| ], |
| ) |
|
|
| with pytest.raises( |
| SystemExit, |
| match="freeze-time manifest are not ready", |
| ): |
| ensure_faithfulness_cases_ready(path) |
|
|
|
|
| def test_eval_run_preflight_rejects_missing_faithfulness_manifest(tmp_path: Path): |
| path = tmp_path / "faithfulness_cases.jsonl" |
| outcomes_path = tmp_path / "faithfulness_case_outcomes.jsonl" |
| _write_jsonl( |
| path, |
| [ |
| { |
| "case_id": "fc_001", |
| "query_id": "qb_001", |
| "query": "speaker with clear vocals", |
| "source_subset": "faithfulness_seed", |
| "source_type": "manual_seed", |
| "product_id": "ASIN1", |
| "product_score": 0.91, |
| "product_rank": 1, |
| "avg_rating": 4.6, |
| "aggregation": "max", |
| "evidence": [ |
| { |
| "text": "Very clear vocals and speech.", |
| "score": 0.91, |
| "product_id": "ASIN1", |
| "rating": 5.0, |
| "review_id": "review_1", |
| } |
| ], |
| } |
| ], |
| ) |
| _write_jsonl( |
| outcomes_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "query": "speaker with clear vocals", |
| "source_subset": "faithfulness_seed", |
| "source_type": "manual_seed", |
| "outcome_status": "materialized", |
| "materialized_case_id": "fc_001", |
| "product_id": "ASIN1", |
| "aggregation": "max", |
| } |
| ], |
| ) |
|
|
| with pytest.raises(SystemExit, match="freeze-time manifest are not ready"): |
| ensure_faithfulness_cases_ready(path, outcomes_path=outcomes_path) |
|
|
|
|
| def test_eval_run_preflight_accepts_faithfulness_cases(tmp_path: Path): |
| path = tmp_path / "faithfulness_cases.jsonl" |
| outcomes_path = tmp_path / "faithfulness_case_outcomes.jsonl" |
| manifest_path = tmp_path / "faithfulness_cases.manifest.json" |
| _write_jsonl( |
| path, |
| [ |
| { |
| "case_id": "fc_001", |
| "query_id": "qb_001", |
| "query": "speaker with clear vocals", |
| "source_subset": "faithfulness_seed", |
| "source_type": "manual_seed", |
| "product_id": "ASIN1", |
| "product_score": 0.91, |
| "product_rank": 1, |
| "avg_rating": 4.6, |
| "aggregation": "max", |
| "evidence": [ |
| { |
| "text": "Very clear vocals and speech.", |
| "score": 0.91, |
| "product_id": "ASIN1", |
| "rating": 5.0, |
| "review_id": "review_1", |
| } |
| ], |
| } |
| ], |
| ) |
| _write_jsonl( |
| outcomes_path, |
| [ |
| { |
| "query_id": "qb_001", |
| "query": "speaker with clear vocals", |
| "source_subset": "faithfulness_seed", |
| "source_type": "manual_seed", |
| "outcome_status": "materialized", |
| "materialized_case_id": "fc_001", |
| "product_id": "ASIN1", |
| "aggregation": "max", |
| } |
| ], |
| ) |
| manifest_path.write_text( |
| json.dumps({"reference_timestamp_ms": 1736553600000}), |
| encoding="utf-8", |
| ) |
|
|
| ensure_faithfulness_cases_ready( |
| path, |
| outcomes_path=outcomes_path, |
| manifest_path=manifest_path, |
| ) |
|
|
|
|
| def test_command_eval_requires_calibration_handoff(monkeypatch): |
| calls: list[str] = [] |
|
|
| monkeypatch.setattr( |
| evaluation_readiness, |
| "ensure_eval_query_bank_ready", |
| lambda: calls.append("query_bank"), |
| ) |
| monkeypatch.setattr( |
| evaluation_readiness, |
| "ensure_faithfulness_cases_ready", |
| lambda: calls.append("faithfulness_cases"), |
| ) |
|
|
| def fail_calibration_handoff() -> None: |
| calls.append("calibration_handoff") |
| raise SystemExit("handoff missing") |
|
|
| monkeypatch.setattr( |
| evaluation_cli, |
| "ensure_calibration_handoff_ready", |
| fail_calibration_handoff, |
| ) |
|
|
| with pytest.raises(SystemExit, match="handoff missing"): |
| evaluation_cli.command_eval( |
| argparse.Namespace( |
| samples=None, |
| ragas_samples=None, |
| url="https://example.com", |
| requests=1, |
| ) |
| ) |
|
|
| assert calls == ["query_bank", "faithfulness_cases", "calibration_handoff"] |
|
|
