"""ASV benchmarks for Plan CLI command throughput. Measures the performance of: - Plan use with multiple projects and flags - Lifecycle-list with various filters - Plan status rendering - Plan cancel with reason """ 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.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" 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", project_links: list[ProjectLink] | None = None, ) -> Plan: now = datetime.now() return Plan( identity=PlanIdentity(plan_id=_PLAN_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, 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", estimation_actor="openai/gpt-4", invariant_actor="openai/gpt-4", reusable=True, read_only=False, created_by=None, timestamps=PlanTimestamps(created_at=now, updated_at=now), ) class PlanCLIUseSuite: """Benchmark plan use command throughput.""" 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_with_projects(self) -> None: """Benchmark plan use with multiple projects.""" _runner.invoke( plan_app, ["use", "local/bench-action", "proj-a", "proj-b", "--arg", "coverage=80"], ) def time_use_with_all_flags(self) -> None: """Benchmark plan use with all flags.""" _runner.invoke( plan_app, [ "use", "local/bench-action", "proj-a", "--automation-profile", "trusted", "--invariant", "No warnings", "--strategy-actor", "openai/gpt-4", "--execution-actor", "openai/gpt-4", "--arg", "coverage=80", ], ) class PlanCLIListSuite: """Benchmark plan list throughput.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.list_plans.return_value = [ _mock_plan(f"local/plan-{i}") for i in range(50) ] 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_list_all(self) -> None: """Benchmark listing all plans.""" _runner.invoke(plan_app, ["list"]) def time_list_with_phase(self) -> None: """Benchmark listing with phase filter.""" _runner.invoke(plan_app, ["list", "--phase", "strategize"]) def time_list_with_state(self) -> None: """Benchmark listing with state filter.""" _runner.invoke(plan_app, ["list", "--state", "queued"]) class PlanCLIStatusSuite: """Benchmark plan status rendering.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.get_plan.return_value = _mock_plan( project_links=[ProjectLink(project_name="proj-a", alias="api")] ) 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_status(self) -> None: """Benchmark rendering plan status.""" _runner.invoke(plan_app, ["status", _PLAN_ULID])