import importlib import importlib.util import json import tomllib from datetime import datetime, timezone from io import StringIO from pathlib import Path from types import SimpleNamespace import pytest from agent_lab.application.benchmark import ( BenchmarkCaseId, BenchmarkMode, BenchmarkRunResult, ) def _cli_module(): spec = importlib.util.find_spec("agent_lab.presentation.benchmark_cli") assert spec is not None, "benchmark_cli module must be implemented" return importlib.import_module("agent_lab.presentation.benchmark_cli") def _write_config(path: Path, **overrides: object) -> Path: payload = { "schema_version": 1, "base_url": "https://json.example/v1", "model": "json-model", "runs_per_case": 1, "cases": ["ordinary_chat"], "modes": ["dual_agent"], } payload.update(overrides) path.write_text(json.dumps(payload)) return path def _run_result( *, case_id: BenchmarkCaseId = BenchmarkCaseId.ORDINARY_CHAT, mode: BenchmarkMode = BenchmarkMode.DUAL_AGENT, iteration: int = 1, status: str = "passed", error: str | None = None, ) -> BenchmarkRunResult: return BenchmarkRunResult( case_id=case_id, mode=mode, iteration=iteration, status=status, initial_provider_ttft_ms=1, visible_ttft_ms=2, turn_wall_time_ms=3, prompt_tokens=2, completion_tokens=3, total_tokens=5, cached_tokens=0, model_call_count=1, fallback_count=0, tool_count=0, event_names=[], batch_event_names=[], tool_event_names=[], event_sources=[], tool_statuses=[], tool_latencies_ms=[], semantic_failures=[], error=error, model_calls=[], ) class FakeRunner: def __init__(self, results: list[BenchmarkRunResult]) -> None: self.results = results async def run(self) -> list[BenchmarkRunResult]: return self.results class FailingTextStream(StringIO): def __init__( self, *, write_error: BaseException | None = None, flush_error: BaseException | None = None, ) -> None: super().__init__() self.write_error = write_error self.flush_error = flush_error def write(self, value: str) -> int: if self.write_error is not None: raise self.write_error return super().write(value) def flush(self) -> None: if self.flush_error is not None: raise self.flush_error super().flush() def _public_report_paths(output_dir: Path) -> tuple[Path, Path]: published_entries = [ path for path in output_dir.iterdir() if not path.name.startswith(".") ] assert len(published_entries) == 1 published_bundle = published_entries[0] assert published_bundle.is_symlink() return ( published_bundle / "report.json", published_bundle / "report.md", ) def _report_paths_from_stdout( stdout: StringIO, output_dir: Path, ) -> tuple[Path, Path]: report_paths: dict[str, Path] = {} for line in stdout.getvalue().splitlines(): key, separator, value = line.partition("=") if separator and key in {"json_report", "markdown_report"}: assert key not in report_paths report_paths[key] = Path(value) assert set(report_paths) == {"json_report", "markdown_report"} json_path = report_paths["json_report"] markdown_path = report_paths["markdown_report"] assert (json_path, markdown_path) == _public_report_paths(output_dir) assert json_path.is_file() assert markdown_path.is_file() return json_path, markdown_path def _dependencies( cli, *, results: list[BenchmarkRunResult], api_key: str = "", runner_calls: list[dict[str, object]] | None = None, report_writer=None, stderr: StringIO | None = None, stdout: StringIO | None = None, ): calls = runner_calls if runner_calls is not None else [] def runner_factory(**kwargs): calls.append(kwargs) return FakeRunner(results) return cli.BenchmarkCliDependencies( settings_factory=lambda: SimpleNamespace( openai_api_key=api_key, openai_base_url="https://env-must-not-win.example/v1", openai_default_model="env-must-not-win", ), runner_factory=runner_factory, report_writer=report_writer or cli.write_benchmark_report, now=lambda: datetime(2026, 7, 14, 1, 2, 3, tzinfo=timezone.utc), stdout=stdout or StringIO(), stderr=stderr or StringIO(), ) def test_cli_overrides_json_and_repeatable_case_mode_replace_lists(tmp_path: Path): cli = _cli_module() config_path = _write_config( tmp_path / "config.json", cases=["ordinary_chat", "web_search_two_answers"], modes=["dual_agent", "chat_agent_tools"], ) output_dir = tmp_path / "reports" stdout = StringIO() runner_calls: list[dict[str, object]] = [] dependencies = _dependencies( cli, results=[ _run_result( case_id=case_id, mode=BenchmarkMode.CHAT_AGENT_TOOLS, iteration=iteration, ) for case_id in ( BenchmarkCaseId.ORDINARY_CHAT, BenchmarkCaseId.SESSION_TERMINATE, ) for iteration in range(1, 4) ], api_key="env-key", runner_calls=runner_calls, stdout=stdout, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--base-url", "https://cli.example/v1", "--model", "cli-model", "--runs", "3", "--case", "ordinary_chat", "--case", "session_terminate", "--mode", "chat_agent_tools", "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 0 assert len(runner_calls) == 1 resolved = runner_calls[0] assert resolved["api_key"] == "env-key" assert resolved["mock"] is False config = resolved["config"] assert config.base_url == "https://cli.example/v1" assert config.model == "cli-model" assert config.runs_per_case == 3 assert config.cases == [ BenchmarkCaseId.ORDINARY_CHAT, BenchmarkCaseId.SESSION_TERMINATE, ] assert config.modes == [BenchmarkMode.CHAT_AGENT_TOOLS] _report_paths_from_stdout(stdout, output_dir) def test_mock_run_needs_no_api_key_and_exits_zero(tmp_path: Path): cli = _cli_module() config_path = _write_config(tmp_path / "config.json") output_dir = tmp_path / "reports" stdout = StringIO() stderr = StringIO() dependencies = _dependencies( cli, results=[_run_result()], api_key="", stdout=stdout, stderr=stderr, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--mock", "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 0 assert "attempted=1 passed=1 failed=0" in stdout.getvalue() _report_paths_from_stdout(stdout, output_dir) assert stderr.getvalue() == "" def test_partial_failed_run_exits_one_writes_redacted_reports(tmp_path: Path): cli = _cli_module() secret = "sk-cli-real-secret" config_path = _write_config( tmp_path / "config.json", runs_per_case=2, ) output_dir = tmp_path / "reports" stdout = StringIO() stderr = StringIO() dependencies = _dependencies( cli, results=[ _run_result(iteration=1), _run_result( iteration=2, status="failed", error=f"Authorization: Bearer {secret}", ), ], api_key=secret, stdout=stdout, stderr=stderr, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 1 assert "attempted=2 passed=1 failed=1" in stdout.getvalue() assert secret not in stdout.getvalue() assert stderr.getvalue() == "" json_path, markdown_path = _report_paths_from_stdout(stdout, output_dir) report_text = "\n".join( path.read_text() for path in (json_path, markdown_path) ) assert secret not in report_text assert "[REDACTED]" in report_text @pytest.mark.parametrize( "payload", [ {"schema_version": 1, "api_key": "secret-in-json"}, {"schema_version": 1, "base_url": "not-a-url", "model": "m"}, ], ) def test_bad_config_exits_two_without_leaking_secret( tmp_path: Path, payload: dict[str, object] ): cli = _cli_module() config_path = tmp_path / "bad.json" config_path.write_text(json.dumps(payload)) stderr = StringIO() runner_calls: list[dict[str, object]] = [] dependencies = _dependencies( cli, results=[], runner_calls=runner_calls, stderr=stderr, ) exit_code = cli.run_cli( ["--config", str(config_path), "--mock"], dependencies=dependencies, ) assert exit_code == 2 assert runner_calls == [] assert "secret-in-json" not in stderr.getvalue() assert "benchmark error:" in stderr.getvalue() def test_final_validation_allows_cli_to_supply_missing_base_url_and_model( tmp_path: Path, ): cli = _cli_module() config_path = tmp_path / "minimal.json" config_path.write_text( json.dumps( { "schema_version": 1, "cases": ["ordinary_chat"], "modes": ["dual_agent"], } ) ) runner_calls: list[dict[str, object]] = [] dependencies = _dependencies( cli, results=[_run_result()], runner_calls=runner_calls, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--base-url", "https://cli.example/v1", "--model", "cli-model", "--mock", "--output-dir", str(tmp_path / "reports"), ], dependencies=dependencies, ) assert exit_code == 0 assert runner_calls[0]["config"].model == "cli-model" def test_live_run_without_api_key_exits_two(tmp_path: Path): cli = _cli_module() config_path = _write_config(tmp_path / "config.json") stderr = StringIO() runner_calls: list[dict[str, object]] = [] dependencies = _dependencies( cli, results=[], api_key="", runner_calls=runner_calls, stderr=stderr, ) exit_code = cli.run_cli( ["--config", str(config_path)], dependencies=dependencies, ) assert exit_code == 2 assert runner_calls == [] assert "api key is required" in stderr.getvalue().lower() def test_report_failure_exits_two_and_redacts_stderr(tmp_path: Path): cli = _cli_module() secret = "sk-report-failure-secret" config_path = _write_config(tmp_path / "config.json") stderr = StringIO() def failing_writer(report, output_dir): del report, output_dir raise OSError(f"API-Key: {secret}") dependencies = _dependencies( cli, results=[_run_result()], api_key=secret, report_writer=failing_writer, stderr=stderr, ) exit_code = cli.run_cli( ["--config", str(config_path)], dependencies=dependencies, ) assert exit_code == 2 assert secret not in stderr.getvalue() assert "[REDACTED]" in stderr.getvalue() def test_unexpected_runner_failure_still_attempts_report_write(tmp_path: Path): cli = _cli_module() secret = "runner-secret" config_path = _write_config(tmp_path / "config.json") writes: list[object] = [] stderr = StringIO() stdout = StringIO() class FailingRunner: async def run(self): raise RuntimeError(f"Bearer {secret}") def write_report(report, output_dir): writes.append(report) output = Path(output_dir) output.mkdir(parents=True) json_path = output / "report.json" md_path = output / "report.md" json_path.write_text("{}") md_path.write_text("# report") return json_path, md_path dependencies = cli.BenchmarkCliDependencies( settings_factory=lambda: SimpleNamespace(openai_api_key=secret), runner_factory=lambda **kwargs: FailingRunner(), report_writer=write_report, now=lambda: datetime(2026, 7, 14, tzinfo=timezone.utc), stdout=stdout, stderr=stderr, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--output-dir", str(tmp_path / "reports"), ], dependencies=dependencies, ) assert exit_code == 2 assert len(writes) == 1 assert writes[0].overall.attempted == 1 assert writes[0].overall.failed == 1 assert writes[0].overall.system_errors == ["Bearer [REDACTED]"] assert "result count mismatch" not in writes[0].overall.system_errors[0] assert "attempted=1 passed=0 failed=1" in stdout.getvalue() assert secret not in stdout.getvalue() assert secret not in stderr.getvalue() assert "Bearer [REDACTED]" in stderr.getvalue() @pytest.mark.parametrize("failure_point", ["write", "flush"]) def test_summary_output_failure_after_report_write_exits_two_and_redacts_stderr( tmp_path: Path, failure_point: str, ): cli = _cli_module() secret = "sk-summary-output-secret" config_path = _write_config(tmp_path / "config.json") output_dir = tmp_path / "reports" error = BrokenPipeError(f"Bearer {secret}") stdout = FailingTextStream( write_error=error if failure_point == "write" else None, flush_error=error if failure_point == "flush" else None, ) stderr = StringIO() dependencies = _dependencies( cli, results=[_run_result()], api_key=secret, stdout=stdout, stderr=stderr, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 2 json_path, markdown_path = _public_report_paths(output_dir) assert json_path.is_file() assert markdown_path.is_file() assert secret not in stderr.getvalue() assert "Bearer [REDACTED]" in stderr.getvalue() def test_execution_error_stderr_failure_still_exits_two(tmp_path: Path): cli = _cli_module() secret = "sk-execution-error-secret" config_path = _write_config(tmp_path / "config.json") output_dir = tmp_path / "reports" stderr = FailingTextStream( write_error=BrokenPipeError(f"Bearer {secret}") ) dependencies = _dependencies( cli, results=[], api_key=secret, stderr=stderr, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 2 json_path, markdown_path = _public_report_paths(output_dir) assert json_path.is_file() assert markdown_path.is_file() assert secret not in stderr.getvalue() def test_summary_output_keyboard_interrupt_still_exits_130(tmp_path: Path): cli = _cli_module() config_path = _write_config(tmp_path / "config.json") output_dir = tmp_path / "reports" dependencies = _dependencies( cli, results=[_run_result()], stdout=FailingTextStream(write_error=KeyboardInterrupt()), ) exit_code = cli.run_cli( [ "--config", str(config_path), "--mock", "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 130 @pytest.mark.parametrize( ("results", "attempted", "passed", "failed", "actual_count"), [ ([], 1, 0, 1, 0), ([_run_result(iteration=1), _run_result(iteration=2)], 3, 2, 1, 2), ], ) def test_runner_result_cardinality_mismatch_is_reported_and_exits_two( tmp_path: Path, results: list[BenchmarkRunResult], attempted: int, passed: int, failed: int, actual_count: int, ): cli = _cli_module() config_path = _write_config(tmp_path / "config.json") output_dir = tmp_path / "reports" stdout = StringIO() stderr = StringIO() dependencies = _dependencies( cli, results=results, stdout=stdout, stderr=stderr, ) exit_code = cli.run_cli( [ "--config", str(config_path), "--mock", "--output-dir", str(output_dir), ], dependencies=dependencies, ) assert exit_code == 2 json_path, markdown_path = _report_paths_from_stdout(stdout, output_dir) payload = json.loads(json_path.read_text()) expected_error = ( f"runner result count mismatch: expected 1, got {actual_count}" ) assert payload["overall"] == { "attempted": attempted, "passed": passed, "failed": failed, "warnings": payload["overall"]["warnings"], "system_errors": [expected_error], } assert expected_error in markdown_path.read_text() assert f"attempted={attempted} passed={passed} failed={failed}" in ( stdout.getvalue() ) assert expected_error in stderr.getvalue() def test_help_and_required_config_use_argparse(capsys: pytest.CaptureFixture[str]): cli = _cli_module() with pytest.raises(SystemExit) as help_exit: cli.run_cli(["--help"]) assert help_exit.value.code == 0 assert "--config" in capsys.readouterr().out with pytest.raises(SystemExit) as missing_exit: cli.run_cli([]) assert missing_exit.value.code == 2 @pytest.mark.parametrize( "invalid_args", [ ["--runs", "api_key=runs-secret"], ["--case", "api_key=case-secret"], ], ) def test_argparse_errors_use_only_generic_safe_message( capsys: pytest.CaptureFixture[str], invalid_args: list[str], ): cli = _cli_module() with pytest.raises(SystemExit) as exc_info: cli.run_cli( [ "--config", "unused.json", *invalid_args, ] ) assert exc_info.value.code == 2 stderr = capsys.readouterr().err assert "runs-secret" not in stderr assert "case-secret" not in stderr assert "api_key" not in stderr assert "invalid command-line arguments" in stderr assert "invalid int value" not in stderr assert "invalid choice" not in stderr assert "choose from" not in stderr def test_main_raises_system_exit(monkeypatch: pytest.MonkeyPatch): cli = _cli_module() monkeypatch.setattr(cli, "run_cli", lambda argv=None: 1) with pytest.raises(SystemExit) as exc_info: cli.main() assert exc_info.value.code == 1 def test_pyproject_registers_benchmark_console_script(): pyproject = tomllib.loads(Path("pyproject.toml").read_text()) assert pyproject["project"]["scripts"]["agent-lab-benchmark"] == ( "agent_lab.presentation.benchmark_cli:main" )