329 lines
13 KiB
Python
329 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Run molecule tests sequentially while polling Gitea for other runner failures.
|
|
|
|
Each pair is encoded as one of:
|
|
|
|
- **Single-role (4-part):** ``scenario|platform_name|platform_image|platform_command``
|
|
- **Multi-role (5-part):** ``role|scenario|platform_name|platform_image|platform_command``
|
|
|
|
Pairs are executed one at a time (molecule scenarios share temp directories and
|
|
Docker networks, so parallel execution within a single runner is unsafe).
|
|
|
|
A background thread polls the Gitea API. If any other molecule matrix runner
|
|
reports failure, the current molecule subprocess is killed and this runner
|
|
exits early with code 1.
|
|
|
|
Usage::
|
|
|
|
# Single-role
|
|
python3 -m devx.molecule.molecule_ci_guard pair1 pair2 ...
|
|
# Multi-role
|
|
python3 -m devx.molecule.molecule_ci_guard --roles-root ansible/roles pair1 pair2 ...
|
|
|
|
Environment variables:
|
|
GITEA_URL Base URL of the Gitea instance.
|
|
CI_GITEA_API_TOKEN API token with repo access (CI_GITEA_TOKEN accepted for legacy).
|
|
RUN_ID Workflow run ID (GITHUB_RUN_ID).
|
|
JOB_NAME Base job name (GITHUB_JOB), e.g. "molecule-tests".
|
|
MATRIX_INDEX Current matrix index (runner-index).
|
|
GITEA_REPOSITORY Repository in "owner/repo" format.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import os
|
|
import signal
|
|
import subprocess # nosec B404
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import click
|
|
import requests
|
|
|
|
from devx.config import REPO_NAME, REPO_OWNER
|
|
from devx.i18n import _
|
|
from devx.tokens import get_ci_token
|
|
|
|
POLL_INTERVAL = 10
|
|
|
|
|
|
def get_running_jobs(gitea_url: str, owner: str, repo: str, token: str, run_id: int) -> list[dict]:
|
|
"""Return jobs for the given workflow run."""
|
|
url = f"{gitea_url}/api/v1/repos/{owner}/{repo}/actions/runs/{run_id}/jobs"
|
|
headers = {"Authorization": f"token {token}"}
|
|
response = requests.get(url, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data.get("jobs", [])
|
|
|
|
|
|
def any_other_runner_failed(jobs: list[dict], current_job_name: str, current_index: int) -> bool:
|
|
"""Return True if any other molecule matrix job has failed."""
|
|
for job in jobs:
|
|
name = job.get("name", "")
|
|
if not name.startswith(current_job_name):
|
|
continue
|
|
if name == f"{current_job_name} ({current_index})" or name == current_job_name:
|
|
continue
|
|
if job.get("conclusion") == "failure":
|
|
return True
|
|
return False
|
|
|
|
|
|
def poll_for_other_failures(
|
|
gitea_url: str,
|
|
owner: str,
|
|
repo: str,
|
|
token: str,
|
|
run_id: int,
|
|
job_name: str,
|
|
current_index: int,
|
|
stop_event: threading.Event,
|
|
failed_event: threading.Event,
|
|
) -> None:
|
|
"""Background thread: poll API and signal if another runner fails."""
|
|
while not stop_event.is_set():
|
|
try:
|
|
jobs = get_running_jobs(gitea_url, owner, repo, token, run_id)
|
|
if any_other_runner_failed(jobs, job_name, current_index):
|
|
click.echo(_("Another molecule runner failed. Stopping this runner early."))
|
|
failed_event.set()
|
|
return
|
|
except requests.RequestException as exc:
|
|
click.echo(_("API poll warning: {exc}", exc=exc))
|
|
stop_event.wait(POLL_INTERVAL)
|
|
|
|
|
|
def build_molecule_cmd(scenario: str) -> list[str]:
|
|
"""Build the molecule command for a scenario."""
|
|
cmd = ["molecule", "test"]
|
|
if scenario != "default":
|
|
cmd.extend(["-s", scenario])
|
|
return cmd
|
|
|
|
|
|
def parse_pair(pair: str) -> tuple[str, str, str, str, str]:
|
|
"""Parse a pair string into (role, scenario, platform_name, platform_image, platform_command).
|
|
|
|
Supports both 4-part (single-role) and 5-part (multi-role) formats.
|
|
For 4-part pairs, role is empty (caller uses default role dir).
|
|
Spaces in the command field are encoded as ``__SPACE__`` to survive
|
|
shell word-splitting when ``$TEST_PAIRS`` is expanded unquoted.
|
|
"""
|
|
parts = pair.split("|")
|
|
if len(parts) == 4:
|
|
return "", parts[0], parts[1], parts[2], parts[3].replace("__SPACE__", " ")
|
|
if len(parts) == 5:
|
|
return parts[0], parts[1], parts[2], parts[3], parts[4].replace("__SPACE__", " ")
|
|
raise click.ClickException(f"Invalid pair format: {pair!r} (expected 4 or 5 pipe-delimited parts)")
|
|
|
|
|
|
def build_env_for_pair(pair: str, base_env: dict[str, str]) -> dict[str, str]:
|
|
"""Build environment for a single molecule pair."""
|
|
_role, _scenario, platform_name, platform_image, platform_command = parse_pair(pair)
|
|
env = base_env.copy()
|
|
env["MOLECULE_PLATFORM_NAME"] = platform_name
|
|
env["MOLECULE_PLATFORM_IMAGE"] = platform_image
|
|
if platform_command:
|
|
env["MOLECULE_PLATFORM_COMMAND"] = platform_command
|
|
elif "MOLECULE_PLATFORM_COMMAND" in env:
|
|
del env["MOLECULE_PLATFORM_COMMAND"]
|
|
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
|
|
|
|
|
|
def resolve_role_dir(role: str, roles_root: Path | None, repo_root: Path) -> Path:
|
|
"""Resolve the working directory for a molecule pair.
|
|
|
|
For multi-role pairs (role non-empty), uses ``roles_root/role``.
|
|
For single-role pairs, auto-discovers the first role with a molecule/
|
|
subdirectory under ``repo_root/ansible/roles/``.
|
|
"""
|
|
if role:
|
|
if roles_root is None:
|
|
roles_root = repo_root / "ansible" / "roles"
|
|
return roles_root / role
|
|
roles_dir = repo_root / "ansible" / "roles"
|
|
if roles_dir.is_dir():
|
|
role_dirs = sorted(d for d in roles_dir.iterdir() if (d / "molecule").is_dir())
|
|
if role_dirs:
|
|
return role_dirs[0]
|
|
return roles_dir / "role" # will produce a clear "not found" error
|
|
|
|
|
|
@click.command()
|
|
@click.argument("pairs", nargs=-1, required=True)
|
|
@click.option(
|
|
"--roles-root",
|
|
type=click.Path(exists=True, file_okay=False, path_type=Path),
|
|
default=None,
|
|
help="Root directory for multi-role pairs (e.g. ansible/roles). Required when pairs use 5-part format.",
|
|
)
|
|
def cli(pairs: tuple[str, ...], roles_root: Path | None) -> None:
|
|
"""Run molecule pairs sequentially, stop if another CI runner fails."""
|
|
gitea_url = os.environ.get("GITEA_URL", "")
|
|
try:
|
|
token = get_ci_token()
|
|
except click.ClickException:
|
|
token = None
|
|
run_id = int(os.environ.get("RUN_ID", "0"))
|
|
job_name = os.environ.get("JOB_NAME", "molecule-tests")
|
|
current_index = int(os.environ.get("MATRIX_INDEX", "0"))
|
|
repository = os.environ.get("GITEA_REPOSITORY", "")
|
|
owner, _sep, repo = repository.partition("/")
|
|
if not owner or not repo:
|
|
owner, repo = REPO_OWNER, REPO_NAME
|
|
|
|
if not all([gitea_url, token, run_id]):
|
|
click.echo(_("GITEA_URL/CI_GITEA_TOKEN/RUN_ID not set; running without cross-runner cancellation."))
|
|
|
|
# When devx is installed as a pip package, __file__ resolves to the
|
|
# site-packages directory, not the repo root. Use GITHUB_WORKSPACE
|
|
# (set by Gitea Actions) or cwd as the repo root.
|
|
repo_root = Path(os.environ.get("GITHUB_WORKSPACE", os.getcwd())).resolve()
|
|
|
|
base_env = os.environ.copy()
|
|
base_env.setdefault("DOCKER_HOST", f"unix:///run/user/{os.getuid()}/docker.sock")
|
|
base_env.setdefault("ANSIBLE_INJECT_INVOCATION", "1")
|
|
|
|
stop_event = threading.Event()
|
|
failed_event = threading.Event()
|
|
|
|
if gitea_url and token and run_id:
|
|
poller = threading.Thread(
|
|
target=poll_for_other_failures,
|
|
args=(
|
|
gitea_url,
|
|
owner,
|
|
repo,
|
|
token,
|
|
run_id,
|
|
job_name,
|
|
current_index,
|
|
stop_event,
|
|
failed_event,
|
|
),
|
|
daemon=True,
|
|
)
|
|
poller.start()
|
|
|
|
try:
|
|
for pair in pairs:
|
|
if failed_event.is_set():
|
|
sys.exit(1)
|
|
|
|
role, scenario, platform_name, _img, _cmd = parse_pair(pair)
|
|
click.echo(_("Running: {scenario} on {platform}", scenario=scenario, platform=platform_name))
|
|
|
|
cmd = build_molecule_cmd(scenario)
|
|
env = build_env_for_pair(pair, base_env)
|
|
cwd = resolve_role_dir(role, roles_root, repo_root)
|
|
|
|
process = subprocess.Popen( # nosec B603
|
|
cmd,
|
|
cwd=str(cwd),
|
|
env=env,
|
|
preexec_fn=os.setsid,
|
|
)
|
|
|
|
try:
|
|
while process.poll() is None:
|
|
if failed_event.is_set():
|
|
with contextlib.suppress(ProcessLookupError):
|
|
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
with contextlib.suppress(ProcessLookupError):
|
|
os.killpg(os.getpgid(process.pid), signal.SIGKILL)
|
|
process.wait()
|
|
# Clean up containers left behind by the killed test.
|
|
click.echo(_("Cleaning up: running molecule destroy for {scenario}", scenario=scenario))
|
|
destroy_cmd = ["molecule", "destroy"]
|
|
if scenario != "default":
|
|
destroy_cmd.extend(["-s", scenario])
|
|
with contextlib.suppress(subprocess.SubprocessError, OSError):
|
|
subprocess.run( # nosec B603, B607
|
|
destroy_cmd,
|
|
cwd=str(cwd),
|
|
env=env,
|
|
check=False,
|
|
capture_output=True,
|
|
timeout=120,
|
|
)
|
|
sys.exit(1)
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
with contextlib.suppress(ProcessLookupError):
|
|
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
|
|
process.wait()
|
|
# Clean up containers left behind by the interrupted test.
|
|
click.echo(_("Cleaning up: running molecule destroy for {scenario}", scenario=scenario))
|
|
destroy_cmd = ["molecule", "destroy"]
|
|
if scenario != "default":
|
|
destroy_cmd.extend(["-s", scenario])
|
|
with contextlib.suppress(subprocess.SubprocessError, OSError):
|
|
subprocess.run( # nosec B603, B607
|
|
destroy_cmd,
|
|
cwd=str(cwd),
|
|
env=env,
|
|
check=False,
|
|
capture_output=True,
|
|
timeout=120,
|
|
)
|
|
sys.exit(1)
|
|
|
|
rc = process.returncode
|
|
|
|
if rc != 0:
|
|
click.echo(_("FAILED: {pair} exited with code {code}", pair=pair, code=rc))
|
|
# Run molecule destroy to clean up containers left behind by the
|
|
# failed test. Without this, containers stay running and accumulate
|
|
# on the runner, consuming disk/memory and degrading CI performance.
|
|
click.echo(_("Cleaning up: running molecule destroy for {scenario}", scenario=scenario))
|
|
destroy_cmd = ["molecule", "destroy"]
|
|
if scenario != "default":
|
|
destroy_cmd.extend(["-s", scenario])
|
|
with contextlib.suppress(subprocess.SubprocessError, OSError):
|
|
subprocess.run( # nosec B603, B607
|
|
destroy_cmd,
|
|
cwd=str(cwd),
|
|
env=env,
|
|
check=False,
|
|
capture_output=True,
|
|
timeout=120,
|
|
)
|
|
sys.exit(rc)
|
|
|
|
click.echo(_("PASSED: {pair}", pair=pair))
|
|
|
|
# Prune Docker data between scenarios to prevent disk exhaustion
|
|
# in Docker-in-Docker molecule containers (each scenario pulls
|
|
# hundreds of MB of images that accumulate across pairs).
|
|
with contextlib.suppress(subprocess.SubprocessError, OSError):
|
|
subprocess.run( # nosec B603, B607
|
|
["docker", "system", "prune", "-af", "--volumes"],
|
|
check=False,
|
|
capture_output=True,
|
|
timeout=60,
|
|
)
|
|
|
|
click.echo(_("All molecule tests passed."))
|
|
finally:
|
|
stop_event.set()
|
|
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
cli()
|