perf(tests): add ASV benchmarks for test suite runtime regression tracking
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 14s
CI / quality (pull_request) Successful in 16s
CI / build (pull_request) Successful in 16s
CI / typecheck (pull_request) Successful in 33s
CI / security (pull_request) Successful in 45s
CI / unit_tests (pull_request) Successful in 1m52s
CI / docker (pull_request) Successful in 9s
CI / integration_tests (pull_request) Successful in 2m57s
CI / coverage (pull_request) Successful in 3m31s
CI / benchmark-regression (pull_request) Successful in 23m16s
CI / lint (push) Successful in 13s
CI / quality (push) Successful in 16s
CI / build (push) Successful in 19s
CI / security (push) Successful in 29s
CI / typecheck (push) Successful in 34s
CI / benchmark-regression (push) Has been skipped
CI / unit_tests (push) Successful in 1m49s
CI / docker (push) Successful in 38s
CI / integration_tests (push) Successful in 2m46s
CI / coverage (push) Successful in 3m38s
CI / benchmark-publish (push) Has been cancelled

Add ASV benchmark modules to track test suite performance over time:
- bench_unit_tests.py: Feature discovery, step module loading, parallel
  chunk computation, behave config parsing, and test suite metrics
- bench_coverage_report.py: Coverage collection pipeline timing and
  report generation overhead
- bench_subprocess_overhead.py: Subprocess count tracking and per-feature
  startup cost measurement

These benchmarks establish baselines for the post-optimization state
and enable CI-driven regression detection.

