"""ASV benchmarks for CLI --format rendering throughput. Measures the overhead of json, yaml, plain, table, and rich formats for both action and plan CLI commands. """ from __future__ import annotations import importlib import sys from datetime import datetime from pathlib import Path 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.action import app as action_app # noqa: E402 from cleveragents.cli.commands.plan import app as plan_app # noqa: E402 from cleveragents.domain.models.core.action import Action, ActionState # noqa: E402 from cleveragents.domain.models.core.plan import ( # noqa: E402 NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) _runner = CliRunner() _ULID = "01KHDE6WWS2171PWW3GJEBXZ8S" def _mock_action(name: str = "local/bench-action") -> Action: return Action( namespaced_name=NamespacedName.parse(name), description="Benchmark action", long_description=None, definition_of_done="Benchmarks pass", strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", reusable=True, read_only=False, state=ActionState.AVAILABLE, created_by=None, created_at=datetime.now(), updated_at=datetime.now(), ) def _mock_plan(name: str = "local/bench-plan") -> Plan: now = datetime.now() return Plan( identity=PlanIdentity(plan_id=_ULID), namespaced_name=NamespacedName.parse(name), description="Benchmark plan", definition_of_done="Benchmarks pass", action_name="local/bench-action", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", project_links=[ProjectLink(project_name="proj-a")], timestamps=PlanTimestamps(created_at=now, updated_at=now), created_by=None, reusable=True, read_only=False, ) class ActionListFormatSuite: """Benchmark action list --format throughput.""" params: list[str] = ["json", "yaml", "plain", "table", "rich"] # noqa: RUF012 param_names: list[str] = ["format"] # noqa: RUF012 def setup(self, fmt: str) -> None: self._svc = MagicMock() self._svc.list_actions.return_value = [ _mock_action(f"local/action-{i}") for i in range(20) ] self._patcher = patch( "cleveragents.cli.commands.action._get_lifecycle_service", return_value=self._svc, ) self._patcher.start() def teardown(self, fmt: str) -> None: self._patcher.stop() def time_list(self, fmt: str) -> None: _runner.invoke(action_app, ["list", "--format", fmt]) class PlanStatusFormatSuite: """Benchmark plan status --format throughput.""" params: list[str] = ["json", "yaml", "plain", "table", "rich"] # noqa: RUF012 param_names: list[str] = ["format"] # noqa: RUF012 def setup(self, fmt: str) -> None: self._svc = MagicMock() self._svc.get_plan.return_value = _mock_plan() self._patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._svc, ) self._patcher.start() def teardown(self, fmt: str) -> None: self._patcher.stop() def time_status(self, fmt: str) -> None: _runner.invoke(plan_app, ["status", _ULID, "--format", fmt])