perf(tests): replace behave-parallel subprocess model with in-process parallelism #490
+292
-215
@@ -1,8 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tarfile
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
import nox
|
||||
@@ -13,11 +11,7 @@ SUPPORTED_PYTHONS = ["3.13"]
|
||||
nox.options.reuse_existing_virtualenvs = True
|
||||
nox.options.error_on_external_run = True
|
||||
|
||||
BEHAVE_PARALLEL_VERSION = "1.2.4a1"
|
||||
BEHAVE_PARALLEL_URL = (
|
||||
"https://files.pythonhosted.org/packages/05/9d/22f74dd77bc4fa85d391564a232c49b4e99cfdeac7bfdee8151ea4606632/"
|
||||
f"behave-parallel-{BEHAVE_PARALLEL_VERSION}.tar.gz"
|
||||
)
|
||||
BEHAVE_PARALLEL_VERSION = "2.0.0"
|
||||
|
||||
|
||||
def _default_processes() -> int:
|
||||
@@ -88,58 +82,193 @@ def _create_template_db(session: nox.Session) -> str:
|
||||
|
||||
|
||||
def _install_behave_parallel(session: nox.Session) -> None:
|
||||
"""Install behave-parallel with a custom CLI wrapper."""
|
||||
"""Install behave-parallel with an in-process parallel runner.
|
||||
|
||||
Instead of spawning one subprocess per feature file (the old model),
|
||||
the new CLI runs features in-process via behave's Python API. Step
|
||||
definitions and environment hooks are loaded once; each feature file
|
||||
is then parsed and executed without interpreter startup overhead.
|
||||
|
||||
Parallel execution uses ``multiprocessing.Pool`` with the ``fork``
|
||||
start method so that already-imported modules are shared (copy-on-write)
|
||||
across workers.
|
||||
"""
|
||||
tmp_dir = Path(session.create_tmp())
|
||||
archive_path = tmp_dir / "behave-parallel.tar.gz"
|
||||
with (
|
||||
urllib.request.urlopen(BEHAVE_PARALLEL_URL) as response,
|
||||
archive_path.open("wb") as target,
|
||||
):
|
||||
target.write(response.read())
|
||||
source_dir = tmp_dir / "behave-parallel-inprocess"
|
||||
source_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with tarfile.open(archive_path, "r:gz") as tar:
|
||||
tar.extractall(tmp_dir)
|
||||
|
||||
source_dir = next(tmp_dir.glob("behave-parallel-*/"))
|
||||
|
||||
# Recreate package with a parallel-aware CLI that aggregates summaries
|
||||
pkg_dir = source_dir / "behave_parallel"
|
||||
pkg_dir.mkdir(parents=True, exist_ok=True)
|
||||
(pkg_dir / "__init__.py").write_text("\n")
|
||||
(pkg_dir / "cli.py").write_text(
|
||||
"""
|
||||
(pkg_dir / "cli.py").write_text(_BEHAVE_PARALLEL_CLI_SOURCE)
|
||||
|
||||
setup_path = source_dir / "setup.py"
|
||||
setup_path.write_text(
|
||||
"from setuptools import find_packages, setup\n"
|
||||
"setup(\n"
|
||||
' name="behave-parallel",\n'
|
||||
f' version="{BEHAVE_PARALLEL_VERSION}",\n'
|
||||
" packages=find_packages(),\n"
|
||||
' install_requires=["behave>=1.2.6"],\n'
|
||||
" entry_points={\n"
|
||||
' "console_scripts": '
|
||||
'["behave-parallel=behave_parallel.cli:main"],\n'
|
||||
" },\n"
|
||||
")\n"
|
||||
)
|
||||
|
||||
session.install("setuptools", "wheel")
|
||||
session.install(str(source_dir))
|
||||
|
||||
|
||||
# The in-process behave-parallel CLI source is kept as a module-level
|
||||
# constant so the noxfile itself stays readable and ``ruff`` can still
|
||||
# lint the surrounding Python without tripping over a giant raw string
|
||||
# embedded inside a function body.
|
||||
_BEHAVE_PARALLEL_CLI_SOURCE = r'''
|
||||
"""In-process parallel behave runner.
|
||||
|
||||
Replaces the old subprocess-per-feature model with direct use of
|
||||
behave's ``Runner`` API. Step definitions and environment hooks are
|
||||
loaded once per process; feature files are parsed and executed without
|
||||
Python interpreter startup overhead.
|
||||
|
||||
Parallelism modes
|
||||
-----------------
|
||||
* **Sequential** (``--processes 1`` or ``BEHAVE_PARALLEL_COVERAGE=1``):
|
||||
All features run in a single ``Runner.run()`` call.
|
||||
* **Parallel** (``--processes N``, N > 1, no coverage):
|
||||
Features are split into *N* equal-size chunks. A
|
||||
``multiprocessing.Pool`` with the ``fork`` start method dispatches
|
||||
each chunk to a worker that creates its own ``Runner`` (hooks and
|
||||
step definitions are re-loaded cheaply because all heavy modules are
|
||||
already in memory from the parent).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import concurrent.futures
|
||||
import io
|
||||
import multiprocessing
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
import time
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from pathlib import Path
|
||||
from typing import Dict, Iterable, List, Tuple
|
||||
|
||||
DEFAULT_FEATURE_ROOT = "features/"
|
||||
SUMMARY_PATTERNS = {
|
||||
"features": re.compile(
|
||||
r"(\\d+)\\s+feature(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
|
||||
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
|
||||
),
|
||||
"scenarios": re.compile(
|
||||
r"(\\d+)\\s+scenario(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
|
||||
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
|
||||
),
|
||||
"steps": re.compile(
|
||||
r"(\\d+)\\s+step(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
|
||||
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
|
||||
),
|
||||
}
|
||||
DURATION_PATTERN = re.compile(r"Took\\s+(?:(\\d+)m\\s*)?([\\d\\.]+)s")
|
||||
|
||||
|
||||
def _extract_features_and_args(argv: List[str]) -> Tuple[List[str], List[str]]:
|
||||
positional: List[str] = []
|
||||
options: List[str] = []
|
||||
# ---------------------------------------------------------------------------
|
||||
# Summary helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _empty_summary():
|
||||
return {
|
||||
"features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
|
||||
"scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
|
||||
"steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
|
||||
"duration": 0.0,
|
||||
}
|
||||
|
||||
|
||||
def _extract_summary(runner):
|
||||
"""Build a summary dict from a completed behave Runner."""
|
||||
summary = _empty_summary()
|
||||
for feature in runner.features:
|
||||
status_name = feature.status.name
|
||||
if status_name == "passed":
|
||||
summary["features"]["passed"] += 1
|
||||
elif status_name == "skipped":
|
||||
summary["features"]["skipped"] += 1
|
||||
else:
|
||||
summary["features"]["failed"] += 1
|
||||
|
||||
summary["duration"] += feature.duration or 0.0
|
||||
|
||||
for scenario in feature.walk_scenarios():
|
||||
sname = scenario.status.name
|
||||
if sname == "passed":
|
||||
summary["scenarios"]["passed"] += 1
|
||||
elif sname == "skipped":
|
||||
summary["scenarios"]["skipped"] += 1
|
||||
elif sname == "failed":
|
||||
summary["scenarios"]["failed"] += 1
|
||||
else:
|
||||
summary["scenarios"]["errors"] += 1
|
||||
|
||||
for step in scenario.steps:
|
||||
stname = step.status.name
|
||||
if stname == "passed":
|
||||
summary["steps"]["passed"] += 1
|
||||
elif stname == "skipped":
|
||||
summary["steps"]["skipped"] += 1
|
||||
elif stname == "failed":
|
||||
summary["steps"]["failed"] += 1
|
||||
else:
|
||||
summary["steps"]["errors"] += 1
|
||||
return summary
|
||||
|
||||
|
||||
def _merge_summaries(summaries):
|
||||
total = _empty_summary()
|
||||
for s in summaries:
|
||||
for bucket in ("features", "scenarios", "steps"):
|
||||
for field in ("passed", "failed", "errors", "skipped"):
|
||||
total[bucket][field] += s.get(bucket, {}).get(field, 0)
|
||||
total["duration"] += float(s.get("duration", 0.0))
|
||||
return total
|
||||
|
||||
|
||||
def _format_duration(seconds):
|
||||
minutes = int(seconds // 60)
|
||||
remainder = seconds % 60
|
||||
if minutes:
|
||||
return f"{minutes}m {remainder:.3f}s"
|
||||
return f"{remainder:.3f}s"
|
||||
|
||||
|
||||
def _print_overall_summary(total, wall_seconds=None):
|
||||
print("\nOverall summary:")
|
||||
for bucket in ("features", "scenarios", "steps"):
|
||||
b = total[bucket]
|
||||
print(
|
||||
f"{b['passed']} {bucket} passed, {b['failed']} failed, "
|
||||
f"{b['errors']} errored, {b['skipped']} skipped"
|
||||
)
|
||||
if total["duration"]:
|
||||
print(f"Took {_format_duration(float(total['duration']))}")
|
||||
if wall_seconds is not None:
|
||||
print(f"Wall time: {_format_duration(wall_seconds)}")
|
||||
|
||||
|
||||
def _has_failures(total):
|
||||
return (
|
||||
total["features"]["failed"] > 0
|
||||
or total["features"]["errors"] > 0
|
||||
or total["scenarios"]["failed"] > 0
|
||||
or total["scenarios"]["errors"] > 0
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Feature discovery
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _iter_features(paths):
|
||||
collected = []
|
||||
for path in paths:
|
||||
p = Path(path)
|
||||
if p.is_dir():
|
||||
collected.extend(sorted(str(fp) for fp in p.rglob("*.feature")))
|
||||
else:
|
||||
collected.append(str(p))
|
||||
return collected
|
||||
|
||||
|
||||
def _extract_features_and_args(argv):
|
||||
positional = []
|
||||
options = []
|
||||
for arg in argv:
|
||||
if arg.startswith("-"):
|
||||
options.append(arg)
|
||||
@@ -150,128 +279,82 @@ def _extract_features_and_args(argv: List[str]) -> Tuple[List[str], List[str]]:
|
||||
return [DEFAULT_FEATURE_ROOT], options
|
||||
|
||||
|
||||
def _empty_summary() -> Dict[str, Dict[str, int] | float]:
|
||||
return {
|
||||
"features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
|
||||
"scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
|
||||
"steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
|
||||
"duration": 0.0,
|
||||
}
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-process behave execution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_runner(feature_paths, behave_args):
|
||||
"""Create a behave Runner with proper configuration defaults.
|
||||
|
||||
Mirrors the format-defaulting logic from ``behave.__main__.run_behave``
|
||||
so that ``-q`` and bare invocations get a sensible formatter instead
|
||||
of crashing on ``config.format is None``.
|
||||
"""
|
||||
from behave.configuration import Configuration
|
||||
from behave.runner import Runner
|
||||
from behave.runner_util import reset_runtime
|
||||
|
||||
reset_runtime()
|
||||
args = list(behave_args) + [str(p) for p in feature_paths]
|
||||
config = Configuration(command_args=args)
|
||||
if not config.format:
|
||||
config.format = [config.default_format]
|
||||
return Runner(config)
|
||||
|
||||
|
||||
def _parse_summary(text: str) -> Dict[str, Dict[str, int] | float]:
|
||||
summary = _empty_summary()
|
||||
for key, pattern in SUMMARY_PATTERNS.items():
|
||||
match = pattern.search(text)
|
||||
if match:
|
||||
summary[key]["passed"] = int(match.group(1))
|
||||
summary[key]["failed"] = int(match.group(2))
|
||||
summary[key]["errors"] = int(match.group(3) or 0)
|
||||
summary[key]["skipped"] = int(match.group(4))
|
||||
duration_match = DURATION_PATTERN.search(text)
|
||||
if duration_match:
|
||||
minutes = duration_match.group(1)
|
||||
seconds = float(duration_match.group(2))
|
||||
if minutes:
|
||||
seconds += int(minutes) * 60
|
||||
summary["duration"] = seconds
|
||||
return summary
|
||||
def _run_features_inprocess(feature_paths, behave_args):
|
||||
"""Run *all* feature_paths in a single behave Runner invocation.
|
||||
|
||||
Returns ``(failed: bool, summary: dict)``.
|
||||
"""
|
||||
runner = _make_runner(feature_paths, behave_args)
|
||||
failed = runner.run()
|
||||
summary = _extract_summary(runner)
|
||||
return failed, summary
|
||||
|
||||
|
||||
def _worker_run_features(payload):
|
||||
"""Entry point for multiprocessing workers.
|
||||
|
||||
def _merge_summaries(
|
||||
summaries: List[Dict[str, Dict[str, int] | float]],
|
||||
) -> Dict[str, Dict[str, int] | float]:
|
||||
total = _empty_summary()
|
||||
for summary in summaries:
|
||||
for bucket in ("features", "scenarios", "steps"):
|
||||
for field in ("passed", "failed", "errors", "skipped"):
|
||||
total[bucket][field] += summary.get(bucket, {}).get(field, 0)
|
||||
total["duration"] += float(summary.get("duration", 0.0))
|
||||
return total
|
||||
Runs a chunk of feature files in a forked child process. Heavy
|
||||
Python modules (cleveragents, behave, SQLAlchemy, ...) are already
|
||||
loaded in the parent and shared via copy-on-write after ``fork``.
|
||||
|
||||
``load_step_definitions()`` and ``load_hooks()`` still execute
|
||||
inside each worker (they ``exec()`` the step .py files and
|
||||
environment.py), but every ``import`` they trigger is a cache hit.
|
||||
"""
|
||||
feature_paths, behave_args = payload
|
||||
|
||||
stdout_buf = io.StringIO()
|
||||
stderr_buf = io.StringIO()
|
||||
|
||||
runner = _make_runner(feature_paths, behave_args)
|
||||
|
||||
with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
|
||||
failed = runner.run()
|
||||
|
||||
summary = _extract_summary(runner)
|
||||
return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary
|
||||
|
||||
|
||||
def _format_duration(seconds: float) -> str:
|
||||
minutes = int(seconds // 60)
|
||||
remainder = seconds % 60
|
||||
if minutes:
|
||||
return f"{minutes}m {remainder:.3f}s"
|
||||
return f"{remainder:.3f}s"
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI entry point
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _print_overall_summary(total: Dict[str, Dict[str, int] | float]) -> None:
|
||||
print("\\nOverall summary:")
|
||||
for bucket in ("features", "scenarios", "steps"):
|
||||
b = total[bucket]
|
||||
print(
|
||||
f"{b['passed']} {bucket} passed, {b['failed']} failed, "
|
||||
f"{b['errors']} errored, {b['skipped']} skipped"
|
||||
)
|
||||
if total["duration"]:
|
||||
print(f"Took {_format_duration(float(total['duration']))} (sum across workers)")
|
||||
|
||||
|
||||
|
||||
def _run_feature(
|
||||
base_args: List[str], feature: str
|
||||
) -> Tuple[int, str, str, str, Dict[str, Dict[str, int] | float]]:
|
||||
args = list(base_args)
|
||||
if "__SLIPCOVER_OUT__" in args:
|
||||
output_dir = os.environ.get("SLIPCOVER_OUTPUT_DIR", "build")
|
||||
output_name = f".slipcover.{uuid.uuid4().hex}.json"
|
||||
output_path = os.path.join(output_dir, output_name)
|
||||
args = [output_path if arg == "__SLIPCOVER_OUT__" else arg for arg in args]
|
||||
proc = subprocess.run([*args, feature], capture_output=True, text=True)
|
||||
combined_output = (proc.stdout or "") + "\\n" + (proc.stderr or "")
|
||||
summary = _parse_summary(combined_output)
|
||||
return proc.returncode, feature, proc.stdout or "", proc.stderr or "", summary
|
||||
|
||||
|
||||
def _build_base_args(other_args: List[str]) -> List[str]:
|
||||
if os.environ.get("BEHAVE_PARALLEL_COVERAGE"):
|
||||
source = os.environ.get("SLIPCOVER_SOURCE")
|
||||
omit = os.environ.get("SLIPCOVER_OMIT")
|
||||
output_path = "__SLIPCOVER_OUT__"
|
||||
base_args = [
|
||||
sys.executable,
|
||||
"-m",
|
||||
"slipcover",
|
||||
"--json",
|
||||
"--out",
|
||||
output_path,
|
||||
]
|
||||
if source:
|
||||
base_args.extend(["--source", source])
|
||||
if omit:
|
||||
base_args.extend(["--omit", omit])
|
||||
base_args.extend(["-m", "behave", *other_args])
|
||||
return base_args
|
||||
return [sys.executable, "-m", "behave", *other_args]
|
||||
|
||||
|
||||
def _iter_features(paths: Iterable[str]) -> List[str]:
|
||||
collected: List[str] = []
|
||||
for path in paths:
|
||||
p = Path(path)
|
||||
if p.is_dir():
|
||||
collected.extend(str(fp) for fp in p.rglob("*.feature"))
|
||||
else:
|
||||
collected.append(str(p))
|
||||
return collected
|
||||
|
||||
|
||||
def main(argv: List[str] | None = None) -> None:
|
||||
def main(argv=None):
|
||||
argv = list(sys.argv[1:] if argv is None else argv)
|
||||
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("--processes", "-j", type=int, default=None)
|
||||
known, remaining = parser.parse_known_args(argv)
|
||||
|
||||
processes = known.processes or os.cpu_count() or 1
|
||||
|
||||
feature_args, other_args = _extract_features_and_args(remaining)
|
||||
|
||||
# Pass-through --help / --version to behave
|
||||
if any(flag in remaining for flag in ("-h", "--help", "--version")):
|
||||
import subprocess
|
||||
code = subprocess.run(
|
||||
[sys.executable, "-m", "behave", *other_args, *feature_args]
|
||||
).returncode
|
||||
@@ -282,61 +365,60 @@ def main(argv: List[str] | None = None) -> None:
|
||||
print("No feature files found", file=sys.stderr)
|
||||
sys.exit(0)
|
||||
|
||||
base_args = _build_base_args(other_args)
|
||||
coverage_mode = bool(os.environ.get("BEHAVE_PARALLEL_COVERAGE"))
|
||||
|
||||
summaries: List[Dict[str, Dict[str, int] | float]] = []
|
||||
failures: List[Tuple[int, str]] = []
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=processes) as executor:
|
||||
futures = [
|
||||
executor.submit(_run_feature, base_args, feature)
|
||||
for feature in feature_paths
|
||||
start = time.monotonic()
|
||||
|
||||
if processes <= 1 or coverage_mode or len(feature_paths) == 1:
|
||||
# ---- sequential in-process mode ----
|
||||
failed, total = _run_features_inprocess(feature_paths, other_args)
|
||||
else:
|
||||
# ---- parallel in-process mode (multiprocessing fork) ----
|
||||
# Pre-import heavy modules so forked children get them for free.
|
||||
try:
|
||||
import cleveragents # noqa: F401
|
||||
except ImportError:
|
||||
pass
|
||||
try:
|
||||
import behave # noqa: F401
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# Split features into roughly equal chunks.
|
||||
chunk_size = max(1, (len(feature_paths) + processes - 1) // processes)
|
||||
chunks = [
|
||||
feature_paths[i : i + chunk_size]
|
||||
for i in range(0, len(feature_paths), chunk_size)
|
||||
]
|
||||
for fut in concurrent.futures.as_completed(futures):
|
||||
code, feature, stdout, stderr, summary = fut.result()
|
||||
|
||||
ctx = multiprocessing.get_context("fork")
|
||||
with ctx.Pool(processes=min(processes, len(chunks))) as pool:
|
||||
results = pool.map(
|
||||
_worker_run_features,
|
||||
[(chunk, other_args) for chunk in chunks],
|
||||
)
|
||||
|
||||
failed = False
|
||||
summaries = []
|
||||
for worker_failed, stdout, stderr, summary in results:
|
||||
if stdout:
|
||||
print(f"=== Output for {feature} ===")
|
||||
print(stdout, end="")
|
||||
if stderr:
|
||||
print(f"=== Stderr for {feature} ===", file=sys.stderr)
|
||||
print(stderr, file=sys.stderr, end="")
|
||||
print(stderr, end="", file=sys.stderr)
|
||||
failed = failed or worker_failed
|
||||
summaries.append(summary)
|
||||
if code:
|
||||
failures.append((code, feature))
|
||||
total = _merge_summaries(summaries)
|
||||
|
||||
total = _merge_summaries(summaries)
|
||||
_print_overall_summary(total)
|
||||
wall = time.monotonic() - start
|
||||
_print_overall_summary(total, wall_seconds=wall)
|
||||
|
||||
if failures:
|
||||
for code, feature in failures:
|
||||
print(f"behave failed for {feature} with code {code}", file=sys.stderr)
|
||||
if failed or _has_failures(total):
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
"""
|
||||
)
|
||||
|
||||
setup_path = source_dir / "setup.py"
|
||||
setup_path.write_text(
|
||||
"""
|
||||
from setuptools import find_packages, setup
|
||||
|
||||
setup(
|
||||
name="behave-parallel",
|
||||
version="1.2.4a1",
|
||||
packages=find_packages(),
|
||||
include_package_data=True,
|
||||
install_requires=["behave>=1.2.6"],
|
||||
entry_points={
|
||||
"console_scripts": ["behave-parallel=behave_parallel.cli:main"],
|
||||
},
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
session.install("setuptools", "wheel")
|
||||
session.install(str(source_dir))
|
||||
'''
|
||||
|
||||
|
||||
# =============================================================================
|
||||
@@ -526,10 +608,10 @@ COVERAGE_THRESHOLD = 97
|
||||
def coverage_report(session: nox.Session):
|
||||
"""Generate coverage report from Behave tests.
|
||||
|
||||
Runs behave tests in parallel via behave-parallel (same approach as
|
||||
unit_tests) with each worker collecting slipcover JSON coverage data.
|
||||
After all workers finish, the data files are merged with
|
||||
``slipcover --merge``.
|
||||
Runs all behave features in a single process under slipcover.
|
||||
The in-process behave-parallel runner avoids subprocess overhead,
|
||||
so a single slipcover invocation collects coverage for the entire
|
||||
suite -- no per-worker files or merge step required.
|
||||
|
||||
Coverage threshold is enforced at >=97%.
|
||||
|
||||
@@ -565,56 +647,51 @@ def coverage_report(session: nox.Session):
|
||||
"src/cleveragents/discovery/*",
|
||||
]
|
||||
)
|
||||
session.env["SLIPCOVER_SOURCE"] = source_paths
|
||||
session.env["SLIPCOVER_OMIT"] = omit_patterns
|
||||
session.env["SLIPCOVER_OUTPUT_DIR"] = "build"
|
||||
# Tell behave-parallel workers to run under slipcover
|
||||
|
||||
# Force sequential mode so slipcover wraps the entire process.
|
||||
session.env["BEHAVE_PARALLEL_COVERAGE"] = "1"
|
||||
|
||||
# Clean up any existing slipcover data
|
||||
for path in Path("build").glob(".slipcover.*.json"):
|
||||
path.unlink()
|
||||
|
||||
# Run behave tests in parallel; each worker collects coverage separately
|
||||
# Build behave-parallel args (sequential for coverage).
|
||||
behave_cmd = session.bin + "/behave-parallel"
|
||||
parallel_args = _behave_parallel_args(session.posargs)
|
||||
|
||||
if session.posargs and session.posargs[0].endswith(".feature"):
|
||||
args = [
|
||||
behave_args = [
|
||||
behave_cmd,
|
||||
"-q",
|
||||
"--tags=-discovery",
|
||||
"--no-capture",
|
||||
*parallel_args,
|
||||
*session.posargs,
|
||||
]
|
||||
else:
|
||||
args = [
|
||||
behave_args = [
|
||||
behave_cmd,
|
||||
"-q",
|
||||
"--tags=-discovery",
|
||||
"--no-capture",
|
||||
*parallel_args,
|
||||
"features/",
|
||||
*session.posargs,
|
||||
]
|
||||
|
||||
session.run(*args)
|
||||
|
||||
data_files = sorted(Path("build").glob(".slipcover.*.json"))
|
||||
if not data_files:
|
||||
session.error("No slipcover data files found in build/")
|
||||
|
||||
# Merge per-worker coverage data into a single JSON report
|
||||
# Wrap the entire behave-parallel run under slipcover.
|
||||
# A single process produces a single JSON output file directly.
|
||||
# Allow exit code 1 (test failures) — coverage data is still produced.
|
||||
session.run(
|
||||
"python",
|
||||
"-m",
|
||||
"slipcover",
|
||||
"--merge",
|
||||
*[str(path) for path in data_files],
|
||||
"--json",
|
||||
"--out",
|
||||
"build/coverage.json",
|
||||
"--source",
|
||||
source_paths,
|
||||
"--omit",
|
||||
omit_patterns,
|
||||
"--",
|
||||
*behave_args,
|
||||
success_codes=[0, 1],
|
||||
)
|
||||
|
||||
# Generate XML report for CI
|
||||
|
||||
Reference in New Issue
Block a user