"""ASV benchmarks for CLI Robot flow fixture setup overhead. Measures the performance overhead of: - Fixture setup for action creation (YAML write + parse) - Mock service initialisation for plan lifecycle commands - ChangeSet store creation and entry recording - Full lifecycle fixture setup (action + plan + changeset + sandbox) """ from __future__ import annotations import os import sys import tempfile from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch try: from cleveragents.cli.commands.action import app as action_app from cleveragents.cli.commands.plan import app as plan_app from cleveragents.domain.models.core.action import Action, ActionState from cleveragents.domain.models.core.change import ( ChangeEntry, ChangeOperation, InMemoryChangeSetStore, ) from cleveragents.domain.models.core.plan import ( AutomationProfileProvenance, AutomationProfileRef, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.cli.commands.action import app as action_app from cleveragents.cli.commands.plan import app as plan_app from cleveragents.domain.models.core.action import Action, ActionState from cleveragents.domain.models.core.change import ( ChangeEntry, ChangeOperation, InMemoryChangeSetStore, ) from cleveragents.domain.models.core.plan import ( AutomationProfileProvenance, AutomationProfileRef, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) from typer.testing import CliRunner _PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZBN" _VALID_YAML = """\ name: local/bench-lifecycle-action description: Benchmark lifecycle action strategy_actor: openai/gpt-4 execution_actor: openai/gpt-4 definition_of_done: Benchmarks pass """ _runner = CliRunner() def _mock_action(name: str = "local/bench-lifecycle-action") -> Action: """Create a minimal Action for benchmarks.""" return Action( namespaced_name=NamespacedName.parse(name), description="Benchmark lifecycle action", long_description=None, definition_of_done="Benchmarks pass", strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", state=ActionState.AVAILABLE, reusable=True, read_only=False, created_at=datetime.now(), updated_at=datetime.now(), created_by=None, ) def _mock_plan( phase: PlanPhase = PlanPhase.STRATEGIZE, state: ProcessingState = ProcessingState.QUEUED, ) -> Plan: """Create a minimal Plan for benchmarks.""" now = datetime.now() return Plan( identity=PlanIdentity(plan_id=_PLAN_ULID), namespaced_name=NamespacedName.parse("local/bench-plan"), description="Benchmark plan", definition_of_done="Benchmarks pass", action_name="local/bench-lifecycle-action", phase=phase, processing_state=state, project_links=[ProjectLink(project_name="proj-bench")], 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 CLIRobotFixtureSetupSuite: """Benchmark fixture setup overhead for CLI Robot flow tests.""" def setup(self) -> None: """Prepare reusable mock objects.""" self._mock_service = MagicMock() self._mock_service.create_action.return_value = _mock_action() self._mock_service.get_action_by_name.return_value = _mock_action() self._mock_service.use_action.return_value = _mock_plan() 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 ) def time_yaml_fixture_write(self) -> None: """Benchmark writing a YAML fixture file.""" fd, path = tempfile.mkstemp(suffix=".yaml") with os.fdopen(fd, "w") as fh: fh.write(_VALID_YAML) Path(path).unlink(missing_ok=True) def time_mock_service_creation(self) -> None: """Benchmark creating and configuring a mock service.""" svc = MagicMock() svc.create_action.return_value = _mock_action() svc.get_action_by_name.return_value = _mock_action() svc.use_action.return_value = _mock_plan() def time_changeset_store_creation(self) -> None: """Benchmark creating an InMemoryChangeSetStore and starting a changeset.""" store = InMemoryChangeSetStore() store.start(_PLAN_ULID) def time_changeset_entry_recording(self) -> None: """Benchmark recording a change entry in the store.""" store = InMemoryChangeSetStore() cid = store.start(_PLAN_ULID) store.record( cid, ChangeEntry( plan_id=_PLAN_ULID, resource_id="res-bench", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path="src/bench.py", ), ) class CLIRobotLifecycleSuite: """Benchmark the full CLI lifecycle flow with mocked service.""" def setup(self) -> None: """Prepare mock service and YAML file.""" self._mock_service = MagicMock() self._mock_service.create_action.return_value = _mock_action() self._mock_service.get_action_by_name.return_value = _mock_action() self._mock_service.use_action.return_value = _mock_plan() 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 ) fd, self._yaml_path = tempfile.mkstemp(suffix=".yaml") with os.fdopen(fd, "w") as fh: fh.write(_VALID_YAML) self._action_patcher = patch( "cleveragents.cli.commands.action._get_lifecycle_service", return_value=self._mock_service, ) self._plan_patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._action_patcher.start() self._plan_patcher.start() def teardown(self) -> None: """Clean up patchers and temp file.""" self._action_patcher.stop() self._plan_patcher.stop() Path(self._yaml_path).unlink(missing_ok=True) def time_action_create(self) -> None: """Benchmark action create via CLI.""" _runner.invoke(action_app, ["create", "--config", self._yaml_path]) def time_plan_use(self) -> None: """Benchmark plan use via CLI.""" _runner.invoke(plan_app, ["use", "local/bench-lifecycle-action", "proj-bench"]) def time_plan_execute(self) -> None: """Benchmark plan execute via CLI.""" _runner.invoke(plan_app, ["execute", _PLAN_ULID]) def time_plan_apply(self) -> None: """Benchmark plan apply via CLI.""" _runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID]) def time_full_lifecycle_flow(self) -> None: """Benchmark the full create -> use -> execute -> apply flow.""" _runner.invoke(action_app, ["create", "--config", self._yaml_path]) _runner.invoke(plan_app, ["use", "local/bench-lifecycle-action", "proj-bench"]) _runner.invoke(plan_app, ["execute", _PLAN_ULID]) _runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])