"""ASV benchmarks for CLI argument parsing overhead. Measures the performance of: - Action CLI argument parsing (create, list, show, archive) - Plan CLI argument parsing (use with various flag combinations) - Plan lifecycle commands (execute, apply, status, cancel) These benchmarks focus on CLI overhead (argument parsing + routing) rather than service layer execution time, since the service is mocked. """ from __future__ import annotations import importlib import os import sys import tempfile 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 AutomationProfileProvenance, AutomationProfileRef, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) _runner = CliRunner() _PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S" _VALID_YAML = """\ name: local/bench-smoke-action description: Smoke bench action strategy_actor: openai/gpt-4 execution_actor: openai/gpt-4 definition_of_done: Benchmarks pass """ def _mock_action(name: str = "local/bench-smoke-action") -> Action: return Action( namespaced_name=NamespacedName.parse(name), description="Smoke bench 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-smoke-plan", phase: PlanPhase = PlanPhase.STRATEGIZE, state: ProcessingState = ProcessingState.QUEUED, project_links: list[ProjectLink] | None = None, ) -> Plan: now = datetime.now() return Plan( identity=PlanIdentity(plan_id=_PLAN_ULID), namespaced_name=NamespacedName.parse(name), description="Smoke bench plan", definition_of_done="Benchmarks pass", action_name="local/bench-smoke-action", phase=phase, processing_state=state, project_links=project_links or [], arguments={"target_coverage": 80}, arguments_order=["target_coverage"], automation_profile=AutomationProfileRef( profile_name="trusted", provenance=AutomationProfileProvenance.PLAN, ), strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", reusable=True, read_only=False, created_by=None, timestamps=PlanTimestamps(created_at=now, updated_at=now), ) class ActionCLIParsingSuite: """Benchmark action CLI argument parsing overhead.""" def setup(self) -> None: """Set up mock service and temp config file.""" self._mock_service = MagicMock() self._mock_service.create_action.return_value = _mock_action() self._mock_service.list_actions.return_value = [ _mock_action(f"local/action-{i}") for i in range(20) ] self._mock_service.get_action_by_name.return_value = _mock_action() self._mock_service.archive_action.return_value = _mock_action() self._patcher = patch( "cleveragents.cli.commands.action._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() # Create temp YAML file fd, self._yaml_path = tempfile.mkstemp(suffix=".yaml") with os.fdopen(fd, "w") as fh: fh.write(_VALID_YAML) def teardown(self) -> None: self._patcher.stop() if os.path.exists(self._yaml_path): os.unlink(self._yaml_path) def time_action_create(self) -> None: """Benchmark action create with --config.""" _runner.invoke(action_app, ["create", "--config", self._yaml_path]) def time_action_list(self) -> None: """Benchmark action list.""" _runner.invoke(action_app, ["list"]) def time_action_list_with_filters(self) -> None: """Benchmark action list with namespace and state filters.""" _runner.invoke( action_app, ["list", "--namespace", "local", "--state", "available"] ) def time_action_show(self) -> None: """Benchmark action show.""" _runner.invoke(action_app, ["show", "local/bench-smoke-action"]) def time_action_archive(self) -> None: """Benchmark action archive.""" _runner.invoke(action_app, ["archive", "local/bench-smoke-action"]) class PlanUseArgParsingSuite: """Benchmark plan use CLI argument parsing with various flag combinations.""" def setup(self) -> None: """Set up mock service.""" self._mock_service = MagicMock() self._mock_service.get_action_by_name.return_value = _mock_action() self._mock_service.use_action.return_value = _mock_plan( project_links=[ ProjectLink(project_name="proj-a"), ProjectLink(project_name="proj-b"), ] ) self._patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_use_minimal(self) -> None: """Benchmark plan use with minimal args.""" _runner.invoke(plan_app, ["use", "local/bench-smoke-action", "proj-a"]) def time_use_multi_project(self) -> None: """Benchmark plan use with multiple projects.""" _runner.invoke( plan_app, ["use", "local/bench-smoke-action", "proj-a", "proj-b", "proj-c"], ) def time_use_with_all_flags(self) -> None: """Benchmark plan use with all optional flags.""" _runner.invoke( plan_app, [ "use", "local/bench-smoke-action", "proj-a", "--automation-profile", "trusted", "--invariant", "No warnings", "--invariant", "Keep compat", "--strategy-actor", "openai/gpt-4", "--execution-actor", "openai/gpt-4", "--estimation-actor", "openai/gpt-4", "--invariant-actor", "openai/gpt-4", "--automation-level", "full_automation", "--arg", "target_coverage=80", "--arg", "verbose=true", ], ) def time_use_with_args_only(self) -> None: """Benchmark plan use with multiple --arg flags.""" _runner.invoke( plan_app, [ "use", "local/bench-smoke-action", "proj-a", "--arg", "coverage=80", "--arg", "verbose=true", "--arg", "ratio=3.14", ], ) class PlanLifecycleCmdParsingSuite: """Benchmark plan lifecycle command parsing overhead.""" def setup(self) -> None: """Set up mock service.""" self._mock_service = MagicMock() self._mock_service.execute_plan.return_value = _mock_plan( phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED ) self._mock_service.apply_plan.return_value = _mock_plan( phase=PlanPhase.APPLY, state=ProcessingState.QUEUED ) self._mock_service.get_plan.return_value = _mock_plan( project_links=[ProjectLink(project_name="proj-a", alias="api")] ) self._mock_service.cancel_plan.return_value = _mock_plan( phase=PlanPhase.STRATEGIZE, state=ProcessingState.CANCELLED ) self._mock_service.list_plans.return_value = [ _mock_plan(f"local/plan-{i}") for i in range(10) ] self._patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_execute(self) -> None: """Benchmark plan execute.""" _runner.invoke(plan_app, ["execute", _PLAN_ULID]) def time_apply(self) -> None: """Benchmark plan apply.""" _runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID]) def time_status_by_id(self) -> None: """Benchmark plan status with specific ID.""" _runner.invoke(plan_app, ["status", _PLAN_ULID]) def time_status_list_all(self) -> None: """Benchmark plan status listing all plans.""" _runner.invoke(plan_app, ["status"]) def time_cancel_with_reason(self) -> None: """Benchmark plan cancel with reason.""" _runner.invoke(plan_app, ["cancel", _PLAN_ULID, "--reason", "Benchmarking"]) def time_list(self) -> None: """Benchmark list.""" _runner.invoke(plan_app, ["list"]) def time_list_filtered(self) -> None: """Benchmark list with filters.""" _runner.invoke( plan_app, ["list", "--phase", "strategize", "--state", "queued"], )