ISSUES CLOSED: #486
This commit was merged in pull request #503.
This commit is contained in:
2026-03-02 06:30:00 +00:00
committed by Forgejo
parent 86e245c585
commit 69f5b2e3e6
3 changed files with 559 additions and 0 deletions
+206
View File
@@ -0,0 +1,206 @@
"""ASV benchmarks for coverage collection pipeline runtime.
Measures the performance of operations that contribute to the
``nox -s coverage_report`` wall-clock time:
- Slipcover configuration extraction from noxfile / pyproject
- Coverage threshold validation and percentage parsing
- Coverage JSON report parsing and summary extraction
- XML report generation overhead (file I/O proxy)
"""
from __future__ import annotations
import ast
import json
import re
from pathlib import Path
_PROJECT_ROOT = Path(__file__).resolve().parents[1]
_NOXFILE_PATH = _PROJECT_ROOT / "noxfile.py"
_PYPROJECT_PATH = _PROJECT_ROOT / "pyproject.toml"
_COVERAGE_JSON = _PROJECT_ROOT / "build" / "coverage.json"
_COVERAGE_XML = _PROJECT_ROOT / "build" / "coverage.xml"
# Threshold used by the project for coverage enforcement.
_COVERAGE_THRESHOLD = 97
class CoverageConfigExtractionSuite:
"""Benchmark extracting coverage configuration from project files.
The coverage report session reads noxfile.py and pyproject.toml to
determine source paths, omit patterns, and the fail-under threshold.
"""
timeout = 60
def setup(self) -> None:
"""Read configuration files once for reuse."""
if _NOXFILE_PATH.exists():
self.noxfile_text: str = _NOXFILE_PATH.read_text()
else:
self.noxfile_text = ""
if _PYPROJECT_PATH.exists():
self.pyproject_text: str = _PYPROJECT_PATH.read_text()
else:
self.pyproject_text = ""
def time_parse_noxfile_ast(self) -> None:
"""Benchmark full AST parse of noxfile.py."""
if self.noxfile_text:
ast.parse(self.noxfile_text)
def time_extract_coverage_threshold(self) -> None:
"""Benchmark extracting COVERAGE_THRESHOLD from noxfile AST."""
if self.noxfile_text:
tree = ast.parse(self.noxfile_text)
for node in ast.walk(tree):
if isinstance(node, ast.Assign):
for target in node.targets:
if (
isinstance(target, ast.Name)
and target.id == "COVERAGE_THRESHOLD"
and isinstance(node.value, ast.Constant)
):
_ = node.value.value
def time_extract_omit_patterns(self) -> None:
"""Benchmark extracting omit patterns from noxfile text."""
if self.noxfile_text:
re.findall(r'"(\*[^"]+)"', self.noxfile_text)
def time_parse_pyproject_coverage_section(self) -> None:
"""Benchmark extracting [tool.coverage] from pyproject.toml."""
if self.pyproject_text:
import tomllib
data = tomllib.loads(self.pyproject_text)
_ = data.get("tool", {}).get("coverage", {})
def time_extract_source_paths(self) -> None:
"""Benchmark extracting coverage source paths from pyproject."""
if self.pyproject_text:
import tomllib
data = tomllib.loads(self.pyproject_text)
_ = (
data.get("tool", {})
.get("coverage", {})
.get("run", {})
.get("source", [])
)
def time_extract_fail_under_regex(self) -> None:
"""Benchmark regex extraction of --fail-under value from noxfile."""
if self.noxfile_text:
re.findall(r"--fail-under=(\d+)", self.noxfile_text)
class CoverageReportParsingSuite:
"""Benchmark parsing coverage report outputs.
After slipcover runs, the coverage pipeline parses JSON and XML
reports to extract totals and enforce thresholds.
"""
timeout = 60
def setup(self) -> None:
"""Load coverage report data if available."""
self.json_text: str = ""
self.xml_text: str = ""
if _COVERAGE_JSON.exists():
self.json_text = _COVERAGE_JSON.read_text()
if _COVERAGE_XML.exists():
self.xml_text = _COVERAGE_XML.read_text()
def time_parse_coverage_json(self) -> None:
"""Benchmark parsing the slipcover JSON report."""
if self.json_text:
json.loads(self.json_text)
def time_extract_coverage_percentage(self) -> None:
"""Benchmark extracting coverage percentage from JSON report."""
if self.json_text:
payload = json.loads(self.json_text)
summary = payload.get("summary") or payload.get("totals") or {}
for key in (
"percent_covered",
"percent_covered_display",
"line_coverage_percent",
):
value = summary.get(key)
if value is not None:
break
def time_threshold_comparison(self) -> None:
"""Benchmark the threshold enforcement logic."""
if self.json_text:
payload = json.loads(self.json_text)
summary = payload.get("summary") or payload.get("totals") or {}
percent_value = None
for key in (
"percent_covered",
"percent_covered_display",
"line_coverage_percent",
):
percent_value = summary.get(key)
if percent_value is not None:
break
if percent_value is None:
covered = summary.get("lines_covered") or summary.get("covered_lines")
total = summary.get("lines") or summary.get("total_lines")
if covered is not None and total:
percent_value = (float(covered) / float(total)) * 100.0
if isinstance(percent_value, str):
percent_value = percent_value.strip().rstrip("%")
if percent_value is not None:
pct = round(float(percent_value), 1)
_ = pct >= _COVERAGE_THRESHOLD
def time_parse_coverage_xml(self) -> None:
"""Benchmark parsing the coverage XML report."""
if self.xml_text:
import xml.etree.ElementTree as ET
ET.fromstring(self.xml_text)
class CoverageFileSuite:
"""Benchmark coverage-related file I/O operations."""
timeout = 60
def setup(self) -> None:
"""Prepare paths for file I/O benchmarks."""
self.build_dir: Path = _PROJECT_ROOT / "build"
def time_discover_slipcover_files(self) -> None:
"""Benchmark discovering .slipcover.*.json files in build/."""
if self.build_dir.exists():
list(self.build_dir.glob(".slipcover.*.json"))
def time_discover_coverage_artifacts(self) -> None:
"""Benchmark discovering all coverage-related artifacts."""
if self.build_dir.exists():
_json = list(self.build_dir.glob("coverage*.json"))
_xml = list(self.build_dir.glob("coverage*.xml"))
_txt = list(self.build_dir.glob("coverage*.txt"))
def track_source_file_count(self) -> int:
"""Track number of Python source files under coverage."""
src_dir = _PROJECT_ROOT / "src"
if src_dir.exists():
return len(list(src_dir.rglob("*.py")))
return 0
def track_source_total_bytes(self) -> int:
"""Track total byte size of Python source files."""
src_dir = _PROJECT_ROOT / "src"
if src_dir.exists():
return sum(f.stat().st_size for f in src_dir.rglob("*.py"))
return 0
setattr(CoverageFileSuite.track_source_file_count, "unit", "files")
setattr(CoverageFileSuite.track_source_total_bytes, "unit", "bytes")
+153
View File
@@ -0,0 +1,153 @@
"""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
setattr(SubprocessCountSuite.track_subprocess_count, "unit", "subprocesses")
setattr(SubprocessCountSuite.track_recursive_feature_count, "unit", "features")
setattr(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))
setattr(WorkerPoolSizingSuite.track_features_per_cpu, "unit", "features/cpu")
setattr(WorkerPoolSizingSuite.track_optimal_chunk_size, "unit", "features/chunk")
+200
View File
@@ -0,0 +1,200 @@
"""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"))
setattr(StepModuleDiscoverySuite.track_step_file_count, "unit", "files")
setattr(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
setattr(TrackTestSuiteMetrics.track_feature_file_count, "unit", "features")
setattr(TrackTestSuiteMetrics.track_total_feature_bytes, "unit", "bytes")
setattr(TrackTestSuiteMetrics.track_scenario_line_count, "unit", "scenarios")