cleveragents-core/benchmarks/auth_cli_bench.py

"""ASV benchmarks for Auth and Team CLI command throughput.

Measures the performance of:
- Auth login command (token store + CLI rendering)
- Auth logout command
- Auth status command
- Team list command
- Team use command
- TokenStore store/get/clear cycle
"""

from __future__ import annotations

import importlib
import shutil
import sys
import tempfile
from pathlib import Path
from typing import Any
from unittest.mock import patch

# Ensure the local *source* tree is importable.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from typer.testing import CliRunner  # noqa: E402

from cleveragents.cli.commands import auth as auth_mod  # noqa: E402
from cleveragents.cli.commands import team as team_mod  # noqa: E402
from cleveragents.cli.commands.auth import app as auth_app  # noqa: E402
from cleveragents.cli.commands.team import app as team_app  # noqa: E402
from cleveragents.infrastructure.auth.token_store import TokenStore  # noqa: E402

_runner = CliRunner()
_config_data: dict[str, Any] = {}


def _make_store() -> tuple[TokenStore, Path]:
    tmpdir = Path(tempfile.mkdtemp())
    return TokenStore(token_dir=tmpdir, use_keyring=False), tmpdir


def _patched_read_team() -> str | None:
    val = _config_data.get("active_team")
    if val is not None and str(val).strip():
        return str(val).strip()
    return None


def _patched_write_team(team: str | None) -> None:
    global _config_data
    if team is None:
        _config_data.pop("active_team", None)
    else:
        _config_data["active_team"] = team


class AuthLoginSuite:
    """Benchmark auth login throughput."""

    def setup(self) -> None:
        global _config_data
        _config_data = {}
        self._store, self._tmpdir = _make_store()

    def teardown(self) -> None:
        shutil.rmtree(str(self._tmpdir), ignore_errors=True)

    def time_auth_login(self) -> None:
        def _ps() -> TokenStore:
            return self._store

        with (
            patch.object(auth_mod, "_get_token_store", _ps),
            patch.object(auth_mod, "_read_active_team", _patched_read_team),
            patch.object(auth_mod, "_write_active_team", _patched_write_team),
        ):
            _runner.invoke(auth_app, ["login", "--token", "bench-tok"])


class AuthLogoutSuite:
    """Benchmark auth logout throughput."""

    def setup(self) -> None:
        global _config_data
        _config_data = {}
        self._store, self._tmpdir = _make_store()
        self._store.store_token("bench-tok")

    def teardown(self) -> None:
        shutil.rmtree(str(self._tmpdir), ignore_errors=True)

    def time_auth_logout(self) -> None:
        def _ps() -> TokenStore:
            return self._store

        with (
            patch.object(auth_mod, "_get_token_store", _ps),
            patch.object(auth_mod, "_read_active_team", _patched_read_team),
            patch.object(auth_mod, "_write_active_team", _patched_write_team),
        ):
            _runner.invoke(auth_app, ["logout"])


class AuthStatusSuite:
    """Benchmark auth status throughput."""

    def setup(self) -> None:
        global _config_data
        _config_data = {}
        self._store, self._tmpdir = _make_store()

    def teardown(self) -> None:
        shutil.rmtree(str(self._tmpdir), ignore_errors=True)

    def time_auth_status(self) -> None:
        def _ps() -> TokenStore:
            return self._store

        with (
            patch.object(auth_mod, "_get_token_store", _ps),
            patch.object(auth_mod, "_read_active_team", _patched_read_team),
            patch.object(auth_mod, "_write_active_team", _patched_write_team),
        ):
            _runner.invoke(auth_app, ["status"])


class TeamListSuite:
    """Benchmark team list throughput."""

    def setup(self) -> None:
        global _config_data
        _config_data = {}
        self._store, self._tmpdir = _make_store()

    def teardown(self) -> None:
        shutil.rmtree(str(self._tmpdir), ignore_errors=True)

    def time_team_list(self) -> None:
        with (
            patch.object(team_mod, "_read_active_team", _patched_read_team),
            patch.object(team_mod, "_write_active_team", _patched_write_team),
        ):
            _runner.invoke(team_app, ["list"])


class TeamUseSuite:
    """Benchmark team use throughput."""

    def setup(self) -> None:
        global _config_data
        _config_data = {}
        self._store, self._tmpdir = _make_store()

    def teardown(self) -> None:
        shutil.rmtree(str(self._tmpdir), ignore_errors=True)

    def time_team_use(self) -> None:
        with (
            patch.object(team_mod, "_read_active_team", _patched_read_team),
            patch.object(team_mod, "_write_active_team", _patched_write_team),
        ):
            _runner.invoke(team_app, ["use", "bench-team"])


class TokenStoreSuite:
    """Benchmark token store operations."""

    def setup(self) -> None:
        self._store, self._tmpdir = _make_store()

    def teardown(self) -> None:
        shutil.rmtree(str(self._tmpdir), ignore_errors=True)

    def time_store_token(self) -> None:
        self._store.store_token("bench-token-value")

    def time_get_token(self) -> None:
        self._store.store_token("bench-token-value")
        self._store.get_token()

    def time_clear_token(self) -> None:
        self._store.store_token("bench-token-value")
        self._store.clear_token()