grm/scripts/molecule_ci_guard.py

#!/usr/bin/env python3
"""Run molecule tests while polling Gitea for other runner failures.

Usage:
    python3 scripts/molecule_ci_guard.py <pair1> <pair2> ...

Environment variables:
    GITEA_URL       Base URL of the Gitea instance.
    REPO_TOKEN      API token with repo access.
    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).
    OWNER, REPO     Repository owner/name.
"""

from __future__ import annotations

import os
import signal
import subprocess  # nosec B404
import sys
import threading
import time
from pathlib import Path

import click
import requests

from gitea_runner_manager.i18n import _

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
        # Skip the current runner instance
        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)


@click.command()
@click.argument("pairs", nargs=-1, required=True)
def cli(pairs: tuple[str, ...]) -> None:
    """Run molecule pairs and stop if another CI runner fails."""
    gitea_url = os.environ.get("GITEA_URL", "")
    token = os.environ.get("REPO_TOKEN", "")
    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", "oblachno-oss/grm")
    owner, sep, repo = repository.partition("/")
    if not owner or not repo:
        owner, repo = "oblachno-oss", "grm"

    if not all([gitea_url, token, run_id]):
        click.echo(
            _(
                "GITEA_URL/REPO_TOKEN/RUN_ID not set; running without cross-runner cancellation."
            )
        )

    repo_root = Path(__file__).resolve().parent.parent
    script_path = repo_root / "scripts" / "run_molecule_parallel.py"

    env = os.environ.copy()
    env.setdefault("DOCKER_HOST", f"unix:///run/user/{os.getuid()}/docker.sock")
    env.setdefault("ANSIBLE_INJECT_INVOCATION", "1")

    click.echo(_("Starting molecule runner with {count} pairs.", count=len(pairs)))

    process = subprocess.Popen(  # nosec B603
        [sys.executable, str(script_path), *pairs],
        env=env,
        stdout=None,
        stderr=None,
        preexec_fn=os.setsid,
    )

    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:
        while process.poll() is None:
            if failed_event.is_set():
                try:
                    os.killpg(os.getpgid(process.pid), signal.SIGTERM)
                except ProcessLookupError:
                    pass
                try:
                    process.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    try:
                        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
                    except ProcessLookupError:
                        pass
                    process.wait()
                sys.exit(1)
            time.sleep(1)
    finally:
        stop_event.set()

    sys.exit(process.returncode)


if __name__ == "__main__":  # pragma: no cover
    cli()