"""ASV benchmarks for M3 decision tree, validation, and invariant CLI overhead. Measures the performance of: - Invariant add / list / remove CLI commands - Validation add / attach / detach CLI commands - Plan correct (dry-run) CLI command - Decision tree fixture loading - Invariant merge precedence computation """ from __future__ import annotations import importlib import json import sys import tempfile 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.invariant import app as invariant_app # noqa: E402 from cleveragents.cli.commands.plan import app as plan_app # noqa: E402 from cleveragents.cli.commands.validation import app as validation_app # noqa: E402 from cleveragents.domain.models.core.correction import ( # noqa: E402 CorrectionMode, CorrectionStatus, ) from cleveragents.domain.models.core.invariant import ( # noqa: E402 Invariant, InvariantScope, ) _runner = CliRunner() _PLAN_ULID = "01M3SM0KE00000000000000001" _DECISION_ULID = "01M3DEC1S10N0000000000001" _CORRECTION_ULID = "01M3C0RRECT10N00000000001" _ATTACHMENT_ULID = "01M3ATTACH0000000000000001" _INVARIANT_ULID = "01M3INVAR1ANT0000000000001" _FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m3" def _mock_invariant( *, text: str = "Never delete production data", scope: InvariantScope = InvariantScope.GLOBAL, source_name: str = "system", ) -> Invariant: return Invariant( id=_INVARIANT_ULID, text=text, scope=scope, source_name=source_name, ) class M3InvariantAddSuite: """Benchmark invariant add CLI command.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.add_invariant.return_value = _mock_invariant() self._patcher = patch( "cleveragents.cli.commands.invariant._get_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_invariant_add_global(self) -> None: """Benchmark invariant add with global scope.""" _runner.invoke( invariant_app, ["add", "--global", "Never delete production data", "--format", "plain"], ) def time_invariant_add_project(self) -> None: """Benchmark invariant add with project scope.""" _runner.invoke( invariant_app, [ "add", "--project", "local/m3-smoke-proj", "All API changes need tests", "--format", "plain", ], ) class M3InvariantListSuite: """Benchmark invariant list CLI command.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.list_invariants.return_value = [ _mock_invariant(), _mock_invariant( text="All API changes need tests", scope=InvariantScope.PROJECT, source_name="local/m3-smoke-proj", ), ] self._patcher = patch( "cleveragents.cli.commands.invariant._get_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_invariant_list(self) -> None: """Benchmark invariant list command.""" _runner.invoke(invariant_app, ["list", "--format", "plain"]) def time_invariant_list_with_project(self) -> None: """Benchmark invariant list with project filter.""" _runner.invoke( invariant_app, ["list", "--project", "local/m3-smoke-proj", "--format", "plain"], ) class M3ValidationAddSuite: """Benchmark validation add CLI command.""" def setup(self) -> None: self._mock_service = MagicMock() mock_validation = MagicMock() mock_validation.name = "local/coverage-check" mock_validation.as_cli_dict.return_value = { "name": "local/coverage-check", "description": "Check code coverage meets threshold", "source": "custom", "mode": "required", } self._mock_service.register_tool.return_value = mock_validation self._patcher = patch( "cleveragents.cli.commands.validation._get_tool_registry_service", return_value=self._mock_service, ) self._patcher.start() config_content = ( "name: local/coverage-check\n" "description: Check code coverage meets threshold\n" "source: custom\n" "mode: required\n" "code: |\n" " def run(inputs):\n" " return {'passed': inputs['coverage'] >= 80}\n" ) with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False, ) as tmp: tmp.write(config_content) self._config_path = tmp.name def teardown(self) -> None: self._patcher.stop() def time_validation_add(self) -> None: """Benchmark validation add from YAML config.""" _runner.invoke( validation_app, ["add", "--config", self._config_path, "--format", "plain"], ) class M3PlanCorrectSuite: """Benchmark plan correct (dry-run) CLI command.""" def setup(self) -> None: mock_correction_svc = MagicMock() mock_request = MagicMock() mock_request.correction_id = _CORRECTION_ULID mock_request.mode = CorrectionMode.REVERT mock_request.target_decision_id = _DECISION_ULID mock_request.guidance = "Use FastAPI instead" mock_request.status = CorrectionStatus.PENDING mock_impact = MagicMock() mock_impact.affected_decisions = [_DECISION_ULID] mock_impact.affected_files = ["src/main.py"] mock_impact.estimated_cost = "low" mock_impact.risk_level = "low" mock_correction_svc.request_correction.return_value = mock_request mock_correction_svc.analyze_impact.return_value = mock_impact self._patcher = patch( "cleveragents.application.services.correction_service.CorrectionService", return_value=mock_correction_svc, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_plan_correct_dry_run(self) -> None: """Benchmark plan correct dry-run command.""" _runner.invoke( plan_app, [ "correct", _DECISION_ULID, "--mode", "revert", "--guidance", "Use FastAPI instead", "--dry-run", "--plan", _PLAN_ULID, "--format", "plain", ], ) class M3FixtureLoadSuite: """Benchmark loading M3 fixture files.""" def time_load_decision_tree_fixture(self) -> None: """Benchmark loading decision_tree_outputs.json fixture.""" fixture_path = _FIXTURES_DIR / "decision_tree_outputs.json" with open(fixture_path) as f: json.load(f) def time_load_validation_attachments_fixture(self) -> None: """Benchmark loading validation_attachments.json fixture.""" fixture_path = _FIXTURES_DIR / "validation_attachments.json" with open(fixture_path) as f: json.load(f) def time_load_invariant_configs_fixture(self) -> None: """Benchmark loading invariant_configs.json fixture.""" fixture_path = _FIXTURES_DIR / "invariant_configs.json" with open(fixture_path) as f: json.load(f)