import re
import sys
from pathlib import Path
TEMPLATE_VERSION = 3
def recipe_body(text, name):
lines = text.splitlines()
start = None
for i, l in enumerate(lines):
if re.match(rf"^{re.escape(name)}(\s+[*+]?\w+.*)?:", l):
start = i
break
if start is None:
return None
body = []
for l in lines[start + 1:]:
if l.strip() and not l.startswith((" ", "\t")):
break
body.append(l)
return "\n".join(body)
def code_of(body):
out = []
for l in body.splitlines():
s = l.strip()
if s.startswith("#") or s.startswith("@#"):
continue
out.append(l)
return "\n".join(out)
RULES = [
("pr", "confirm-env",
"`pr` READS PR_CONFIRM, so a script or agent can satisfy the gate. Naming it in help "
"text is not honouring it -- the rule requires a parameter expansion.",
lambda c: re.search(r"\$\{?PR_CONFIRM", c) is not None),
("pr", "confirm-refuses",
"`pr` REFUSES rather than defaulting to yes when there is no terminal and no stdin",
lambda c: re.search(r"PR_CONFIRM", c) and re.search(r"exit 1|fail ", c)),
("pr", "confirm-explicit-y",
"`pr` still requires an explicit y — widening who can answer, not what counts",
lambda c: re.search(r'CONFIRM"?\s*=\s*"?y', c, re.I) is not None),
("open-pr", "gate-first",
"`open-pr` runs the gate before creating the PR",
lambda c: re.search(r"just\s+pr\b", c) is not None),
("open-pr", "creates-pr",
"`open-pr` is the call site that actually creates the PR",
lambda c: "gh pr create" in c),
("open-pr", "push-if-no-upstream",
"`open-pr` pushes when the branch has no upstream, and only then",
lambda c: "@{upstream}" in c and re.search(r"git push", c) is not None),
("merge-pr", "refuse-red",
"`merge-pr` refuses to merge over a failing check",
lambda c: re.search(r"FAILURE|TIMED_OUT|CANCELLED", c) is not None),
("merge-pr", "refuse-empty",
"`merge-pr` refuses an EMPTY status rollup — 'nothing ran' is not 'everything passed'",
lambda c: re.search(r"statusCheckRollup", c) is not None
and re.search(r"\[\]|empty|no checks", c, re.I) is not None),
("merge-pr", "refuse-pending",
"`merge-pr` refuses while checks are still running rather than racing them",
lambda c: re.search(r"still running|in progress|pending", c, re.I) is not None),
]
def check(justfile: Path):
text = justfile.read_text(encoding="utf-8")
failures = []
for recipe, key, desc, pred in RULES:
body = recipe_body(text, recipe)
if body is None:
failures.append((f"{recipe}:{key}", f"recipe `{recipe}` does not exist"))
continue
if not pred(code_of(body)):
failures.append((f"{recipe}:{key}", desc))
return failures
CONFORMANT = """
pr:
#!/usr/bin/env bash
if [ -n "${PR_CONFIRM:-}" ]; then CONFIRM="$PR_CONFIRM"
elif [ -t 0 ]; then read -r CONFIRM
else read -r -t 10 CONFIRM || CONFIRM=""
[ -n "$CONFIRM" ] || { echo "set PR_CONFIRM=y"; exit 1; }
fi
[ "$CONFIRM" = "y" ] || exit 1
open-pr *ARGS:
#!/usr/bin/env bash
just pr
if ! git rev-parse '@{upstream}' >/dev/null 2>&1; then git push -u origin "$B"; fi
gh pr create
merge-pr:
#!/usr/bin/env bash
STATES=$(gh pr view --json statusCheckRollup --jq '...')
if [ "$STATES" = "[]" ]; then echo "no checks have reported"; exit 1; fi
if echo "$STATES" | grep -q '""'; then echo "still running"; exit 1; fi
if echo "$STATES" | grep -qE 'FAILURE|TIMED_OUT'; then exit 1; fi
gh pr merge --squash
"""
def self_test():
import tempfile
problems = []
with tempfile.TemporaryDirectory() as d:
good = Path(d) / "Justfile"
good.write_text(CONFORMANT, encoding="utf-8")
got = check(good)
if got:
problems.append(f" conformant fixture should pass, but failed: {[k for k, _ in got]}")
breaks = {
"pr:confirm-env": ("PR_CONFIRM", "NOPE"),
"open-pr:gate-first": ("just pr", "echo skipped"),
"open-pr:creates-pr": ("gh pr create", "echo nope"),
"open-pr:push-if-no-upstream": ("@{upstream}", "@{nothing}"),
"merge-pr:refuse-red": ("FAILURE|TIMED_OUT", "NOTHING"),
"merge-pr:refuse-empty": ("statusCheckRollup", "somethingElse"),
"merge-pr:refuse-pending": ("still running", "quiet"),
}
for key, (needle, repl) in breaks.items():
broken = Path(d) / "Broken"
broken.write_text(CONFORMANT.replace(needle, repl), encoding="utf-8")
keys = [k for k, _ in check(broken)]
if key not in keys:
problems.append(f" removing {needle!r} should trip {key}, but tripped {keys}")
commented = Path(d) / "Commented"
commented.write_text(
CONFORMANT.replace('if [ -n "${PR_CONFIRM:-}" ]; then CONFIRM="$PR_CONFIRM"',
'# PR_CONFIRM is handled elsewhere\n if false; then CONFIRM=x'),
encoding="utf-8")
if "pr:confirm-env" not in [k for k, _ in check(commented)]:
problems.append(" a COMMENT mentioning PR_CONFIRM satisfied pr:confirm-env")
missing = Path(d) / "Missing"
missing.write_text(CONFORMANT.replace("open-pr *ARGS:", "unrelated:"), encoding="utf-8")
if not any(k.startswith("open-pr") for k, _ in check(missing)):
problems.append(" a MISSING open-pr recipe did not fail the check")
if problems:
print(f"gate_conformance self-test FAILED (template v{TEMPLATE_VERSION}):", file=sys.stderr)
print("\n".join(problems), file=sys.stderr)
return 1
print(f"gate_conformance.py self-test passed (template v{TEMPLATE_VERSION})")
return 0
def main(argv):
if "--self-test" in argv:
return self_test()
args = [a for a in argv if not a.startswith("-")]
here = Path(__file__).resolve().parent.parent
candidates = [Path(args[0])] if args else [here / "Justfile", here / "justfile"]
justfile = next((c for c in candidates if c.is_file()), None)
if justfile is None:
print(f"error: no justfile found (tried {[str(c) for c in candidates]})", file=sys.stderr)
return 1
failures = check(justfile)
if failures:
print(f"gate conformance FAILED for {justfile.name} "
f"(template v{TEMPLATE_VERSION}):", file=sys.stderr)
for key, desc in failures:
print(f" [{key}] {desc}", file=sys.stderr)
print("", file=sys.stderr)
print(" These guards are shared behaviour across rusticprofile, retch and etr. Each one", file=sys.stderr)
print(" exists because it was missing once and something bad followed: a PR merged over", file=sys.stderr)
print(" a red leg, a merge over a rollup nothing had reported into, a gate no script", file=sys.stderr)
print(" could answer, an open-pr that printed 'Gate passed' and then failed.", file=sys.stderr)
return 1
print(f"gate conformance ok: pr / open-pr / merge-pr (template v{TEMPLATE_VERSION})")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))