cleveragents-core/benchmarks/devcontainer_handler_bench.py

"""ASV benchmarks for devcontainer handler auto-discovery throughput.

Measures the performance of:
- Devcontainer discovery on projects with many subdirectories
- Handler resolver cache performance for DevcontainerHandler
- Discovery result construction overhead
- is_trigger_type check throughput
"""

from __future__ import annotations

import json
import sys
import tempfile
from pathlib import Path
from typing import ClassVar

try:
    from cleveragents.resource.handlers.discovery import (
        DevcontainerDiscoveryResult,
        discover_devcontainers,
        is_trigger_type,
    )
    from cleveragents.resource.handlers.resolver import (
        clear_handler_cache,
        resolve_handler,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.resource.handlers.discovery import (
        DevcontainerDiscoveryResult,
        discover_devcontainers,
        is_trigger_type,
    )
    from cleveragents.resource.handlers.resolver import (
        clear_handler_cache,
        resolve_handler,
    )

_VALID_DC_JSON: str = json.dumps(
    {"name": "bench-devcontainer", "image": "ubuntu:latest"}
)


class TimeDiscoveryThroughput:
    """Benchmark auto-discovery throughput on projects with many subdirs."""

    timeout = 60
    params: ClassVar[list[int]] = [10, 50, 100]
    param_names: ClassVar[list[str]] = ["num_subdirs"]

    def setup(self, num_subdirs: int) -> None:
        """Create a temp directory with many subdirectories."""
        self.tmp_dir = tempfile.mkdtemp()
        base = Path(self.tmp_dir)
        # Create the devcontainer config
        dc_dir = base / ".devcontainer"
        dc_dir.mkdir()
        (dc_dir / "devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
        # Create many sibling directories (noise)
        for i in range(num_subdirs):
            (base / f"subdir_{i:04d}").mkdir()

    def teardown(self, num_subdirs: int) -> None:
        """Clean up temp directory."""
        import shutil

        shutil.rmtree(self.tmp_dir, ignore_errors=True)

    def time_discovery_with_subdirs(self, num_subdirs: int) -> None:
        """Time discovery scanning with many peer directories."""
        discover_devcontainers(self.tmp_dir, "git-checkout")


class TimeDiscoveryEmpty:
    """Benchmark discovery on directory with no devcontainer."""

    timeout = 30

    def setup(self) -> None:
        """Create empty temp directory."""
        self.tmp_dir = tempfile.mkdtemp()
        base = Path(self.tmp_dir)
        for i in range(50):
            (base / f"subdir_{i:04d}").mkdir()

    def teardown(self) -> None:
        """Clean up."""
        import shutil

        shutil.rmtree(self.tmp_dir, ignore_errors=True)

    def time_discovery_no_match(self) -> None:
        """Time discovery when no devcontainer exists."""
        discover_devcontainers(self.tmp_dir, "git-checkout")


class TimeHandlerResolver:
    """Benchmark handler resolver cache for DevcontainerHandler."""

    timeout = 30
    _ref = "cleveragents.resource.handlers.devcontainer:DevcontainerHandler"

    def setup(self) -> None:
        """Pre-warm resolver cache."""
        clear_handler_cache()
        resolve_handler(self._ref)

    def time_cached_resolution(self) -> None:
        """Time cached handler resolution (should be near-zero)."""
        resolve_handler(self._ref)

    def time_cold_resolution(self) -> None:
        """Time cold handler resolution (import + instantiate)."""
        clear_handler_cache()
        resolve_handler(self._ref)


class TimeIsTriggerType:
    """Benchmark is_trigger_type check throughput."""

    timeout = 10

    def time_trigger_type_positive(self) -> None:
        """Time is_trigger_type for a trigger type."""
        for _ in range(1000):
            is_trigger_type("git-checkout")

    def time_trigger_type_negative(self) -> None:
        """Time is_trigger_type for a non-trigger type."""
        for _ in range(1000):
            is_trigger_type("devcontainer-instance")


class TimeDiscoveryResult:
    """Benchmark DevcontainerDiscoveryResult construction."""

    timeout = 30

    def setup(self) -> None:
        """Create temp file for result construction."""
        self.tmp_dir = tempfile.mkdtemp()
        self.dc_file = Path(self.tmp_dir) / "devcontainer.json"
        self.dc_file.write_text(_VALID_DC_JSON, encoding="utf-8")
        self.config_data: dict[str, object] = json.loads(_VALID_DC_JSON)

    def teardown(self) -> None:
        """Clean up."""
        import shutil

        shutil.rmtree(self.tmp_dir, ignore_errors=True)

    def time_result_construction(self) -> None:
        """Time DevcontainerDiscoveryResult instantiation."""
        for _ in range(1000):
            DevcontainerDiscoveryResult(
                config_path=self.dc_file,
                config_data=self.config_data,
                parent_location=self.tmp_dir,
            )