cleveragents-core/benchmarks/bench_subprocess_overhead.py

"""ASV benchmarks for subprocess and per-feature overhead.

Tracks subprocess-related metrics that affect test suite runtime:
- Feature file count (proxy for subprocess count in old model)
- Per-feature startup cost estimation via single-feature dry parse
- Parallel worker pool sizing computation
- Environment variable propagation overhead
"""

from __future__ import annotations

import math
import os
import sys
from pathlib import Path

_PROJECT_ROOT = Path(__file__).resolve().parents[1]
_FEATURES_DIR = _PROJECT_ROOT / "features"
_SRC = str(_PROJECT_ROOT / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)


class SubprocessCountSuite:
    """Track subprocess count proxies.

    In the old subprocess-per-feature model, each ``.feature`` file
    spawned a separate Python interpreter.  These metrics track the
    scale factor to quantify the benefit of the in-process runner.
    """

    timeout = 60

    def track_subprocess_count(self) -> int:
        """Count feature files that would be run as subprocesses.

        Under the legacy model, each ``.feature`` file spawned one
        subprocess.  This metric tracks how many would have been
        created, serving as a regression signal for test suite growth.
        """
        return len([f for f in os.listdir(_FEATURES_DIR) if f.endswith(".feature")])

    def track_recursive_feature_count(self) -> int:
        """Count all .feature files recursively (including subdirectories)."""
        return len(list(_FEATURES_DIR.rglob("*.feature")))

    def track_step_definition_count(self) -> int:
        """Count step definition Python files."""
        steps_dir = _FEATURES_DIR / "steps"
        if steps_dir.exists():
            return len(list(steps_dir.glob("*.py")))
        return 0


SubprocessCountSuite.track_subprocess_count.unit = "subprocesses"
SubprocessCountSuite.track_recursive_feature_count.unit = "features"
SubprocessCountSuite.track_step_definition_count.unit = "files"


class PerFeatureOverheadSuite:
    """Benchmark per-feature startup and parsing overhead.

    Measures operations that execute once per feature file during
    a test run — parsing, path resolution, and environment setup.
    """

    timeout = 120

    def setup(self) -> None:
        """Discover feature files for benchmarking."""
        self.feature_paths: list[Path] = sorted(_FEATURES_DIR.rglob("*.feature"))
        self.feature_texts: list[str] = []
        # Read a sample of features (first 10) to avoid excessive I/O
        for fp in self.feature_paths[:10]:
            self.feature_texts.append(fp.read_text(encoding="utf-8"))

    def time_single_feature_read(self) -> None:
        """Benchmark reading a single feature file from disk."""
        if self.feature_paths:
            self.feature_paths[0].read_text(encoding="utf-8")

    def time_feature_line_count(self) -> None:
        """Benchmark counting lines in sampled feature files."""
        for text in self.feature_texts:
            len(text.splitlines())

    def time_feature_scenario_extraction(self) -> None:
        """Benchmark extracting scenario names from feature text."""
        for text in self.feature_texts:
            _scenarios = [
                line.strip()
                for line in text.splitlines()
                if line.strip().startswith(("Scenario:", "Scenario Outline:"))
            ]

    def time_feature_tag_extraction(self) -> None:
        """Benchmark extracting tags from feature files."""
        for text in self.feature_texts:
            _tags = [
                line.strip()
                for line in text.splitlines()
                if line.strip().startswith("@")
            ]

    def time_path_resolution_all_features(self) -> None:
        """Benchmark resolving absolute paths for all feature files."""
        for fp in self.feature_paths:
            fp.resolve()


class WorkerPoolSizingSuite:
    """Benchmark worker pool sizing and chunk distribution.

    The in-process parallel runner computes optimal chunk sizes
    based on CPU count and feature count.  This suite ensures
    the computation remains efficient as the suite grows.
    """

    timeout = 60

    def setup(self) -> None:
        """Prepare feature list and CPU configurations."""
        self.feature_count: int = len(list(_FEATURES_DIR.rglob("*.feature")))
        self.cpu_configs: list[int] = [1, 2, 4, 8, 16, 32, 64]

    def time_pool_sizing_computation(self) -> None:
        """Benchmark pool size and chunk computation for all CPU configs."""
        for cpus in self.cpu_configs:
            processes = min(cpus, self.feature_count)
            chunk_size = max(1, math.ceil(self.feature_count / processes))
            _chunk_count = math.ceil(self.feature_count / chunk_size)

    def time_environment_variable_lookup(self) -> None:
        """Benchmark environment variable lookups used by the test runner."""
        _ = os.environ.get("TEST_PROCESSES")
        _ = os.environ.get("BEHAVE_PARALLEL_COVERAGE")
        _ = os.environ.get("CLEVERAGENTS_TEMPLATE_DB")
        _ = os.environ.get("PYTHONPATH")
        _ = os.environ.get("NO_COLOR")

    def track_features_per_cpu(self) -> float:
        """Track features-per-CPU ratio for load balancing insight."""
        cpus = os.cpu_count() or 1
        return round(self.feature_count / cpus, 2)

    def track_optimal_chunk_size(self) -> int:
        """Track the optimal chunk size for current CPU count."""
        cpus = os.cpu_count() or 1
        return max(1, math.ceil(self.feature_count / cpus))


WorkerPoolSizingSuite.track_features_per_cpu.unit = "features/cpu"
WorkerPoolSizingSuite.track_optimal_chunk_size.unit = "features/chunk"