cleveragents-core/benchmarks/cli_format_bench.py

"""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
    AutomationLevel,
    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,
        automation_level=AutomationLevel.MANUAL,
        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"]
    param_names: list[str] = ["format"]

    def setup(self) -> 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) -> 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"]
    param_names: list[str] = ["format"]

    def setup(self) -> 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) -> None:
        self._patcher.stop()

    def time_status(self, fmt: str) -> None:
        _runner.invoke(plan_app, ["status", _ULID, "--format", fmt])