Compare commits

..
6 Commits
Author SHA1 Message Date
devx-ci-bot a7dcaee5c6 release: v0.8.1 [skip ci] 2026-06-23 20:45:02 +02:00
emil 02f8d3757b DEVX-14: fix: set fresh MOLECULE_HOME per pair to avoid stale config cache
Post-merge / detect-type (push) Successful in 14s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 20s
Post-merge / sync-wiki (push) Successful in 58s
Post-merge / badges (push) Successful in 1m6s
Post-merge / release (push) Successful in 48s
Post-merge / vikunja (push) Successful in 21s
2026-06-23 18:43:58 +00:00
devx-ci-bot 4311fb7648 release: v0.8.0 [skip ci] 2026-06-23 20:11:18 +02:00
emil 2ead959fcf DEVX-14: feat: fix molecule platforms to use sleep infinity, add --platforms-file
Post-merge / detect-type (push) Successful in 10s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 20s
Post-merge / release (push) Successful in 47s
Post-merge / vikunja (push) Successful in 20s
Post-merge / sync-wiki (push) Successful in 50s
Post-merge / badges (push) Successful in 1m0s
2026-06-23 18:10:19 +00:00
devx-ci-bot 9a60009d29 release: v0.7.0 [skip ci] 2026-06-23 18:26:54 +02:00
emil c20dfd185a DEVX-13: feat: add per-test timing quality gate to check_test_speed
Post-merge / detect-type (push) Successful in 12s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 26s
Post-merge / release (push) Successful in 43s
Post-merge / vikunja (push) Successful in 16s
Post-merge / sync-wiki (push) Successful in 48s
Post-merge / badges (push) Successful in 58s
2026-06-23 16:25:50 +00:00
16 changed files with 1624 additions and 1251 deletions
+1 -1
View File
@@ -27,7 +27,7 @@ jobs:
PYTHONPATH: src PYTHONPATH: src
run: | run: |
. .venv/bin/activate . .venv/bin/activate
python3 -m devx.tools.check_test_speed --max-seconds 60 python3 -m devx.tools.check_test_speed --max-seconds 60 --max-single-seconds 2.0
- name: Documentation coverage check - name: Documentation coverage check
env: env:
PYTHONPATH: src PYTHONPATH: src
+1 -1
View File
@@ -1 +1 @@
DEVX-12 DEVX-14
+18
View File
@@ -2,6 +2,24 @@
All notable changes to this project will be documented in this file. All notable changes to this project will be documented in this file.
## [0.8.1] - 2026-06-23
### Bug Fixes
- Set fresh MOLECULE_HOME per pair to avoid stale config cache
## [0.8.0] - 2026-06-23
### Features
- Fix molecule platforms to use sleep infinity, add --platforms-file
## [0.7.0] - 2026-06-23
### Features
- Add per-test timing quality gate to check_test_speed
## [0.6.0] - 2026-06-23 ## [0.6.0] - 2026-06-23
### Features ### Features
+7 -1
View File
@@ -76,7 +76,13 @@ Validate commit messages for conventional commit format.
### `devx tools check-test-speed` ### `devx tools check-test-speed`
Run unit tests and enforce a maximum execution-time budget. Run unit tests and enforce execution-time budgets:
- **Total suite time** must not exceed `--max-seconds` (default: 10s).
- **Per-test time** — no individual test may exceed `--max-single-seconds` (default: 0.5s, 0 to disable).
```bash
python3 -m devx.tools.check_test_speed --max-seconds 10 --max-single-seconds 0.5
```
### `devx tools configure-repo` ### `devx tools configure-repo`
+4 -3
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# pre-commit hook: fail if unit tests take longer than 10 seconds. # pre-commit hook: fail if unit tests are too slow.
# Aligned with CI timeout (ci.yml uses --max-seconds 10). # Checks both total suite time (10s) and per-test time (0.5s).
# Aligned with CI (ci.yml uses same thresholds).
set -e set -e
export PYTHONPATH=src export PYTHONPATH=src
python3 -m devx.tools.check_test_speed --max-seconds 10 python3 -m devx.tools.check_test_speed --max-seconds 10 --max-single-seconds 0.5
+1 -1
View File
@@ -1,3 +1,3 @@
"""devx — reusable development and CI/CD tools for oblachno-oss projects.""" """devx — reusable development and CI/CD tools for oblachno-oss projects."""
__version__ = "0.6.0" __version__ = "0.8.1"
+14 -5
View File
@@ -25,7 +25,7 @@ from pathlib import Path
import click import click
from devx.i18n import _ from devx.i18n import _
from devx.molecule.platforms import PLATFORMS from devx.molecule.platforms import PLATFORMS, load_platforms
DEFAULT_MAX_RUNNERS = 3 DEFAULT_MAX_RUNNERS = 3
MOLECULE_ROOT = Path("ansible/roles/gitea-runner/molecule") MOLECULE_ROOT = Path("ansible/roles/gitea-runner/molecule")
@@ -238,6 +238,13 @@ def _write_github_env(key: str, value: str) -> None:
help="Roles directory for multi-role discovery (scans */molecule/*/). " help="Roles directory for multi-role discovery (scans */molecule/*/). "
"Use this for projects with multiple Ansible roles. Default: disabled (single-role mode).", "Use this for projects with multiple Ansible roles. Default: disabled (single-role mode).",
) )
@click.option(
"--platforms-file",
type=click.Path(exists=True, file_okay=True, path_type=Path),
default=None,
help="JSON file with custom platform list (each entry: name, image, command). "
"Overrides the default platform matrix. Useful for projects with custom test images.",
)
def cli( def cli(
runner_index: int | None, runner_index: int | None,
max_runners: int, max_runners: int,
@@ -247,7 +254,9 @@ def cli(
skip_if_excess: bool, skip_if_excess: bool,
molecule_root: Path | None, molecule_root: Path | None,
roles_root: Path | None, roles_root: Path | None,
platforms_file: Path | None,
) -> None: ) -> None:
platforms = load_platforms(platforms_file)
# Multi-role mode: discover (role, scenario) pairs across all roles # Multi-role mode: discover (role, scenario) pairs across all roles
if roles_root is not None: if roles_root is not None:
role_scenarios = discover_multi_role_scenarios(roles_root) role_scenarios = discover_multi_role_scenarios(roles_root)
@@ -256,10 +265,10 @@ def cli(
click.echo(f"{role}|{scenario}") click.echo(f"{role}|{scenario}")
return return
if list_platforms: if list_platforms:
for p in PLATFORMS: for p in platforms:
click.echo(f"{p['name']}|{p['image']}|{p['command']}") click.echo(f"{p['name']}|{p['image']}|{p['command']}")
return return
pairs_mr = build_multi_role_pairs(role_scenarios) pairs_mr = build_multi_role_pairs(role_scenarios, platforms)
if runner_index is None: if runner_index is None:
groups = distribute_multi_role(pairs_mr, max_runners) groups = distribute_multi_role(pairs_mr, max_runners)
for i, group in enumerate(groups): for i, group in enumerate(groups):
@@ -291,10 +300,10 @@ def cli(
click.echo(s) click.echo(s)
return return
if list_platforms: if list_platforms:
for p in PLATFORMS: for p in platforms:
click.echo(f"{p['name']}|{p['image']}|{p['command']}") click.echo(f"{p['name']}|{p['image']}|{p['command']}")
return return
pairs = build_pairs(scenarios) pairs = build_pairs(scenarios, platforms)
if runner_index is None: if runner_index is None:
groups = distribute(pairs, max_runners) groups = distribute(pairs, max_runners)
for i, group in enumerate(groups): for i, group in enumerate(groups):
+6
View File
@@ -134,6 +134,12 @@ def build_env_for_pair(pair: str, base_env: dict[str, str]) -> dict[str, str]:
elif "MOLECULE_PLATFORM_COMMAND" in env: elif "MOLECULE_PLATFORM_COMMAND" in env:
del env["MOLECULE_PLATFORM_COMMAND"] del env["MOLECULE_PLATFORM_COMMAND"]
env["ANSIBLE_ALLOW_BROKEN_CONDITIONALS"] = "true" env["ANSIBLE_ALLOW_BROKEN_CONDITIONALS"] = "true"
# Use a fresh MOLECULE_HOME per pair to avoid stale config cache
# from previous CI runs (causes "Instances missing" errors).
if "MOLECULE_HOME" not in env:
import tempfile
env["MOLECULE_HOME"] = tempfile.mkdtemp(prefix="molecule-ci-")
return env return env
+33 -7
View File
@@ -10,13 +10,39 @@ dev tools and CI scripts.
from __future__ import annotations from __future__ import annotations
#: Supported OS platform matrix. import json
from pathlib import Path
#: Default supported OS platform matrix.
#: Each entry maps a short name to (image, command). #: Each entry maps a short name to (image, command).
#: The command must be systemd since rootless Docker requires #: Uses the project's pre-built molecule-test-base image with
#: loginctl/systemctl --user. #: ``sleep infinity`` (NOT systemd) to avoid cgroup v2 failures.
PLATFORMS: list[dict[str, str]] = [ PLATFORMS: list[dict[str, str]] = [
{"name": "ubuntu-2204", "image": "geerlingguy/docker-ubuntu2204-ansible:latest", "command": "/lib/systemd/systemd"}, {
{"name": "ubuntu-2404", "image": "geerlingguy/docker-ubuntu2404-ansible:latest", "command": "/lib/systemd/systemd"}, "name": "ubuntu-2604",
{"name": "debian-12", "image": "geerlingguy/docker-debian12-ansible:latest", "command": "/lib/systemd/systemd"}, "image": "git.oblachno.oblachno.fyi/oblachno/molecule-test-base:latest",
{"name": "archlinux", "image": "marcstraube/archlinux-ansible:latest", "command": "/usr/lib/systemd/systemd"}, "command": "sleep infinity",
},
] ]
def load_platforms(platforms_file: str | Path | None = None) -> list[dict[str, str]]:
"""Load platforms from a JSON file, falling back to PLATFORMS.
Args:
platforms_file: Path to a JSON file with a list of platform dicts.
Each dict must have ``name``, ``image``, and ``command`` keys.
Returns:
List of platform dictionaries.
"""
if platforms_file is None:
return PLATFORMS
path = Path(platforms_file)
if not path.is_file():
return PLATFORMS
with path.open() as f:
data = json.load(f)
if not isinstance(data, list) or not data:
return PLATFORMS
return data
+95 -11
View File
@@ -1,12 +1,21 @@
#!/usr/bin/env python3 #!/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: 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 from __future__ import annotations
import os
import re import re
import subprocess # nosec B404 import subprocess # nosec B404
@@ -14,18 +23,32 @@ import click
from devx.i18n import _ 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"] 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") _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]: 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 result = subprocess.run( # nosec B603
TEST_COMMAND, TEST_COMMAND,
capture_output=True, capture_output=True,
text=True, text=True,
check=False, check=False,
env=env,
) )
return result.stdout, result.stderr 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.")) 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: 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: if duration > max_seconds:
raise click.ClickException( raise click.ClickException(
_( _(
@@ -57,19 +95,58 @@ def check_speed(duration: float, max_seconds: float) -> None:
) )
def main(max_seconds: float) -> None: def check_per_test_speed(
"""Run tests, parse timing, and enforce the budget.""" 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() stdout, stderr = run_tests()
combined = stdout + "\n" + stderr combined = stdout + "\n" + stderr
click.echo(combined, err=False) click.echo(combined, err=False)
duration = parse_duration(combined) duration = parse_duration(combined)
check_speed(duration, max_seconds) 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( 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, duration=duration,
max=max_seconds, max=max_seconds,
single=max_single_seconds,
) )
) )
@@ -80,10 +157,17 @@ def main(max_seconds: float) -> None:
type=float, type=float,
default=DEFAULT_MAX_SECONDS, default=DEFAULT_MAX_SECONDS,
show_default=True, show_default=True,
help="Maximum allowed execution time in seconds.", help="Maximum allowed total execution time in seconds.",
) )
def cli(max_seconds: float) -> None: @click.option(
main(max_seconds) "--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 if __name__ == "__main__": # pragma: no cover
+21 -7
View File
@@ -916,13 +916,6 @@
"ru": "Tests passed.", "ru": "Tests passed.",
"zh": "Tests passed." "zh": "Tests passed."
}, },
"Unit tests passed in {duration:.2f}s (under {max}s limit).": {
"en": "Unit tests passed in {duration:.2f}s (under {max}s limit).",
"bg": "Unit tests passed in {duration:.2f}s (under {max}s limit).",
"de": "Unit tests passed in {duration:.2f}s (under {max}s limit).",
"ru": "Unit tests passed in {duration:.2f}s (under {max}s limit).",
"zh": "Unit tests passed in {duration:.2f}s (under {max}s limit)."
},
"Unit tests too slow: {duration:.2f}s (max allowed: {max}s).\n Fix: run 'make pytest-cov' to profile, then optimise slow tests.\n Hint: avoid unnecessary imports, use lighter mocks, or cache fixtures.": { "Unit tests too slow: {duration:.2f}s (max allowed: {max}s).\n Fix: run 'make pytest-cov' to profile, then optimise slow tests.\n Hint: avoid unnecessary imports, use lighter mocks, or cache fixtures.": {
"en": "Unit tests too slow: {duration:.2f}s (max allowed: {max}s).\n Fix: run 'make pytest-cov' to profile, then optimise slow tests.\n Hint: avoid unnecessary imports, use lighter mocks, or cache fixtures.", "en": "Unit tests too slow: {duration:.2f}s (max allowed: {max}s).\n Fix: run 'make pytest-cov' to profile, then optimise slow tests.\n Hint: avoid unnecessary imports, use lighter mocks, or cache fixtures.",
"bg": "Unit tests too slow: {duration:.2f}s (max allowed: {max}s).\n Fix: run 'make pytest-cov' to profile, then optimise slow tests.\n Hint: avoid unnecessary imports, use lighter mocks, or cache fixtures.", "bg": "Unit tests too slow: {duration:.2f}s (max allowed: {max}s).\n Fix: run 'make pytest-cov' to profile, then optimise slow tests.\n Hint: avoid unnecessary imports, use lighter mocks, or cache fixtures.",
@@ -1195,5 +1188,26 @@
"de": "Roles directory not found: {path}", "de": "Roles directory not found: {path}",
"ru": "Roles directory not found: {path}", "ru": "Roles directory not found: {path}",
"zh": "Roles directory not found: {path}" "zh": "Roles directory not found: {path}"
},
"Per-test speed check FAILED: {count} test(s) exceed {limit}s limit.": {
"en": "Per-test speed check FAILED: {count} test(s) exceed {limit}s limit.",
"bg": "Per-test speed check FAILED: {count} test(s) exceed {limit}s limit.",
"de": "Per-test speed check FAILED: {count} test(s) exceed {limit}s limit.",
"ru": "Per-test speed check FAILED: {count} test(s) exceed {limit}s limit.",
"zh": "Per-test speed check FAILED: {count} test(s) exceed {limit}s limit."
},
"Test '{name}' took {elapsed:.2f}s (limit: {limit}s). Optimise: use lighter fixtures, reduce I/O, or mock external calls.": {
"en": "Test '{name}' took {elapsed:.2f}s (limit: {limit}s). Optimise: use lighter fixtures, reduce I/O, or mock external calls.",
"bg": "Test '{name}' took {elapsed:.2f}s (limit: {limit}s). Optimise: use lighter fixtures, reduce I/O, or mock external calls.",
"de": "Test '{name}' took {elapsed:.2f}s (limit: {limit}s). Optimise: use lighter fixtures, reduce I/O, or mock external calls.",
"ru": "Test '{name}' took {elapsed:.2f}s (limit: {limit}s). Optimise: use lighter fixtures, reduce I/O, or mock external calls.",
"zh": "Test '{name}' took {elapsed:.2f}s (limit: {limit}s). Optimise: use lighter fixtures, reduce I/O, or mock external calls."
},
"Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit).": {
"en": "Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit).",
"bg": "Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit).",
"de": "Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit).",
"ru": "Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit).",
"zh": "Unit tests passed in {duration:.2f}s (under {max}s limit, all tests under {single}s per-test limit)."
} }
} }
+2 -1
View File
@@ -566,7 +566,8 @@ class TestVikunjaClient:
json={"done": True}, json={"done": True},
) )
def test_http_error_raises_api_error(self) -> None: @patch("devx.api_clients.time.sleep")
def test_http_error_raises_api_error(self, mock_sleep: MagicMock) -> None:
client = VikunjaClient("https://work.example.com", "tok") client = VikunjaClient("https://work.example.com", "tok")
mock_resp = MagicMock() mock_resp = MagicMock()
mock_resp.raise_for_status.side_effect = _mock_http_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error") mock_resp.raise_for_status.side_effect = _mock_http_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
+164 -10
View File
@@ -1,4 +1,4 @@
"""Unit tests for scripts/check_test_speed.py.""" """Unit tests for devx.tools.check_test_speed."""
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
@@ -8,10 +8,13 @@ from click.testing import CliRunner
from devx.tools.check_test_speed import ( from devx.tools.check_test_speed import (
DEFAULT_MAX_SECONDS, DEFAULT_MAX_SECONDS,
DEFAULT_MAX_SINGLE_SECONDS,
TEST_COMMAND, TEST_COMMAND,
check_per_test_speed,
check_speed, check_speed,
cli, cli,
parse_duration, parse_duration,
parse_per_test_durations,
run_tests, run_tests,
) )
@@ -23,12 +26,23 @@ class TestRunTests:
stdout, stderr = run_tests() stdout, stderr = run_tests()
assert stdout == "out" assert stdout == "out"
assert stderr == "err" assert stderr == "err"
mock_run.assert_called_once_with( mock_run.assert_called_once()
TEST_COMMAND, call_kwargs = mock_run.call_args
capture_output=True, assert call_kwargs.args[0] == TEST_COMMAND
text=True, assert call_kwargs.kwargs["capture_output"] is True
check=False, assert call_kwargs.kwargs["text"] is True
) assert call_kwargs.kwargs["check"] is False
env = call_kwargs.kwargs["env"]
assert "--durations=0" in env["PYTEST_ADDOPTS"]
@patch("devx.tools.check_test_speed.subprocess.run")
def test_run_tests_preserves_existing_pytest_addopts(self, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(stdout="out", stderr="err", returncode=0)
with patch.dict("os.environ", {"PYTEST_ADDOPTS": "-x"}, clear=False):
run_tests()
env = mock_run.call_args.kwargs["env"]
assert "--durations=0" in env["PYTEST_ADDOPTS"]
assert "-x" in env["PYTEST_ADDOPTS"]
class TestParseDuration: class TestParseDuration:
@@ -48,6 +62,38 @@ class TestParseDuration:
assert "Could not parse" in str(exc.value) assert "Could not parse" in str(exc.value)
class TestParsePerTestDurations:
def test_parses_call_lines(self) -> None:
output = "0.01s call tests/test_foo.py::test_bar\n"
durations = parse_per_test_durations(output)
assert len(durations) == 1
assert durations[0] == ("tests/test_foo.py::test_bar", 0.01)
def test_parses_setup_and_teardown(self) -> None:
output = (
"0.02s setup tests/test_foo.py::test_bar\n"
"0.01s call tests/test_foo.py::test_bar\n"
"0.00s teardown tests/test_foo.py::test_bar\n"
)
durations = parse_per_test_durations(output)
assert len(durations) == 3
names = [d[0] for d in durations]
assert "tests/test_foo.py::test_bar" in names
def test_sorted_slowest_first(self) -> None:
output = "0.01s call tests/test_a.py::test_slow\n0.50s call tests/test_b.py::test_fast\n"
durations = parse_per_test_durations(output)
assert durations[0][1] >= durations[1][1]
assert durations[0][1] == 0.50
def test_empty_output(self) -> None:
assert parse_per_test_durations("") == []
def test_ignores_non_duration_lines(self) -> None:
output = "Some random line\n234 passed in 0.70s\n"
assert parse_per_test_durations(output) == []
class TestCheckSpeed: class TestCheckSpeed:
def test_under_budget_passes(self) -> None: def test_under_budget_passes(self) -> None:
check_speed(1.0, 2.0) # should not raise check_speed(1.0, 2.0) # should not raise
@@ -64,6 +110,31 @@ class TestCheckSpeed:
assert "max allowed: 2.0s" in msg assert "max allowed: 2.0s" in msg
class TestCheckPerTestSpeed:
def test_no_violations_when_all_fast(self) -> None:
durations = [("test_a", 0.1), ("test_b", 0.2)]
assert check_per_test_speed(durations, 0.5) == []
def test_violation_when_test_exceeds_limit(self) -> None:
durations = [("test_slow", 0.6), ("test_fast", 0.1)]
violations = check_per_test_speed(durations, 0.5)
assert len(violations) == 1
assert "test_slow" in violations[0]
assert "0.60s" in violations[0]
def test_multiple_violations(self) -> None:
durations = [("test_a", 0.7), ("test_b", 0.6), ("test_c", 0.1)]
violations = check_per_test_speed(durations, 0.5)
assert len(violations) == 2
def test_exact_limit_passes(self) -> None:
durations = [("test_a", 0.5)]
assert check_per_test_speed(durations, 0.5) == []
def test_empty_durations(self) -> None:
assert check_per_test_speed([], 0.5) == []
def test_main_module_block() -> None: def test_main_module_block() -> None:
import devx.tools.check_test_speed as cts import devx.tools.check_test_speed as cts
@@ -77,39 +148,71 @@ class TestMain:
@patch("devx.tools.check_test_speed.run_tests") @patch("devx.tools.check_test_speed.run_tests")
@patch("devx.tools.check_test_speed.parse_duration") @patch("devx.tools.check_test_speed.parse_duration")
@patch("devx.tools.check_test_speed.check_speed") @patch("devx.tools.check_test_speed.check_speed")
@patch("devx.tools.check_test_speed.parse_per_test_durations")
@patch("devx.tools.check_test_speed.check_per_test_speed")
def test_successful_run( def test_successful_run(
self, self,
mock_check_per: MagicMock,
mock_parse_per: MagicMock,
mock_check: MagicMock, mock_check: MagicMock,
mock_parse: MagicMock, mock_parse: MagicMock,
mock_run: MagicMock, mock_run: MagicMock,
) -> None: ) -> None:
mock_run.return_value = ("stdout\n", "stderr\n") mock_run.return_value = ("stdout\n", "stderr\n")
mock_parse.return_value = 1.5 mock_parse.return_value = 1.5
mock_parse_per.return_value = []
mock_check_per.return_value = []
runner = CliRunner() runner = CliRunner()
result = runner.invoke(cli, []) result = runner.invoke(cli, [])
assert result.exit_code == 0 assert result.exit_code == 0
assert "1.50s" in result.output assert "1.50s" in result.output
assert "under 2.0s limit" in result.output assert "under 10.0s limit" in result.output
mock_run.assert_called_once() mock_run.assert_called_once()
mock_parse.assert_called_once_with("stdout\n\nstderr\n") mock_parse.assert_called_once_with("stdout\n\nstderr\n")
mock_check.assert_called_once_with(1.5, DEFAULT_MAX_SECONDS) mock_check.assert_called_once_with(1.5, DEFAULT_MAX_SECONDS)
mock_parse_per.assert_called_once()
mock_check_per.assert_called_once_with([], DEFAULT_MAX_SINGLE_SECONDS)
@patch("devx.tools.check_test_speed.run_tests") @patch("devx.tools.check_test_speed.run_tests")
@patch("devx.tools.check_test_speed.parse_duration") @patch("devx.tools.check_test_speed.parse_duration")
def test_slow_tests_exit( def test_slow_total_exits(
self, self,
mock_parse: MagicMock, mock_parse: MagicMock,
mock_run: MagicMock, mock_run: MagicMock,
) -> None: ) -> None:
mock_run.return_value = ("out\n", "err\n") mock_run.return_value = ("out\n", "err\n")
mock_parse.return_value = 3.0 mock_parse.return_value = 15.0
runner = CliRunner() runner = CliRunner()
result = runner.invoke(cli, []) result = runner.invoke(cli, [])
assert result.exit_code == 1 assert result.exit_code == 1
assert "too slow" in result.output.lower() assert "too slow" in result.output.lower()
@patch("devx.tools.check_test_speed.run_tests")
@patch("devx.tools.check_test_speed.parse_duration")
@patch("devx.tools.check_test_speed.check_speed")
@patch("devx.tools.check_test_speed.parse_per_test_durations")
@patch("devx.tools.check_test_speed.check_per_test_speed")
def test_per_test_violation_exits(
self,
mock_check_per: MagicMock,
mock_parse_per: MagicMock,
mock_check: MagicMock,
mock_parse: MagicMock,
mock_run: MagicMock,
) -> None:
mock_run.return_value = ("out\n", "err\n")
mock_parse.return_value = 3.0
mock_parse_per.return_value = [("test_slow", 0.8)]
mock_check_per.return_value = ["Test 'test_slow' took 0.80s (limit: 0.5s)."]
runner = CliRunner()
result = runner.invoke(cli, [])
assert result.exit_code == 1
assert "Per-test speed check FAILED" in result.output
assert "test_slow" in result.output
@patch("devx.tools.check_test_speed.run_tests") @patch("devx.tools.check_test_speed.run_tests")
def test_parse_failure_exits( def test_parse_failure_exits(
self, self,
@@ -125,16 +228,67 @@ class TestMain:
@patch("devx.tools.check_test_speed.run_tests") @patch("devx.tools.check_test_speed.run_tests")
@patch("devx.tools.check_test_speed.parse_duration") @patch("devx.tools.check_test_speed.parse_duration")
@patch("devx.tools.check_test_speed.check_speed") @patch("devx.tools.check_test_speed.check_speed")
@patch("devx.tools.check_test_speed.parse_per_test_durations")
@patch("devx.tools.check_test_speed.check_per_test_speed")
def test_custom_max_seconds( def test_custom_max_seconds(
self, self,
mock_check_per: MagicMock,
mock_parse_per: MagicMock,
mock_check: MagicMock, mock_check: MagicMock,
mock_parse: MagicMock, mock_parse: MagicMock,
mock_run: MagicMock, mock_run: MagicMock,
) -> None: ) -> None:
mock_run.return_value = ("out\n", "err\n") mock_run.return_value = ("out\n", "err\n")
mock_parse.return_value = 0.5 mock_parse.return_value = 0.5
mock_parse_per.return_value = []
mock_check_per.return_value = []
runner = CliRunner() runner = CliRunner()
result = runner.invoke(cli, ["--max-seconds", "1.5"]) result = runner.invoke(cli, ["--max-seconds", "1.5"])
assert result.exit_code == 0 assert result.exit_code == 0
mock_check.assert_called_once_with(0.5, 1.5) mock_check.assert_called_once_with(0.5, 1.5)
@patch("devx.tools.check_test_speed.run_tests")
@patch("devx.tools.check_test_speed.parse_duration")
@patch("devx.tools.check_test_speed.check_speed")
@patch("devx.tools.check_test_speed.parse_per_test_durations")
@patch("devx.tools.check_test_speed.check_per_test_speed")
def test_disable_per_test_check(
self,
mock_check_per: MagicMock,
mock_parse_per: MagicMock,
mock_check: MagicMock,
mock_parse: MagicMock,
mock_run: MagicMock,
) -> None:
mock_run.return_value = ("out\n", "err\n")
mock_parse.return_value = 1.0
runner = CliRunner()
result = runner.invoke(cli, ["--max-single-seconds", "0"])
assert result.exit_code == 0
mock_parse_per.assert_not_called()
mock_check_per.assert_not_called()
@patch("devx.tools.check_test_speed.run_tests")
@patch("devx.tools.check_test_speed.parse_duration")
@patch("devx.tools.check_test_speed.check_speed")
@patch("devx.tools.check_test_speed.parse_per_test_durations")
@patch("devx.tools.check_test_speed.check_per_test_speed")
def test_custom_max_single_seconds(
self,
mock_check_per: MagicMock,
mock_parse_per: MagicMock,
mock_check: MagicMock,
mock_parse: MagicMock,
mock_run: MagicMock,
) -> None:
mock_run.return_value = ("out\n", "err\n")
mock_parse.return_value = 1.0
mock_parse_per.return_value = []
mock_check_per.return_value = []
runner = CliRunner()
result = runner.invoke(cli, ["--max-single-seconds", "1.0"])
assert result.exit_code == 0
mock_check_per.assert_called_once_with([], 1.0)
+24 -6
View File
@@ -163,10 +163,7 @@ class TestCli:
runner = CliRunner() runner = CliRunner()
result = runner.invoke(cli, ["--list-platforms"]) result = runner.invoke(cli, ["--list-platforms"])
assert result.exit_code == 0 assert result.exit_code == 0
assert "ubuntu-2204" in result.output assert "ubuntu-2604" in result.output
assert "ubuntu-2404" in result.output
assert "debian-12" in result.output
assert "archlinux" in result.output
def test_no_runner_index_prints_all_groups(self, tmp_path: Path) -> None: def test_no_runner_index_prints_all_groups(self, tmp_path: Path) -> None:
from click.testing import CliRunner from click.testing import CliRunner
@@ -198,7 +195,7 @@ class TestCli:
assert result.exit_code == 0 assert result.exit_code == 0
# Output should contain encoded pairs with platform info # Output should contain encoded pairs with platform info
assert "alpha|" in result.output assert "alpha|" in result.output
assert "ubuntu-2204" in result.output assert "ubuntu-2604" in result.output
class TestGithubEnv: class TestGithubEnv:
@@ -409,7 +406,7 @@ class TestCliMultiRole:
runner = CliRunner() runner = CliRunner()
result = runner.invoke(cli, ["--roles-root", str(roles), "--list-platforms"]) result = runner.invoke(cli, ["--roles-root", str(roles), "--list-platforms"])
assert result.exit_code == 0 assert result.exit_code == 0
assert "ubuntu-2204" in result.output assert "ubuntu-2604" in result.output
def test_roles_root_no_runner_index_prints_groups(self, tmp_path: Path) -> None: def test_roles_root_no_runner_index_prints_groups(self, tmp_path: Path) -> None:
"""--roles-root without --runner-index prints all groups.""" """--roles-root without --runner-index prints all groups."""
@@ -422,6 +419,27 @@ class TestCliMultiRole:
assert "Runner 0:" in result.output assert "Runner 0:" in result.output
assert "Runner 1:" in result.output assert "Runner 1:" in result.output
def test_platforms_file_overrides_default(self, tmp_path: Path) -> None:
"""--platforms-file loads custom platforms from JSON."""
import json
from click.testing import CliRunner
from devx.molecule.distribute_molecule import cli
roles = tmp_path / "roles"
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
platforms_file = tmp_path / "platforms.json"
custom = [{"name": "custom-os", "image": "custom:latest", "command": "sleep infinity"}]
platforms_file.write_text(json.dumps(custom))
runner = CliRunner()
result = runner.invoke(
cli, ["--roles-root", str(roles), "--platforms-file", str(platforms_file), "--list-platforms"]
)
assert result.exit_code == 0
assert "custom-os" in result.output
assert "custom:latest" in result.output
def test_roles_root_skips_non_dir_role(self, tmp_path: Path) -> None: def test_roles_root_skips_non_dir_role(self, tmp_path: Path) -> None:
"""Non-directory entries in roles root are skipped.""" """Non-directory entries in roles root are skipped."""
roles = tmp_path / "roles" roles = tmp_path / "roles"
+3
View File
@@ -131,6 +131,7 @@ class TestCli:
clear=True, clear=True,
), ),
patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01), patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01),
patch("devx.molecule.molecule_ci_guard.POLL_INTERVAL", 0.01),
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen, patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect), patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect),
patch("os.killpg") as mock_killpg, patch("os.killpg") as mock_killpg,
@@ -177,6 +178,7 @@ class TestCli:
clear=True, clear=True,
), ),
patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01), patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01),
patch("devx.molecule.molecule_ci_guard.POLL_INTERVAL", 0.01),
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen, patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect), patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect),
patch("os.killpg", side_effect=ProcessLookupError("no such process")), patch("os.killpg", side_effect=ProcessLookupError("no such process")),
@@ -221,6 +223,7 @@ class TestCli:
clear=True, clear=True,
), ),
patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01), patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01),
patch("devx.molecule.molecule_ci_guard.POLL_INTERVAL", 0.01),
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen, patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect), patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect),
patch("os.killpg") as mock_killpg, patch("os.killpg") as mock_killpg,
+40 -7
View File
@@ -1,11 +1,13 @@
"""Unit tests for scripts/ci/platforms.py.""" """Unit tests for devx.molecule.platforms."""
from devx.molecule.platforms import PLATFORMS import json
from devx.molecule.platforms import PLATFORMS, load_platforms
class TestPlatforms: class TestPlatforms:
def test_platforms_not_empty(self) -> None: def test_platforms_not_empty(self) -> None:
assert len(PLATFORMS) >= 4 assert len(PLATFORMS) >= 1
def test_each_platform_has_required_keys(self) -> None: def test_each_platform_has_required_keys(self) -> None:
for p in PLATFORMS: for p in PLATFORMS:
@@ -17,9 +19,40 @@ class TestPlatforms:
names = [p["name"] for p in PLATFORMS] names = [p["name"] for p in PLATFORMS]
assert len(names) == len(set(names)) assert len(names) == len(set(names))
def test_platforms_use_sleep_infinity(self) -> None:
"""All default platforms must use sleep infinity, not systemd."""
for p in PLATFORMS:
assert p["command"] == "sleep infinity", f"Platform {p['name']} uses {p['command']}"
def test_known_platforms_present(self) -> None: def test_known_platforms_present(self) -> None:
names = {p["name"] for p in PLATFORMS} names = {p["name"] for p in PLATFORMS}
assert "ubuntu-2204" in names assert "ubuntu-2604" in names
assert "ubuntu-2404" in names
assert "debian-12" in names
assert "archlinux" in names class TestLoadPlatforms:
def test_load_platforms_default(self, tmp_path) -> None: # type: ignore[no-untyped-def]
"""load_platforms with no file returns PLATFORMS."""
result = load_platforms(None)
assert result == PLATFORMS
def test_load_platforms_from_file(self, tmp_path) -> None: # type: ignore[no-untyped-def]
"""load_platforms reads custom platforms from JSON file."""
custom = [
{"name": "custom-os", "image": "custom:latest", "command": "sleep infinity"},
]
f = tmp_path / "platforms.json"
f.write_text(json.dumps(custom))
result = load_platforms(f)
assert result == custom
def test_load_platforms_missing_file_falls_back(self, tmp_path) -> None: # type: ignore[no-untyped-def]
"""load_platforms falls back to PLATFORMS when file doesn't exist."""
result = load_platforms(tmp_path / "nonexistent.json")
assert result == PLATFORMS
def test_load_platforms_empty_list_falls_back(self, tmp_path) -> None: # type: ignore[no-untyped-def]
"""load_platforms falls back to PLATFORMS when file has empty list."""
f = tmp_path / "platforms.json"
f.write_text("[]")
result = load_platforms(f)
assert result == PLATFORMS