DEVX-13: feat: add per-test timing quality gate to check_test_speed

This commit is contained in:
emil
2026-06-23 16:25:50 +00:00
parent 547fef4f27
commit c20dfd185a
9 changed files with 1488 additions and 1225 deletions
+95 -11
View File
@@ -1,12 +1,21 @@
#!/usr/bin/env python3
"""Run unit tests and enforce a maximum execution-time budget.
"""Run unit tests and enforce execution-time budgets.
Checks two quality gates:
1. **Total suite time** must not exceed ``--max-seconds``.
2. **Per-test time** — no individual test may exceed ``--max-single-seconds``.
Usage:
python3 -m devx.tools.check_test_speed [--max-seconds N]
python3 -m devx.tools.check_test_speed [--max-seconds N] [--max-single-seconds S]
The module runs ``make test-unit`` with ``PYTEST_ADDOPTS=--durations=0`` so
that pytest emits per-test timing lines alongside the summary. Both the
total wall-clock time and individual test durations are parsed and validated.
"""
from __future__ import annotations
import os
import re
import subprocess # nosec B404
@@ -14,18 +23,32 @@ import click
from devx.i18n import _
DEFAULT_MAX_SECONDS = 2.0
DEFAULT_MAX_SECONDS = 10.0
DEFAULT_MAX_SINGLE_SECONDS = 0.5
TEST_COMMAND = ["make", "test-unit"]
# Matches pytest summary line: "234 passed in 0.70s"
_TIMING_RE = re.compile(r"(\d+) passed.* in ([0-9.]+)s")
# Matches per-test duration lines from --durations=0:
# 0.51s call tests/test_foo.py::test_bar
_DURATION_LINE_RE = re.compile(r"^(\d+\.?\d*)s\s+(?:setup|call|teardown)\s+(.+)$")
def run_tests() -> tuple[str, str]:
"""Execute the unit-test suite and return (stdout, stderr)."""
"""Execute the unit-test suite and return (stdout, stderr).
Sets ``PYTEST_ADDOPTS=--durations=0`` so pytest emits per-test timings.
"""
env = os.environ.copy()
existing = env.get("PYTEST_ADDOPTS", "")
env["PYTEST_ADDOPTS"] = f"--durations=0 {existing}".strip()
result = subprocess.run( # nosec B603
TEST_COMMAND,
capture_output=True,
text=True,
check=False,
env=env,
)
return result.stdout, result.stderr
@@ -43,8 +66,23 @@ def parse_duration(output: str) -> float:
raise click.ClickException(_("Could not parse test execution time from output."))
def parse_per_test_durations(output: str) -> list[tuple[str, float]]:
"""Extract per-test timings from ``--durations=0`` output.
Returns a list of ``(test_name, seconds)`` tuples sorted by duration
(slowest first).
"""
durations: list[tuple[str, float]] = []
for line in output.splitlines():
match = _DURATION_LINE_RE.match(line.strip())
if match:
durations.append((match.group(2).strip(), float(match.group(1))))
durations.sort(key=lambda x: x[1], reverse=True)
return durations
def check_speed(duration: float, max_seconds: float) -> None:
"""Validate duration is within budget; raise on violation."""
"""Validate total duration is within budget; raise on violation."""
if duration > max_seconds:
raise click.ClickException(
_(
@@ -57,19 +95,58 @@ def check_speed(duration: float, max_seconds: float) -> None:
)
def main(max_seconds: float) -> None:
"""Run tests, parse timing, and enforce the budget."""
def check_per_test_speed(
durations: list[tuple[str, float]],
max_single_seconds: float,
) -> list[str]:
"""Return a list of violation messages for tests exceeding the per-test limit.
An empty list means all tests are within budget.
"""
violations: list[str] = []
for name, elapsed in durations:
if elapsed > max_single_seconds:
violations.append(
_(
"Test '{name}' took {elapsed:.2f}s (limit: {limit}s). "
"Optimise: use lighter fixtures, reduce I/O, or mock external calls.",
name=name,
elapsed=elapsed,
limit=max_single_seconds,
)
)
return violations
def main(max_seconds: float, max_single_seconds: float) -> None:
"""Run tests, parse timings, and enforce both budgets."""
stdout, stderr = run_tests()
combined = stdout + "\n" + stderr
click.echo(combined, err=False)
duration = parse_duration(combined)
check_speed(duration, max_seconds)
if max_single_seconds > 0:
per_test = parse_per_test_durations(combined)
violations = check_per_test_speed(per_test, max_single_seconds)
if violations:
msg = _(
"Per-test speed check FAILED: {count} test(s) exceed {limit}s limit.",
count=len(violations),
limit=max_single_seconds,
)
click.echo(f"\n{msg}", err=True)
for v in violations:
click.echo(f" - {v}", err=True)
raise click.ClickException(msg)
click.echo(
_(
"Unit tests passed in {duration:.2f}s (under {max}s limit).",
"Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit).",
duration=duration,
max=max_seconds,
single=max_single_seconds,
)
)
@@ -80,10 +157,17 @@ def main(max_seconds: float) -> None:
type=float,
default=DEFAULT_MAX_SECONDS,
show_default=True,
help="Maximum allowed execution time in seconds.",
help="Maximum allowed total execution time in seconds.",
)
def cli(max_seconds: float) -> None:
main(max_seconds)
@click.option(
"--max-single-seconds",
type=float,
default=DEFAULT_MAX_SINGLE_SECONDS,
show_default=True,
help="Maximum allowed per-test time in seconds (0 to disable).",
)
def cli(max_seconds: float, max_single_seconds: float) -> None:
main(max_seconds, max_single_seconds)
if __name__ == "__main__": # pragma: no cover
+1211 -1197
View File
File diff suppressed because it is too large Load Diff