Files

125 lines
3.8 KiB
Python

"""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])