132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Run integration tests with cross-runner failure detection.
|
|
|
|
Wraps ``pytest`` with the same Gitea API polling mechanism used by
|
|
``molecule_ci_guard``. If any other integration-tests matrix runner
|
|
reports failure, the current pytest subprocess is killed and this runner
|
|
exits early with code 1.
|
|
|
|
Usage::
|
|
|
|
python3 -m devx.ci.integration_guard \\
|
|
-- test_file1.py test_file2.py
|
|
|
|
# With pytest options
|
|
python3 -m devx.ci.integration_guard \\
|
|
-- -x -v --tb=short test_file1.py
|
|
|
|
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. "integration-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
|
|
|
|
import click
|
|
|
|
from devx.config import REPO_NAME, REPO_OWNER
|
|
from devx.i18n import _
|
|
from devx.molecule.molecule_ci_guard import (
|
|
poll_for_other_failures,
|
|
)
|
|
from devx.tokens import get_ci_token
|
|
|
|
POLL_INTERVAL = 10
|
|
|
|
|
|
@click.command(context_settings={"ignore_unknown_options": True})
|
|
@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED, required=True)
|
|
def cli(pytest_args: tuple[str, ...]) -> None:
|
|
"""Run pytest with cross-runner failure detection."""
|
|
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", "integration-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."))
|
|
|
|
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()
|
|
|
|
cmd = [sys.executable, "-m", "pytest"]
|
|
cmd.extend(pytest_args)
|
|
|
|
click.echo(_("Running: {cmd}", cmd=" ".join(cmd)))
|
|
|
|
process = subprocess.Popen( # nosec B603
|
|
cmd,
|
|
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()
|
|
click.echo(_("Integration tests cancelled — another runner failed."))
|
|
sys.exit(1)
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
with contextlib.suppress(ProcessLookupError):
|
|
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
|
|
process.wait()
|
|
sys.exit(1)
|
|
finally:
|
|
stop_event.set()
|
|
|
|
rc = process.returncode
|
|
if rc != 0:
|
|
click.echo(_("Integration tests failed with exit code {code}", code=rc))
|
|
else:
|
|
click.echo(_("Integration tests passed."))
|
|
sys.exit(rc)
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
cli()
|