cleveragents-core/benchmarks/bench_unit_tests.py

"""ASV benchmarks for BDD unit test suite runtime.

Measures the performance of test-infrastructure operations that
contribute to overall unit-test wall-clock time:
- Feature file discovery and glob performance
- Behave configuration parsing overhead
- Step-definition module import cost
- Parallel chunk computation for the in-process runner
"""

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"
_STEPS_DIR = _FEATURES_DIR / "steps"
_SRC = str(_PROJECT_ROOT / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)


class FeatureDiscoverySuite:
    """Benchmark feature file discovery performance.

    Measures the I/O and glob cost of discovering all ``.feature``
    files, which is the first step of every test run.
    """

    timeout = 120

    def setup(self) -> None:
        """Cache the features directory path."""
        self.features_dir: Path = _FEATURES_DIR

    def time_glob_all_features(self) -> None:
        """Benchmark recursive glob for all .feature files."""
        list(self.features_dir.rglob("*.feature"))

    def time_sorted_feature_list(self) -> None:
        """Benchmark sorted feature discovery (as used by the runner)."""
        sorted(str(fp) for fp in self.features_dir.rglob("*.feature"))

    def time_listdir_feature_root(self) -> None:
        """Benchmark flat listing of the features directory."""
        [entry for entry in os.listdir(self.features_dir) if entry.endswith(".feature")]

    def time_stat_all_features(self) -> None:
        """Benchmark stat calls on every discovered feature file."""
        for fp in self.features_dir.rglob("*.feature"):
            fp.stat()


class StepModuleDiscoverySuite:
    """Benchmark step-definition module discovery.

    Step definitions are loaded once per process, but the import
    cost contributes to startup overhead.  Tracking this prevents
    accidental regressions from heavy new imports.
    """

    timeout = 120

    def setup(self) -> None:
        """Cache the steps directory path."""
        self.steps_dir: Path = _STEPS_DIR

    def time_discover_step_files(self) -> None:
        """Benchmark discovering all step definition Python files."""
        list(self.steps_dir.glob("*.py"))

    def time_read_step_file_sizes(self) -> None:
        """Benchmark reading sizes of all step files (I/O pressure proxy)."""
        total = 0
        for py_file in self.steps_dir.glob("*.py"):
            total += py_file.stat().st_size

    def track_step_file_count(self) -> int:
        """Track the number of step definition files."""
        return len(list(self.steps_dir.glob("*.py")))

    def track_step_total_bytes(self) -> int:
        """Track total byte size of all step definition files."""
        return sum(f.stat().st_size for f in self.steps_dir.glob("*.py"))


StepModuleDiscoverySuite.track_step_file_count.unit = "files"
StepModuleDiscoverySuite.track_step_total_bytes.unit = "bytes"


class ParallelChunkSuite:
    """Benchmark parallel chunk computation.

    The in-process behave-parallel runner splits feature files into
    chunks for multiprocessing.  This suite tracks the computation
    cost to ensure it scales linearly.
    """

    timeout = 60

    def setup(self) -> None:
        """Discover features and simulate CPU counts."""
        self.feature_paths: list[str] = sorted(
            str(fp) for fp in _FEATURES_DIR.rglob("*.feature")
        )
        self.cpu_counts: list[int] = [1, 2, 4, 8, 16, 32]

    def time_chunk_split_default(self) -> None:
        """Benchmark default chunk splitting (os.cpu_count)."""
        processes = os.cpu_count() or 1
        chunk_size = max(1, math.ceil(len(self.feature_paths) / processes))
        _chunks = [
            self.feature_paths[i : i + chunk_size]
            for i in range(0, len(self.feature_paths), chunk_size)
        ]

    def time_chunk_split_all_counts(self) -> None:
        """Benchmark chunk splitting across various CPU counts."""
        for processes in self.cpu_counts:
            chunk_size = max(1, math.ceil(len(self.feature_paths) / processes))
            _chunks = [
                self.feature_paths[i : i + chunk_size]
                for i in range(0, len(self.feature_paths), chunk_size)
            ]

    def time_feature_path_sorting(self) -> None:
        """Benchmark sorting feature paths (runner hot path)."""
        sorted(self.feature_paths)


class BehaveConfigSuite:
    """Benchmark behave configuration parsing.

    The ``behave.ini`` file is parsed at the start of every test
    run.  This suite tracks the cost.
    """

    timeout = 60

    def setup(self) -> None:
        """Read behave.ini content."""
        behave_ini = _PROJECT_ROOT / "behave.ini"
        if behave_ini.exists():
            self.ini_text: str = behave_ini.read_text()
        else:
            self.ini_text = ""

    def time_parse_behave_ini(self) -> None:
        """Benchmark parsing behave.ini with configparser."""
        if self.ini_text:
            import configparser

            parser = configparser.ConfigParser()
            parser.read_string(self.ini_text)

    def time_read_behave_ini(self) -> None:
        """Benchmark raw file read of behave.ini."""
        behave_ini = _PROJECT_ROOT / "behave.ini"
        if behave_ini.exists():
            behave_ini.read_text()


class TrackTestSuiteMetrics:
    """Track high-level test suite metrics over time.

    These ``track_*`` methods return numeric values that ASV records
    as time-series data, enabling regression detection on the size
    and shape of the test suite itself.
    """

    timeout = 60

    def track_feature_file_count(self) -> int:
        """Track total number of .feature files."""
        return len(list(_FEATURES_DIR.rglob("*.feature")))

    def track_total_feature_bytes(self) -> int:
        """Track total byte size of all feature files."""
        return sum(f.stat().st_size for f in _FEATURES_DIR.rglob("*.feature"))

    def track_scenario_line_count(self) -> int:
        """Track approximate scenario count by counting Scenario: lines."""
        count = 0
        for fp in _FEATURES_DIR.rglob("*.feature"):
            text = fp.read_text(encoding="utf-8")
            for line in text.splitlines():
                stripped = line.strip()
                if stripped.startswith("Scenario:") or stripped.startswith(
                    "Scenario Outline:"
                ):
                    count += 1
        return count


TrackTestSuiteMetrics.track_feature_file_count.unit = "features"
TrackTestSuiteMetrics.track_total_feature_bytes.unit = "bytes"
TrackTestSuiteMetrics.track_scenario_line_count.unit = "scenarios"