Coverage for plugin/scripts/pr_status.py: 100%
155 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 14:46 +0000
1#!/usr/bin/env python3
2"""Report open pull/merge requests and what CI on the origin said about them.
4`/rhiza:quality` files issues, those issues become branches, and the branches become
5requests that were green on the author's machine. The gap this closes is what happens
6next: **local green and origin green are different claims**, and only the second one
7merges. A `make test` that passes on macOS with a warm cache says nothing about a
8matrix job on Windows, a lockfile the runner resolves differently, or a gate the repo
9runs only in CI.
11Asking the forge is the deterministic half of answering that, and it is deterministic
12in the way that hurts — the two CLIs disagree about **shape**, not just about flags:
14 gh pr list --json statusCheckRollup -> per-check objects, two different
15 __typenames, SCREAMING_CASE conclusions
16 glab mr list -F json -> the MRs, with no pipeline in sight;
17 ci get --merge-request <iid> the pipeline is a second call, and its
18 status is lower-case
20So this normalises both into one vocabulary — ``pass``/``fail``/``pending``/``skipped``/
21``cancelled``/``unknown`` — and reports, per request, a rollup plus the individual
22checks. **GitHub's answer is per job; GitLab's is per pipeline**, because a pipeline's
23job list is not a shape this has been able to verify against a real GitLab, and
24inventing one is how ``glab mr create --description-file`` shipped. That asymmetry is
25reported rather than smoothed over: every check carries the exact drill-down command
26for its platform, and on GitLab that command is the one that lists the failing jobs.
28**It reads; it never retries, cancels or pushes.** Deciding what a red job means, and
29fixing it, is `/rhiza:remote`'s job and needs judgement this cannot have.
31Usage:
32 uv run --python 3.12 --no-project python \
33 scripts/pr_status.py [--target-dir DIR] [--branch B | --all] [--limit N]
34 [--json] [--dry-run]
36With neither ``--branch`` nor ``--all``, the branch currently checked out is used; on a
37detached HEAD that falls back to every open request, which is what ``--all`` asks for
38explicitly.
40Exit codes:
41 0 the forge answered (whatever it said about CI), or --dry-run rendered
42 1 the platform CLI failed, is absent, or is not authenticated
43 2 the platform could not be determined
44"""
46from __future__ import annotations
48import argparse
49import json
50import shutil
51import subprocess # nosec B404
52import sys
53from pathlib import Path
54from typing import Any
56sys.path.insert(0, str(Path(__file__).resolve().parent))
57from _rhiza_forge import PlatformError, current_branch, detect_platform # noqa: E402
59EXIT_OK = 0
60EXIT_CLI_FAILED = 1
61EXIT_USAGE = 2
63SUCCESS = "pass"
64FAILURE = "fail"
65PENDING = "pending"
66SKIPPED = "skipped"
67CANCELLED = "cancelled"
68UNKNOWN = "unknown"
70# Worst-first. The rollup is the first state present, so a request with one failing job
71# reads `fail` however many green ones surround it, and a still-running one is never
72# reported as green just because everything finished so far has passed.
73_PRECEDENCE = (FAILURE, PENDING, CANCELLED, UNKNOWN, SUCCESS, SKIPPED)
75# GitHub answers a completed CheckRun with a `conclusion`; anything not listed is a
76# state this has not seen, which is `unknown` rather than a guessed pass.
77_GITHUB_CONCLUSIONS = {
78 "SUCCESS": SUCCESS,
79 "NEUTRAL": SUCCESS,
80 "SKIPPED": SKIPPED,
81 "CANCELLED": CANCELLED,
82 "FAILURE": FAILURE,
83 "TIMED_OUT": FAILURE,
84 "STARTUP_FAILURE": FAILURE,
85 "ACTION_REQUIRED": FAILURE,
86 "STALE": UNKNOWN,
87}
88# A commit status (an external CI posting to the statuses API) has no conclusion — just
89# a state, and a different vocabulary for it.
90_GITHUB_STATES = {
91 "SUCCESS": SUCCESS,
92 "EXPECTED": PENDING,
93 "PENDING": PENDING,
94 "ERROR": FAILURE,
95 "FAILURE": FAILURE,
96}
97# GitLab pipeline statuses, from the pipelines API. `manual` is a pipeline waiting for
98# somebody to press a button, which is pending on a human rather than on a runner — but
99# pending either way, and not something to report as green.
100_GITLAB_STATUSES = {
101 "success": SUCCESS,
102 "failed": FAILURE,
103 "canceled": CANCELLED,
104 "canceling": CANCELLED,
105 "skipped": SKIPPED,
106 "created": PENDING,
107 "waiting_for_resource": PENDING,
108 "preparing": PENDING,
109 "pending": PENDING,
110 "running": PENDING,
111 "manual": PENDING,
112 "scheduled": PENDING,
113}
115# The fields `gh pr list --json` is asked for. Spelled out rather than globbed: gh errors
116# on an unknown field, so this list is itself a contract with the CLI.
117_GH_FIELDS = "number,title,headRefName,url,isDraft,statusCheckRollup"
120class ForgeQueryError(Exception):
121 """A platform CLI was absent, unauthenticated, or answered something unreadable."""
124def build_list_command(platform: str, *, branch: str | None, limit: int) -> list[str]:
125 """Return the argv listing open requests, optionally narrowed to one *branch*."""
126 if platform == "github":
127 command = ["gh", "pr", "list", "--state", "open", "--json", _GH_FIELDS]
128 command += ["--limit", str(limit)]
129 return [*command, "--head", branch] if branch else command
130 command = ["glab", "mr", "list", "--output", "json", "--per-page", str(limit)]
131 return [*command, "--source-branch", branch] if branch else command
134def build_pipeline_command(iid: int | str) -> list[str]:
135 """Return the argv for the head pipeline of GitLab merge request *iid*.
137 ``--merge-request`` rather than ``--branch`` deliberately: GitLab's detached
138 merge-request pipelines run against ``refs/merge-requests/<iid>/head``, so a
139 request whose head pipeline has diverged from its source branch would otherwise be
140 reported against a pipeline that is not the one gating the merge.
141 """
142 return ["glab", "ci", "get", "--merge-request", str(iid), "--output", "json"]
145def _run_id(details_url: str) -> str | None:
146 """Extract the workflow-run id from a GitHub check's details URL.
148 The URL is ``…/actions/runs/<run>/job/<job>``. A commit status posted by an external
149 CI has no such URL, and there is no `gh` command to fetch its log — so None here
150 means "the log lives somewhere gh cannot reach", not "no log".
151 """
152 parts = details_url.split("/")
153 if "runs" not in parts:
154 return None
155 index = parts.index("runs") + 1
156 return parts[index] if index < len(parts) else None
159def build_logs_command(platform: str, check: dict[str, Any]) -> list[str] | None:
160 """Return the argv that drills into a failing *check*, or None when there is none.
162 Emitted rather than executed. A failed job's log is routinely megabytes, and which
163 part of it matters is exactly the judgement this script does not make — so the
164 command is handed to the caller, whose job is to read the answer.
165 """
166 if platform == "github":
167 run = _run_id(check.get("url") or "")
168 return ["gh", "run", "view", run, "--log-failed"] if run else None
169 pipeline = check.get("pipeline_id")
170 if pipeline is None:
171 return None
172 return [
173 "glab", "ci", "get", "--pipeline-id", str(pipeline),
174 "--status", "failed", "--with-job-details",
175 ] # fmt: skip
178def normalize_github_check(entry: dict[str, Any]) -> dict[str, Any]:
179 """Reduce one ``statusCheckRollup`` entry to ``{name, state, url, raw}``.
181 Two shapes arrive in the same array. A ``CheckRun`` is a GitHub Actions job and
182 carries ``status``/``conclusion``; a ``StatusContext`` is an external CI posting to
183 the statuses API and carries ``context``/``state``. Reading the second with the
184 first's keys yields a nameless check in an unknown state, which is how an external
185 required check comes to be invisible in a report that claims to be complete.
186 """
187 if entry.get("__typename") == "StatusContext":
188 return _status_context(entry)
189 return _check_run(entry)
192def _status_context(entry: dict[str, Any]) -> dict[str, Any]:
193 """Normalise a commit status — an external CI posting to the statuses API."""
194 raw = str(entry.get("state") or "")
195 return {
196 "name": entry.get("context") or "(unnamed status)",
197 "state": _GITHUB_STATES.get(raw.upper(), UNKNOWN),
198 "url": entry.get("targetUrl") or "",
199 "raw": raw,
200 }
203def _check_run(entry: dict[str, Any]) -> dict[str, Any]:
204 """Normalise a GitHub Actions job.
206 An unfinished job has no ``conclusion`` at all, so its ``status`` is what gets
207 reported — reading the missing conclusion instead would file every in-flight job
208 under ``unknown`` and bury the ones that genuinely are.
209 """
210 workflow = entry.get("workflowName") or ""
211 name = entry.get("name") or "(unnamed check)"
212 completed = str(entry.get("status") or "").upper() == "COMPLETED"
213 raw = str(entry.get("conclusion" if completed else "status") or "")
214 return {
215 "name": f"{name} ({workflow})" if workflow else name,
216 "state": _GITHUB_CONCLUSIONS.get(raw.upper(), UNKNOWN) if completed else PENDING,
217 "url": entry.get("detailsUrl") or "",
218 "raw": raw,
219 }
222def normalize_gitlab_pipeline(payload: dict[str, Any]) -> dict[str, Any]:
223 """Reduce a GitLab pipeline object to a single check.
225 One check per *pipeline*, not per job. `glab ci get` can be asked for job details,
226 but the shape it prints them in is not something this has been able to check
227 against a live GitLab — and a normaliser written from a guess is the exact bug that
228 shipped a `glab` flag which did not exist. The rollup below is honest at pipeline
229 granularity, and ``build_logs_command`` hands back the command that opens the jobs.
230 """
231 raw = str(payload.get("status") or "")
232 pipeline_id = payload.get("id")
233 return {
234 "name": f"pipeline #{pipeline_id}" if pipeline_id else "pipeline",
235 "state": _GITLAB_STATUSES.get(raw.lower(), UNKNOWN),
236 "url": payload.get("web_url") or "",
237 "raw": raw,
238 "pipeline_id": pipeline_id,
239 }
242def rollup(checks: list[dict[str, Any]]) -> str:
243 """Reduce a request's checks to one state; ``unknown`` when it has none.
245 No checks at all is *not* a pass. A request whose workflow never triggered — a
246 misspelled path filter, a workflow file that fails to parse — shows a clean check
247 list, and reporting that as green is how it gets merged.
248 """
249 present = {check["state"] for check in checks}
250 return next((state for state in _PRECEDENCE if state in present), UNKNOWN)
253def _cli_json(target_dir: Path, command: list[str]) -> Any:
254 """Run *command* and parse its stdout as JSON, or raise ``ForgeQueryError``."""
255 # The resolved path, not the bare name — see the same call in `platform_cli.run` for
256 # why the two differ on Windows.
257 executable = shutil.which(command[0])
258 if executable is None:
259 raise ForgeQueryError(f"{command[0]} is not installed — cannot read CI state")
260 result = subprocess.run( # nosec B603
261 [executable, *command[1:]], cwd=str(target_dir), capture_output=True, text=True, check=False
262 )
263 if result.returncode != 0:
264 raise ForgeQueryError(f"{' '.join(command)} failed: {result.stderr.strip()[:300]}")
265 try:
266 return json.loads(result.stdout or "null")
267 except ValueError as exc:
268 raise ForgeQueryError(f"{command[0]} returned output that is not JSON") from exc
271def _github_requests(target_dir: Path, command: list[str]) -> list[dict[str, Any]]:
272 """Every open PR the *command* lists, each with its normalised checks."""
273 requests = []
274 for entry in _cli_json(target_dir, command) or []:
275 checks = [normalize_github_check(c) for c in entry.get("statusCheckRollup") or []]
276 requests.append(
277 {
278 "id": entry.get("number"),
279 "title": entry.get("title") or "",
280 "branch": entry.get("headRefName") or "",
281 "url": entry.get("url") or "",
282 "draft": bool(entry.get("isDraft")),
283 "state": rollup(checks),
284 "checks": checks,
285 }
286 )
287 return requests
290def _gitlab_requests(target_dir: Path, command: list[str]) -> list[dict[str, Any]]:
291 """Every open MR the *command* lists, each with its head pipeline as one check.
293 The pipeline is a second call per request, so a request whose pipeline cannot be
294 read reports ``unknown`` and keeps going: one MR with no pipeline yet must not cost
295 the caller the report on all the others.
296 """
297 requests = []
298 for entry in _cli_json(target_dir, command) or []:
299 iid = entry.get("iid")
300 try:
301 pipeline = _cli_json(target_dir, build_pipeline_command(iid)) or {}
302 except ForgeQueryError:
303 pipeline = {}
304 checks = [normalize_gitlab_pipeline(pipeline)] if pipeline else []
305 requests.append(
306 {
307 "id": iid,
308 "title": entry.get("title") or "",
309 "branch": entry.get("source_branch") or "",
310 "url": entry.get("web_url") or "",
311 "draft": bool(entry.get("draft")),
312 "state": rollup(checks),
313 "checks": checks,
314 }
315 )
316 return requests
319def collect(
320 target_dir: Path,
321 *,
322 branch: str | None,
323 limit: int,
324 dry_run: bool = False,
325) -> dict[str, Any]:
326 """Query the forge for *target_dir* and return the normalised report."""
327 platform = detect_platform(target_dir)
328 command = build_list_command(platform, branch=branch, limit=limit)
329 summary: dict[str, Any] = {
330 "platform": platform,
331 "branch": branch,
332 "command": command,
333 "dry_run": dry_run,
334 "requests": [],
335 "notes": [],
336 }
337 if dry_run:
338 summary["notes"].append("dry run — nothing was executed")
339 return summary
341 reader = _github_requests if platform == "github" else _gitlab_requests
342 summary["requests"] = reader(target_dir, command)
343 for request in summary["requests"]:
344 for check in request["checks"]:
345 check["logs_command"] = build_logs_command(platform, check)
346 return summary
349_MARK = {
350 SUCCESS: "✓",
351 FAILURE: "✗",
352 PENDING: "•",
353 SKIPPED: "–",
354 CANCELLED: "⊘",
355 UNKNOWN: "?",
356}
359def render(summary: dict[str, Any]) -> str:
360 """Render *summary* as the human report."""
361 lines = [f"platform {summary['platform']}", f"command {' '.join(summary['command'])}"]
362 if not summary["requests"] and not summary["dry_run"]:
363 lines.append("")
364 lines.append("no open requests matched — nothing is waiting on CI")
365 for request in summary["requests"]:
366 draft = " [draft]" if request["draft"] else ""
367 lines.append("")
368 lines.append(
369 f"#{request['id']} {request['state'].upper()}{draft} {request['title']}"
370 f"\n {request['branch']} {request['url']}"
371 )
372 lines += [f" {line}" for line in _render_checks(request["checks"])]
373 lines += [f"note {note}" for note in summary["notes"]]
374 return "\n".join(lines)
377def _render_checks(checks: list[dict[str, Any]]) -> list[str]:
378 """Render one request's checks, with a drill-down line under each failing one."""
379 if not checks:
380 return ["(no checks reported — did a workflow ever trigger?)"]
381 lines = []
382 for check in checks:
383 mark = _MARK.get(check["state"], "?")
384 lines.append(f"{mark} {check['name']} {check['url']}".rstrip())
385 command = check.get("logs_command")
386 if check["state"] == FAILURE and command:
387 lines.append(f" logs: {' '.join(command)}")
388 return lines
391def main(argv: list[str] | None = None) -> int:
392 """Entry point: report the open requests' CI state and return an exit code."""
393 parser = argparse.ArgumentParser(description="Report open requests and their CI state.")
394 parser.add_argument("--target-dir", default=".", help="Repository root (default: cwd).")
395 scope = parser.add_mutually_exclusive_group()
396 scope.add_argument("--branch", default=None, help="Only the request for this branch.")
397 scope.add_argument(
398 "--all", action="store_true", help="Every open request, not just this branch's."
399 )
400 parser.add_argument("--limit", type=int, default=20, help="Maximum requests to fetch.")
401 parser.add_argument(
402 "--dry-run", action="store_true", help="Print the query without running it."
403 )
404 parser.add_argument(
405 "--json", dest="json_output", action="store_true", help="Emit the report as JSON."
406 )
407 args = parser.parse_args(argv)
409 target_dir = Path(args.target_dir).resolve()
410 branch = None if args.all else (args.branch or current_branch(target_dir))
411 try:
412 summary = collect(target_dir, branch=branch, limit=args.limit, dry_run=args.dry_run)
413 except PlatformError as exc:
414 print(f"error: {exc}", file=sys.stderr)
415 return EXIT_USAGE
416 except ForgeQueryError as exc:
417 print(f"error: {exc}", file=sys.stderr)
418 return EXIT_CLI_FAILED
420 print(json.dumps(summary, indent=2) if args.json_output else render(summary))
421 return EXIT_OK
424if __name__ == "__main__":
425 raise SystemExit(main())