Coverage for .github / scripts / verify_gate_needs.py: 100.00%
121 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1#!/usr/bin/env python3
2"""Fail a workflow's gate job unless every job it needs succeeded.
4Every PR-triggered workflow ends in one ``gate:<workflow>`` job that ``needs``
5each other job in the file and is scheduled with ``always()``. Branch
6protection requires only those gates, so the merge rule is "every job in
7every PR workflow passed" without naming the jobs — matrix legs, renamed jobs
8and new jobs included — and a ruleset entry can never drift from a job name.
10The gate step hands this script the ``needs`` context (``toJSON(needs)``)::
12 {"job-id": {"result": "success|failure|cancelled|skipped", "outputs": {...}}}
14Every job must have ``result == "success"``. A job may be ``skipped`` only when
15named by ``--allow-skipped``, and the ``JOB=DEP.OUTPUT=VALUE`` form additionally
16requires the named output of another needed job to carry that value — the way
17a path-filtered job proves its skip was the filter's decision rather than a
18broken ``if:``. Anything else (failure, cancellation, an unexplained skip, a
19result this script does not recognise) fails the gate.
21Exit codes: ``0`` every job passed, ``1`` at least one job did not, ``2`` the
22input or the skip policy is malformed — so "the gate was misconfigured" never
23reads as "a job failed" or, worse, as a pass. Stdlib only: the gate job needs
24no dependency install.
25"""
27from __future__ import annotations
29import argparse
30import json
31import os
32import sys
33from dataclasses import dataclass
34from pathlib import Path
35from typing import Any
37#: The only result that satisfies the gate without a skip allowance.
38SUCCESS = "success"
39#: The only other result an allowance can excuse.
40SKIPPED = "skipped"
43class PolicyError(ValueError):
44 """The needs document or the skip policy is malformed (exit 2)."""
47@dataclass(frozen=True)
48class OutputCondition:
49 """``DEP.OUTPUT=VALUE``: the skip is legitimate only while this output holds."""
51 dependency: str
52 output: str
53 value: str
55 def describe(self) -> str:
56 return f"{self.dependency}.outputs.{self.output} == {self.value!r}"
59@dataclass(frozen=True)
60class SkipAllowance:
61 """``JOB`` or ``JOB=DEP.OUTPUT=VALUE`` from ``--allow-skipped``."""
63 job: str
64 condition: OutputCondition | None = None
66 def describe(self) -> str:
67 if self.condition is None:
68 return "skip allowed unconditionally"
69 return f"skip allowed while {self.condition.describe()}"
72def parse_allowance(spec: str) -> SkipAllowance:
73 """Parse one ``--allow-skipped`` value; malformed specs are a PolicyError."""
74 job, separator, condition = spec.partition("=")
75 if not job:
76 raise PolicyError(f"--allow-skipped {spec!r}: empty job id")
77 if not separator:
78 return SkipAllowance(job=job)
79 dependency, dot, rest = condition.partition(".")
80 output, equals, value = rest.partition("=")
81 if not (dependency and dot and output and equals and value):
82 raise PolicyError(
83 f"--allow-skipped {spec!r}: expected JOB or JOB=DEP.OUTPUT=VALUE "
84 "(DEP is another needed job id, OUTPUT one of its outputs)"
85 )
86 return SkipAllowance(job=job, condition=OutputCondition(dependency, output, value))
89def parse_needs(document: str) -> dict[str, dict[str, Any]]:
90 """Decode ``toJSON(needs)``; anything but a non-empty object of objects is a PolicyError."""
91 try:
92 needs = json.loads(document)
93 except json.JSONDecodeError as exc:
94 raise PolicyError(f"--needs is not valid JSON: {exc}") from exc
95 if not isinstance(needs, dict) or not needs:
96 raise PolicyError(
97 "--needs must be the non-empty `needs` context object; a gate that "
98 "needs nothing would pass vacuously"
99 )
100 for job_id, entry in needs.items():
101 if not isinstance(entry, dict) or not isinstance(entry.get("result"), str):
102 raise PolicyError(f"needs[{job_id!r}] is missing a string `result`")
103 return needs
106def _index_allowances(
107 needs: dict[str, dict[str, Any]], allowances: list[SkipAllowance]
108) -> dict[str, SkipAllowance]:
109 """Key allowances by job; a stale or duplicate allowance is a policy bug."""
110 by_job: dict[str, SkipAllowance] = {}
111 for allowance in allowances:
112 if allowance.job not in needs:
113 raise PolicyError(
114 f"--allow-skipped names {allowance.job!r}, which this gate does not need"
115 )
116 if allowance.condition is not None and allowance.condition.dependency not in needs:
117 raise PolicyError(
118 f"--allow-skipped for {allowance.job!r} reads "
119 f"{allowance.condition.dependency!r}, which this gate does not need"
120 )
121 if allowance.job in by_job:
122 raise PolicyError(f"--allow-skipped names {allowance.job!r} twice")
123 by_job[allowance.job] = allowance
124 return by_job
127def _judge_skip(needs: dict[str, dict[str, Any]], allowance: SkipAllowance) -> tuple[str, bool]:
128 """Note for a skipped job with an allowance, and whether the skip is excused."""
129 condition = allowance.condition
130 if condition is None:
131 return allowance.describe(), True
132 outputs = needs[condition.dependency].get("outputs")
133 actual = outputs.get(condition.output) if isinstance(outputs, dict) else None
134 if actual == condition.value:
135 return allowance.describe(), True
136 return (
137 f"skipped, but {condition.dependency}.outputs.{condition.output} is "
138 f"{actual!r}, not {condition.value!r}",
139 False,
140 )
143def evaluate(
144 needs: dict[str, dict[str, Any]], allowances: list[SkipAllowance]
145) -> tuple[list[tuple[str, str, str]], list[str]]:
146 """Judge every needed job.
148 Returns ``(rows, violations)``: one ``(job, result, note)`` row per needed
149 job in sorted order, and the human-readable violations.
150 """
151 by_job = _index_allowances(needs, allowances)
152 rows: list[tuple[str, str, str]] = []
153 violations: list[str] = []
154 for job_id in sorted(needs):
155 result = needs[job_id]["result"]
156 if result == SUCCESS:
157 rows.append((job_id, result, ""))
158 continue
159 allowance = by_job.get(job_id)
160 if result == SKIPPED and allowance is not None:
161 note, excused = _judge_skip(needs, allowance)
162 else:
163 note = "skipped without an allowance" if result == SKIPPED else f"result was {result!r}"
164 excused = False
165 rows.append((job_id, result, note))
166 if not excused:
167 violations.append(f"{job_id}: {note}")
168 return rows, violations
171def render_table(rows: list[tuple[str, str, str]]) -> str:
172 """Markdown table of every needed job — the step summary and the log share it."""
173 lines = ["| Job | Result | Note |", "|-----|--------|------|"]
174 for job_id, result, note in rows:
175 marker = "✅" if result == SUCCESS or note.startswith("skip allowed") else "❌"
176 lines.append(f"| `{job_id}` | {marker} `{result}` | {note} |")
177 return "\n".join(lines)
180def write_step_summary(table: str, verdict: str) -> None:
181 """Append the verdict to the job summary when GitHub provides one."""
182 summary_path = os.environ.get("GITHUB_STEP_SUMMARY")
183 if not summary_path:
184 return
185 with Path(summary_path).open("a", encoding="utf-8") as handle:
186 handle.write(f"### Gate: {verdict}\n\n{table}\n\n")
189def main(argv: list[str] | None = None) -> int:
190 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
191 parser.add_argument(
192 "--needs",
193 required=True,
194 help="the gate job's `needs` context as JSON (pass `${{ toJSON(needs) }}` through env)",
195 )
196 parser.add_argument(
197 "--allow-skipped",
198 action="append",
199 default=[],
200 metavar="JOB[=DEP.OUTPUT=VALUE]",
201 help=(
202 "a needed job whose `skipped` result is acceptable; with the "
203 "DEP.OUTPUT=VALUE suffix, only while that output of another needed "
204 "job carries that value (repeatable)"
205 ),
206 )
207 args = parser.parse_args(argv)
209 try:
210 needs = parse_needs(args.needs)
211 allowances = [parse_allowance(spec) for spec in args.allow_skipped]
212 rows, violations = evaluate(needs, allowances)
213 except PolicyError as exc:
214 print(f"::error::gate misconfigured: {exc}", file=sys.stderr)
215 return 2
217 table = render_table(rows)
218 verdict = (
219 "every needed job succeeded"
220 if not violations
221 else f"{len(violations)} job(s) did not succeed"
222 )
223 print(table)
224 write_step_summary(table, verdict)
225 for violation in violations:
226 print(f"::error::{violation}", file=sys.stderr)
227 if violations:
228 print(f"::error::refusing a successful gate: {verdict}", file=sys.stderr)
229 return 1
230 print(f"Gate satisfied: {len(rows)} job(s) succeeded or were skipped by policy.")
231 return 0
234if __name__ == "__main__":
235 raise SystemExit(main())