shoc-backend/scripts/require_commit_checks.py

#!/usr/bin/env python3
"""Fail unless required GitHub check runs succeeded on a commit SHA."""

from __future__ import annotations

import argparse
import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

REQUIRED_CHECK_NAMES = (
    "ci-complete",
)


def _run_recency(run: dict) -> tuple:
    """Order check runs so a later rerun wins over an earlier result."""
    started = run.get("started_at") or ""
    completed = run.get("completed_at") or ""
    run_id = run.get("id") or 0
    return (started, completed, run_id)


def classify_checks(
    check_runs: list[dict], required_names: tuple[str, ...] = REQUIRED_CHECK_NAMES
) -> tuple[str, list[str]]:
    """Return ('success'|'pending'|'failure', detail lines)."""
    by_name: dict[str, dict] = {}
    for run in check_runs:
        name = run.get("name")
        if name not in required_names:
            continue
        current = by_name.get(name)
        if current is None or _run_recency(run) > _run_recency(current):
            by_name[name] = run

    missing = [name for name in required_names if name not in by_name]
    if missing:
        return "pending", [f"missing: {name}" for name in missing]

    details: list[str] = []
    pending = False
    failed = False
    for name in required_names:
        run = by_name[name]
        status = run.get("status")
        conclusion = run.get("conclusion")
        details.append(f"{name} status={status} conclusion={conclusion}")
        if status != "completed":
            pending = True
        elif conclusion != "success":
            failed = True
    if failed:
        return "failure", details
    if pending:
        return "pending", details
    return "success", details


def fetch_check_runs(repo: str, sha: str, token: str) -> list[dict]:
    url = (
        f"https://api.github.com/repos/{repo}/commits/{urllib.parse.quote(sha)}"
        "/check-runs?per_page=100"
    )
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
        },
    )
    try:
        with urllib.request.urlopen(request) as response:
            payload = json.load(response)
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", "replace")
        raise SystemExit(f"GitHub check-runs HTTP {exc.code}: {body}") from exc
    return payload.get("check_runs") or []


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--repo", required=True)
    parser.add_argument("--sha", required=True)
    parser.add_argument("--timeout-seconds", type=int, default=1200)
    parser.add_argument("--poll-seconds", type=int, default=15)
    args = parser.parse_args()
    token = __import__("os").environ.get("GITHUB_TOKEN") or __import__("os").environ.get(
        "GH_TOKEN"
    )
    if not token:
        print("FAIL: GITHUB_TOKEN or GH_TOKEN is required", file=sys.stderr)
        return 1

    deadline = time.time() + args.timeout_seconds
    while True:
        runs = fetch_check_runs(args.repo, args.sha, token)
        state, details = classify_checks(runs)
        for line in details:
            print(line)
        if state == "success":
            print(f"PASS: required checks succeeded on {args.sha}")
            return 0
        if state == "failure":
            print(f"FAIL: required checks did not succeed on {args.sha}", file=sys.stderr)
            return 1
        if time.time() >= deadline:
            print(
                f"FAIL: timed out waiting for required checks on {args.sha}",
                file=sys.stderr,
            )
            return 1
        print(f"waiting {args.poll_seconds}s for checks...")
        time.sleep(args.poll_seconds)


if __name__ == "__main__":
    raise SystemExit(main())