agent-lab/tests/test_benchmark_cli.py

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 _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"
    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,
    )

    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]
    assert len(list(output_dir.glob("*.json"))) == 1
    assert len(list(output_dir.glob("*.md"))) == 1


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()
    assert "json_report=" in stdout.getvalue()
    assert "markdown_report=" in stdout.getvalue()
    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() == ""
    report_text = "\n".join(
        path.read_text() for path in sorted(output_dir.iterdir())
    )
    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
    assert any(output_dir.rglob("*.json"))
    assert any(output_dir.rglob("*.md"))
    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
    assert any(output_dir.rglob("*.json"))
    assert any(output_dir.rglob("*.md"))
    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 = next(output_dir.glob("*.json"))
    markdown_path = next(output_dir.glob("*.md"))
    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"
    )