Coverage for plugin/scripts/check_subprocess_discipline.py: 100%
87 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 14:46 +0000
1#!/usr/bin/env python3
2"""Require every `subprocess` call to notice when the process failed.
4Ruff's `S` rules police *how* a process is launched. They say nothing about the failure
5mode that actually matters in this plugin: launching git correctly and then ignoring that
6it exited non-zero. Ten of the sixteen call sites here pass ``check=False`` and read the
7returncode by hand — correct at every site today, but nothing made it stay that way, so a
8new call site could swallow a failed sync and the build would stay green.
10Three rules, checked with `ast` over `plugin/scripts/`:
121. **``check=`` must be passed explicitly.** ``subprocess.run`` defaults to
13 ``check=False``, so omitting it means "ignore failures" without ever saying so. Whether
14 a failure should raise is exactly the decision this checker wants written down.
162. **A ``check=False`` call must account for the returncode**, by one of:
17 *inspecting* it (``result.returncode`` somewhere in the enclosing function), *handing
18 the process back* (the function returns a ``CompletedProcess``, so the caller decides),
19 or *declaring the exception* with an ``rc-ignored:`` comment on the call.
213. **An ``rc-ignored:`` comment needs a reason** after the colon. A bare marker is a way
22 to silence the checker without thinking, which is what it exists to prevent.
24Rule 2's third arm is not a loophole — some calls genuinely should ignore the code.
25``git config --get user.name`` exits 1 when the key is simply unset, and the empty stdout
26*is* the answer; raising or branching there would invent a failure. The point is that such
27a case is now stated at the call site instead of being indistinguishable from an oversight.
29Usage:
30 uv run --python 3.12 --no-project python \
31 plugin/scripts/check_subprocess_discipline.py [--root DIR]
33Exits 0 when every call site is disciplined, 1 (listing each violation) otherwise.
34"""
36from __future__ import annotations
38import argparse
39import ast
40import sys
41from pathlib import Path
43sys.path.insert(0, str(Path(__file__).resolve().parent))
44from _rhiza_layout import SCRIPTS_DIR # noqa: E402
46# The `subprocess` entry points that launch a process and hand back its result. `Popen` is
47# absent on purpose: it has no `check` argument and nothing here uses it.
48_LAUNCHERS = frozenset({"run", "call", "check_call", "check_output"})
50# `check_call` and `check_output` raise on a non-zero exit by definition, so `check=` is
51# neither accepted nor needed there.
52_ALWAYS_CHECKS = frozenset({"check_call", "check_output"})
54_MARKER = "rc-ignored:"
57def _is_subprocess_launch(node: ast.Call) -> str | None:
58 """Return the `subprocess` function *node* calls, or None when it calls something else."""
59 func = node.func
60 if not isinstance(func, ast.Attribute) or func.attr not in _LAUNCHERS:
61 return None
62 if isinstance(func.value, ast.Name) and func.value.id == "subprocess":
63 return func.attr
64 return None
67def _check_keyword(node: ast.Call) -> ast.keyword | None:
68 """Return the ``check=`` keyword passed to *node*, or None when it was omitted."""
69 return next((kw for kw in node.keywords if kw.arg == "check"), None)
72def _is_false(value: ast.expr) -> bool:
73 """Is *value* the literal ``False``?"""
74 return isinstance(value, ast.Constant) and value.value is False
77def _inspects_returncode(func: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
78 """Does *func* read ``.returncode`` anywhere in its body?"""
79 return any(
80 isinstance(node, ast.Attribute) and node.attr == "returncode" for node in ast.walk(func)
81 )
84def _returns_completed_process(func: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
85 """Does *func* annotate its return type as a ``CompletedProcess``?
87 The "hand it back" arm: a function whose contract is to return the process has
88 delegated the returncode decision to its caller, which this checker then holds to the
89 same three rules.
90 """
91 if func.returns is None:
92 return False
93 return "CompletedProcess" in ast.unparse(func.returns)
96def _marker_reason(lines: list[str], node: ast.Call) -> str | None:
97 """Return the text after ``rc-ignored:`` on or just above *node*, or None.
99 The span starts one line **above** the call, because that is where the comment
100 naturally goes: a multi-line `subprocess.run(...)` often has no room for a trailing
101 comment, and the argument list is the wrong place to explain the exit code anyway.
103 An empty string means the marker is present but unexplained, which rule 3 rejects;
104 None means there is no marker at all.
105 """
106 start = max(0, node.lineno - 2)
107 for line in lines[start : (node.end_lineno or node.lineno)]:
108 if _MARKER in line:
109 return line.split(_MARKER, 1)[1].strip()
110 return None
113def _enclosing_function(
114 tree: ast.Module, node: ast.Call
115) -> ast.FunctionDef | ast.AsyncFunctionDef | None:
116 """Return the innermost function containing *node*, or None at module level.
118 Innermost matters: `_skeleton_common.git_identity` wraps its call in a nested `read`
119 helper, and it is `read` whose handling of the returncode is the question. Of the
120 functions containing the call, the innermost is the one declared last.
121 """
122 containing = [
123 candidate
124 for candidate in ast.walk(tree)
125 if isinstance(candidate, ast.FunctionDef | ast.AsyncFunctionDef)
126 and any(inner is node for inner in ast.walk(candidate))
127 ]
128 return max(containing, key=lambda func: func.lineno) if containing else None
131def _unchecked_violations(
132 tree: ast.Module, lines: list[str], node: ast.Call, name: str, where: str
133) -> list[str]:
134 """Return the violations for a call that passes `check=False`.
136 Split out of `_violations_for_call` so each half answers one question: that one asks
137 whether `check=` is present and false, this one asks whether being false is
138 *justified* — by an explanatory marker, or by the enclosing function accounting for
139 the returncode itself.
140 """
141 reason = _marker_reason(lines, node)
142 if reason is not None:
143 if reason:
144 return []
145 return [
146 f"{where}: `{_MARKER}` needs a reason after the colon, naming which "
147 "returncodes are expected and why ignoring them is right here."
148 ]
150 func = _enclosing_function(tree, node)
151 if func is not None and (_inspects_returncode(func) or _returns_completed_process(func)):
152 return []
153 return [
154 f"{where}: subprocess.{name}(..., check=False) never accounts for the returncode. "
155 "Inspect `.returncode`, return the CompletedProcess so the caller can, or add a "
156 f"`# {_MARKER} <reason>` comment on the call."
157 ]
160def _violations_for_call(
161 tree: ast.Module, lines: list[str], node: ast.Call, name: str, where: str
162) -> list[str]:
163 """Return the rule violations for one `subprocess` call."""
164 keyword = _check_keyword(node)
165 if name in _ALWAYS_CHECKS:
166 return [] # raises by definition; `check=` does not apply
167 if keyword is None:
168 return [
169 f"{where}: subprocess.{name}(...) omits `check=`. Pass it explicitly — the "
170 "default is check=False, which ignores failures without saying so."
171 ]
172 if not _is_false(keyword.value):
173 return []
174 return _unchecked_violations(tree, lines, node, name, where)
177def check_module(path: Path, root: Path) -> list[str]:
178 """Return the discipline violations in one module."""
179 text = path.read_text(encoding="utf-8")
180 lines = text.splitlines()
181 tree = ast.parse(text, filename=str(path))
182 rel = path.relative_to(root).as_posix() if path.is_relative_to(root) else path
184 violations: list[str] = []
185 for node in ast.walk(tree):
186 if not isinstance(node, ast.Call):
187 continue
188 name = _is_subprocess_launch(node)
189 if name is None:
190 continue
191 where = f"{rel}:{node.lineno}"
192 violations.extend(_violations_for_call(tree, lines, node, name, where))
193 return violations
196def check(root: Path) -> list[str]:
197 """Return every discipline violation in *root*'s bundled scripts, in file order.
199 *root* is the **repository** root; the scripts directory is derived from
200 `_rhiza_layout` rather than hardcoded, like every other checker that spans both
201 halves of the repo.
202 """
203 scripts = root / SCRIPTS_DIR
204 found: list[str] = []
205 for module in sorted(scripts.rglob("*.py")):
206 found.extend(check_module(module, root))
207 return found
210def main(argv: list[str] | None = None) -> int:
211 """Entry point: check subprocess discipline and return an exit code."""
212 parser = argparse.ArgumentParser(
213 description="Require every subprocess call to account for a non-zero exit.",
214 )
215 parser.add_argument("--root", default=".", help="Repository root (default: current directory).")
216 args = parser.parse_args(argv)
218 violations = check(Path(args.root).resolve())
220 for violation in violations:
221 print(violation, file=sys.stderr)
222 if violations:
223 print(
224 f"\n{len(violations)} subprocess call(s) do not account for failure. "
225 "See plugin/scripts/check_subprocess_discipline.py for the three rules.",
226 file=sys.stderr,
227 )
228 return 1
229 return 0
232if __name__ == "__main__":
233 raise SystemExit(main())