"""ASV benchmarks for Actor CLI command throughput. Measures the performance of: - Actor add (config parse + validate) - Actor list rendering (rich / json / yaml / plain) - Actor show rendering (rich / json / yaml / plain) - Actor remove """ from __future__ import annotations import importlib import json import os import sys import tempfile from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch # Ensure the local *source* tree is importable even when ASV has an # older build of the package installed. _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.actor import app as actor_app # noqa: E402 from cleveragents.domain.models.core.actor import Actor # noqa: E402 _VALID_CONFIG = { "name": "local/bench", "provider": "openai", "model": "gpt-4", "temperature": 0.5, "max_tokens": 256, } _runner = CliRunner() def _mock_actor( name: str = "local/bench-actor", provider: str = "openai", model: str = "gpt-4", config: dict[str, Any] | None = None, ) -> Actor: blob = config or {"provider": provider, "model": model} return Actor( id=1, name=name, provider=provider, model=model, config_blob=blob, config_hash=Actor.compute_hash(blob), unsafe=False, is_built_in=False, is_default=False, ) class ActorCLIAddSuite: """Benchmark actor add --config throughput.""" def setup(self) -> None: fd, self._path = tempfile.mkstemp(suffix=".json") with os.fdopen(fd, "w") as fh: json.dump(_VALID_CONFIG, fh) self._mock_registry = MagicMock() self._mock_registry.upsert_actor.return_value = _mock_actor() self._patcher = patch( "cleveragents.cli.commands.actor._get_services", return_value=(MagicMock(), self._mock_registry), ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() Path(self._path).unlink(missing_ok=True) def time_add_from_config(self) -> None: """Benchmark add --config end-to-end.""" _runner.invoke(actor_app, ["add", "--config", self._path]) class ActorCLIListSuite: """Benchmark actor list throughput.""" def setup(self) -> None: self._mock_registry = MagicMock() self._mock_registry.list_actors.return_value = [ _mock_actor(f"local/actor-{i}") for i in range(50) ] self._patcher = patch( "cleveragents.cli.commands.actor._get_services", return_value=(MagicMock(), self._mock_registry), ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_list_rich(self) -> None: """Benchmark listing actors (rich default).""" _runner.invoke(actor_app, ["list"]) def time_list_json(self) -> None: """Benchmark listing actors (json format).""" _runner.invoke(actor_app, ["list", "--format", "json"]) def time_list_yaml(self) -> None: """Benchmark listing actors (yaml format).""" _runner.invoke(actor_app, ["list", "--format", "yaml"]) def time_list_plain(self) -> None: """Benchmark listing actors (plain format).""" _runner.invoke(actor_app, ["list", "--format", "plain"]) class ActorCLIShowSuite: """Benchmark actor show throughput.""" def setup(self) -> None: self._mock_registry = MagicMock() self._mock_registry.get_actor.return_value = _mock_actor() self._patcher = patch( "cleveragents.cli.commands.actor._get_services", return_value=(MagicMock(), self._mock_registry), ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_show_rich(self) -> None: """Benchmark showing actor (rich default).""" _runner.invoke(actor_app, ["show", "local/bench-actor"]) def time_show_json(self) -> None: """Benchmark showing actor (json format).""" _runner.invoke(actor_app, ["show", "local/bench-actor", "--format", "json"]) def time_show_yaml(self) -> None: """Benchmark showing actor (yaml format).""" _runner.invoke(actor_app, ["show", "local/bench-actor", "--format", "yaml"]) def time_show_plain(self) -> None: """Benchmark showing actor (plain format).""" _runner.invoke(actor_app, ["show", "local/bench-actor", "--format", "plain"]) class ActorCLIRemoveSuite: """Benchmark actor remove throughput.""" def setup(self) -> None: self._mock_registry = MagicMock() self._patcher = patch( "cleveragents.cli.commands.actor._get_services", return_value=(MagicMock(), self._mock_registry), ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_remove(self) -> None: """Benchmark removing an actor.""" _runner.invoke(actor_app, ["remove", "local/bench-actor"])