Computes delta in number of skipped tests

skip_delta.py

"""Per-commit pass/skip/flaky/fail counts across the last N commits.

Usage: python skip_delta.py [<sha>] [-n 5] [--workflow trunk]
                            [--job-filter 'jammy.*cuda'] [--repo pytorch/pytorch]

Requires: curl_cffi, gh CLI.
"""

import argparse
import gzip
import json
import re
import subprocess
import sys
import urllib.parse
from curl_cffi import requests

KEYS = ("success", "skipped", "flaky", "failure")


def sh(*args: str) -> str:
    return subprocess.check_output(args, text=True).strip()


def gh(path: str) -> dict:
    return json.loads(sh("gh", "api", path))


def find_run(repo: str, sha: str, name: str) -> int | None:
    """Earliest workflow run id for `name` on `sha` (paginates — a commit
    can have hundreds of runs, the original push is rarely on page 1)."""
    ids, page = [], 1
    while True:
        batch = gh(f"repos/{repo}/actions/runs?head_sha={sha}&per_page=100&page={page}").get("workflow_runs", [])
        ids.extend(r["id"] for r in batch if r["name"] == name)
        if len(batch) < 100:
            return min(ids) if ids else None
        page += 1


def ch(query: str, params: dict) -> list[dict]:
    # hud.pytorch.org is behind Vercel anti-bot; impersonate Chrome.
    # The query path has a "/" — Next's [queryName] route is one segment, so percent-encode it.
    url = f"https://hud.pytorch.org/api/clickhouse/{urllib.parse.quote(query, safe='')}"
    r = requests.get(url, params={"parameters": json.dumps(params)}, impersonate="chrome124", timeout=180)
    body = gzip.decompress(r.content) if r.content[:2] == b"\x1f\x8b" else r.content
    if r.status_code != 200:
        raise RuntimeError(f"HUD {r.status_code}: {body[:200]!r}")
    return json.loads(body)


def counts(sha: str, wf_id: int, wf_name: str, job_re: re.Pattern) -> dict[str, int]:
    rows = ch("tests/test_status_counts_on_commits_by_file",
              {"shas": [sha], "workflowIds": f"[{wf_id}]"})
    totals = {k: 0 for k in KEYS}
    for r in rows:
        if r.get("workflow_name") == wf_name and job_re.search(r.get("job_name") or ""):
            for k in KEYS:
                totals[k] += r.get(k, 0)
    return totals


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("sha", nargs="?", default="HEAD")
    ap.add_argument("-n", "--count", type=int, default=5)
    ap.add_argument("--workflow", default="trunk")
    ap.add_argument("--job-filter", default="jammy.*cuda|cuda.*jammy")
    ap.add_argument("--repo", default="pytorch/pytorch")
    args = ap.parse_args()

    job_re = re.compile(args.job_filter, re.I)
    shas = sh("git", "log", "--format=%H", f"-{args.count}", args.sha).splitlines()

    print(f"{'SHA':<10} {'pass':>8} {'Δ':>7}  {'skip':>7} {'Δ':>7}  {'flaky':>5} {'Δ':>6}  {'fail':>5} {'Δ':>6}  Title")
    prev = None
    for sha in reversed(shas):  # oldest -> newest
        subj = sh("git", "log", "--format=%s", "-1", sha)[:60]
        wf = find_run(args.repo, sha, args.workflow)
        if wf is None:
            print(f"{sha[:9]:<10} {'(no run)':>67}  {subj}")
            prev = None
            continue
        c = counts(sha, wf, args.workflow, job_re)
        d = lambda k: "" if prev is None else (f"{c[k]-prev[k]:+d}" if c[k] != prev[k] else "0")
        print(f"{sha[:9]:<10} {c['success']:>8} {d('success'):>7}  "
              f"{c['skipped']:>7} {d('skipped'):>7}  "
              f"{c['flaky']:>5} {d('flaky'):>6}  "
              f"{c['failure']:>5} {d('failure'):>6}  {subj}")
        prev = c
    return 0


if __name__ == "__main__":
    sys.exit(main())