ci: fix PLW1509, PLW1510, PLW2101 linter errors

This commit is contained in:
Maciej Lisiewski committed 2026-10-10 15:20:13 +02:00
1 parent f925b01791
commit 0006a5e229
563 files changed
+1068 -661

No files matched your search

+1
View File
@@ -288,6 +288,7 @@ def _detect_claude_code_version() -> str:
result = subprocess.run(
[cmd, "--version"],
stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5,
check=False,
)
if result.returncode == 0 and result.stdout.strip():
version = result.stdout.strip().split()[0] # "2.1.74 (Claude Code)" or "2.1.74"
+4 -1
View File
@@ -265,6 +265,7 @@ def _find_claude_code_keychain_item() -> Optional[tuple[str, dict[str, Any]]]:
result = subprocess.run(
["security", "find-generic-password", "-s", _CLAUDE_CODE_KEYCHAIN_SERVICE, "-g"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, stdin=subprocess.DEVNULL,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
return None
@@ -294,6 +295,7 @@ def _read_claude_code_keychain_payload() -> Optional[dict[str, Any]]:
result = subprocess.run(
["security", "find-generic-password", "-s", _CLAUDE_CODE_KEYCHAIN_SERVICE, "-w"],
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5, stdin=subprocess.DEVNULL,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
logger.debug("Keychain: security command not available or timed out")
@@ -597,6 +599,7 @@ def _mirror_claude_code_credentials_to_keychain(
account, _merge_keychain_credential_payload(existing, access_token, refresh_token, expires_at_ms))
result = subprocess.run(
argv, input=line, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10,
check=False,
)
except Exception as e: # the file commit already succeeded; a Keychain hiccup must not fail the rotation
logger.debug("Keychain mirror skipped (%s)", e)
@@ -717,7 +720,7 @@ def run_oauth_setup_token() -> Optional[str]:
raise FileNotFoundError("The 'claude' CLI is not installed. Install it with: npm install -g @anthropic-ai/claude-code")
# Interactive: stdio inherited so the user can complete the OAuth prompt. noqa: subprocess-stdin
try:
subprocess.run([claude_path, "setup-token"])
subprocess.run([claude_path, "setup-token"], check=False)
except (KeyboardInterrupt, EOFError):
return None
creds = read_claude_code_credentials()
+1
View File
@@ -53,6 +53,7 @@ def _mint(command: str, label: str) -> tuple[str, Optional[float]]:
completed = subprocess.run(
command, stdin=subprocess.DEVNULL, shell=True, capture_output=True, text=True, errors="replace", timeout=_MINT_TIMEOUT_SECONDS,
env=served_profile_child_env(inherit_credentials=True),
check=False,
)
except subprocess.TimeoutExpired as exc:
raise CommandTokenError(
+1
View File
@@ -98,6 +98,7 @@ def _acp_supported(command: str, args: list[str]) -> bool | None:
capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=5,
stdin=subprocess.DEVNULL,
check=False,
)
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
return None
+1 -1
View File
@@ -218,7 +218,7 @@ def allowlisted_env() -> dict[str, str]:
def _run(argv: list[str], *, timeout: int, text: bool = False, **kwargs) -> subprocess.CompletedProcess:
if text:
kwargs.update(text=True, encoding="utf-8", errors="replace")
return subprocess.run(argv, capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL, **kwargs)
return subprocess.run(argv, capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL, **kwargs) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
def iron_proxy_version(binary: Path) -> str:
+1
View File
@@ -227,6 +227,7 @@ def run_cli(argv: Sequence[str], *, env: dict[str, str], timeout: float, label:
return subprocess.run(
list(argv), env=env, capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=timeout, stdin=stdin,
check=False,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(timeout_message) from exc
+1 -1
View File
@@ -114,7 +114,7 @@ def _git_branch(cwd: str) -> str:
import subprocess
r = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=3, cwd=cwd,
stdin=subprocess.DEVNULL)
stdin=subprocess.DEVNULL, check=False)
except Exception:
return ""
return r.stdout.strip() if r.returncode == 0 else ""
+1
View File
@@ -364,6 +364,7 @@ def check_codex_binary(
proc = subprocess.run(
[codex_bin, "--version"], capture_output=True, text=True, encoding='utf-8', errors='replace',
timeout=10, stdin=subprocess.DEVNULL,
check=False,
)
except FileNotFoundError:
return False, f"codex CLI not found at {codex_bin!r}. Install with: npm i -g @openai/codex"
+2 -2
View File
@@ -65,7 +65,7 @@ def run_with_stdin_secret(argv: Sequence[str], *, env: dict[str, str], secret: s
try:
return subprocess.run(
list(argv), env=env, input=secret + "\n", capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=timeout)
encoding="utf-8", errors="replace", timeout=timeout, check=False)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"{label} unlock timed out after {timeout:.0f}s") from exc
except OSError as exc:
@@ -81,7 +81,7 @@ def run_with_secret_env(argv: Sequence[str], *, env: dict[str, str], secret_env:
try:
return subprocess.run(
list(argv), env=child_env, stdin=subprocess.DEVNULL, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=timeout)
encoding="utf-8", errors="replace", timeout=timeout, check=False)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"{label} unlock timed out after {timeout:.0f}s") from exc
except OSError as exc:
+2 -1
View File
@@ -96,7 +96,7 @@ def _run_phase_command(
) -> PhaseResult:
started = time.monotonic()
try:
proc = subprocess.run(command, cwd=str(root), timeout=timeout, **_SUBPROCESS_KW)
proc = subprocess.run(command, cwd=str(root), timeout=timeout, **_SUBPROCESS_KW, check=False)
output, exit_code, timed_out = proc.stdout or "", proc.returncode, False
except subprocess.TimeoutExpired as exc:
raw = exc.output
@@ -198,6 +198,7 @@ def _compose_live_state_reason(root: Path) -> str | None:
result = subprocess.run(
["docker", "compose", "ps", "--status", "running", "--format", "{{.Name}}"],
cwd=root, capture_output=True, text=True, timeout=15, stdin=subprocess.DEVNULL,
check=False,
)
except FileNotFoundError:
return None
@@ -98,7 +98,7 @@ def main(binary):
assert child.wait(timeout=5) == 0
assert child.stderr.read() == b""
# Even DISPLAY pointing at X11 must not claim global access on Wayland.
denied = subprocess.run([binary, "--check"], env={**env, "WAYLAND_DISPLAY": "wayland-test"}, capture_output=True, timeout=5)
denied = subprocess.run([binary, "--check"], env={**env, "WAYLAND_DISPLAY": "wayland-test"}, capture_output=True, timeout=5, check=False)
assert json.loads(denied.stdout) == {"type": "error", "code": "unavailable"}
print("isolated XI2: summon, key/modifier/mouse/wheel/drag cancellation, duration, EOF, and Wayland checks passed")
finally:
+2
View File
@@ -194,6 +194,7 @@ def _prepare_container_image(
probe = _sp.run(
["docker", "image", "inspect", container_image],
capture_output=True, timeout=10,
check=False,
)
if probe.returncode != 0:
if config.get("verbose"):
@@ -201,6 +202,7 @@ def _prepare_container_image(
pull = _sp.run(
["docker", "pull", container_image],
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=600,
check=False,
)
if pull.returncode != 0:
return _failure_result(
+2 -1
View File
@@ -1258,7 +1258,8 @@ class HermesCLI(CLIInitMixin, CLITuiRuntimeMixin, CLIProcessNotificationsMixin,
result = subprocess.run(
exec_cmd, shell=True, capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=30, env=build_subprocess_env(),
creationflags=windows_hide_flags(), # no console flash on Windows (#56747)
creationflags=windows_hide_flags(), # no console flash on Windows (#56747),
check=False,
)
# See #56747.
output = result.stdout.strip() or result.stderr.strip()
+1 -1
View File
@@ -4072,7 +4072,7 @@ def _worktree_maintenance_repos() -> list[str]:
probe = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=5, cwd=workdir)
errors="replace", timeout=5, cwd=workdir, check=False)
if probe.returncode == 0 and probe.stdout.strip():
repos.add(probe.stdout.strip())
except Exception:
+4 -1
View File
@@ -56,9 +56,10 @@ def reset_browser_state():
subprocess.run(
["taskkill", "/F", "/IM", "agent-browser.exe", "/T"],
capture_output=True,
check=False,
)
else:
subprocess.run(["pkill", "-f", "agent-browser"], capture_output=True)
subprocess.run(["pkill", "-f", "agent-browser"], capture_output=True, check=False)
code = "cdp('Network.clearBrowserCookies')\nprint('cleared')\n"
try:
subprocess.run(
@@ -68,6 +69,7 @@ def reset_browser_state():
capture_output=True,
timeout=120,
env=ENV,
check=False,
)
except Exception:
pass
@@ -93,6 +95,7 @@ for arm, task, model, rep in cells:
text=True,
timeout=args.run_timeout,
env={**ENV, "BUBENCH_TASKS": args.tasks},
check=False,
)
rec = None
for line in (proc.stdout or "").splitlines():
+2 -1
View File
@@ -137,7 +137,7 @@ for task, model, rep in cells:
f.write(json.dumps(rec) + "\n")
continue
env = {**ENV_BASE, **extra_env, "BUBENCH_TASKS": args.tasks}
subprocess.run(["pkill", "-f", "browser_harness"], capture_output=True)
subprocess.run(["pkill", "-f", "browser_harness"], capture_output=True, check=False)
try:
proc = subprocess.run(
[PY, os.path.join(ROOT, "single_run.py"), "pr", task, model, str(rep)],
@@ -145,6 +145,7 @@ for task, model, rep in cells:
text=True,
timeout=args.run_timeout,
env=env,
check=False,
)
rec = None
for line in (proc.stdout or "").splitlines():
+3 -3
View File
@@ -18,11 +18,11 @@ for k in list(ENV):
def run(argv, code=None, timeout=300):
t0 = time.perf_counter()
r = subprocess.run([PY, *argv] if code is None else [PY, "-c", code], cwd=TREE, env=ENV, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL)
r = subprocess.run([PY, *argv] if code is None else [PY, "-c", code], cwd=TREE, env=ENV, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL, check=False)
return time.perf_counter() - t0, r
def warm_pyc():
subprocess.run([PY, "-m", "compileall", "-q", "-j", "16", TREE], cwd=TREE, env=ENV, capture_output=True)
subprocess.run([PY, "-m", "compileall", "-q", "-j", "16", TREE], cwd=TREE, env=ENV, capture_output=True, check=False)
def med(xs): return round(statistics.median(xs), 4)
@@ -59,7 +59,7 @@ for tgt in IMPORT_TARGETS:
rows = []
for _ in range(REPS):
_, r = run([], code=None) if False else (None, None)
r = subprocess.run([PY, "-c", probe, tgt], cwd=TREE, env=ENV, capture_output=True, text=True, timeout=300)
r = subprocess.run([PY, "-c", probe, tgt], cwd=TREE, env=ENV, capture_output=True, text=True, timeout=300, check=False)
try: rows.append(json.loads([l for l in (r.stdout + "\n" + r.stderr).splitlines() if l.lstrip().startswith("{") and "\"dt\"" in l][-1].strip()))
except Exception: rows.append({"dt": None, "mods": None, "rss_kb": None, "err": (r.stderr or r.stdout)[-300:]})
ok = [x for x in rows if x["dt"] is not None and not x["err"]]
@@ -133,8 +133,8 @@ def radon(files):
# radon can't take a list file; run per-directory roots instead
roots = sorted({os.path.relpath(f, TREE).split(os.sep)[0] if os.sep in os.path.relpath(f, TREE) else os.path.relpath(f, TREE) for f in files})
roots = [r for r in roots if r != "tests"]
cc = subprocess.run([R, "cc", "-j", "-s", "-e", "tests/*,tests/**", *roots], cwd=TREE, capture_output=True, text=True).stdout
mi = subprocess.run([R, "mi", "-j", "-e", "tests/*,tests/**", *roots], cwd=TREE, capture_output=True, text=True).stdout
cc = subprocess.run([R, "cc", "-j", "-s", "-e", "tests/*,tests/**", *roots], cwd=TREE, capture_output=True, text=True, check=False).stdout
mi = subprocess.run([R, "mi", "-j", "-e", "tests/*,tests/**", *roots], cwd=TREE, capture_output=True, text=True, check=False).stdout
try:
ccj = json.loads(cc); blocks = [b for v in ccj.values() if isinstance(v, list) for b in v if isinstance(b, dict) and "complexity" in b]
cs = sorted(b["complexity"] for b in blocks)
+1
View File
@@ -116,6 +116,7 @@ def codex(args: list, prompt: str, timeout: int = 3600) -> str:
proc = subprocess.run(
["codex", "exec", *args, "--skip-git-repo-check", prompt],
cwd=str(WORKDIR), capture_output=True, text=True, timeout=timeout,
check=False,
)
return proc.stdout + proc.stderr
+1 -1
View File
@@ -60,7 +60,7 @@ def run_cell(cell):
t0 = time.time()
try:
p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 60,
env=os.environ.copy())
env=os.environ.copy(), check=False)
if p.returncode == 3:
return (cell, "INFRA_ABORT", p.stderr[-500:])
if p.returncode != 0 and not os.path.exists(out):
@@ -12,7 +12,7 @@ import venv
def run(argv, env, cwd):
result = subprocess.run(argv, env=env, cwd=cwd, stdin=subprocess.DEVNULL,
capture_output=True, text=True, timeout=60)
capture_output=True, text=True, timeout=60, check=False)
return {"argv": argv, "returncode": result.returncode,
"stdout": result.stdout, "stderr": result.stderr}
+2 -2
View File
@@ -104,8 +104,8 @@ def _count_live(cls_name: str) -> int:
def _snap(pid: int, db_path: str) -> dict:
st = open(f"/proc/{pid}/status", encoding="utf-8").read()
g = lambda k: int(st.split(k + ":")[1].split()[0])
tcp = subprocess.run(f"ss -tanp 2>/dev/null | grep -c 'pid={pid},'", shell=True, capture_output=True, text=True).stdout.strip()
kids = subprocess.run(["ps", "-o", "args=", "--ppid", str(pid)], capture_output=True, text=True).stdout
tcp = subprocess.run(f"ss -tanp 2>/dev/null | grep -c 'pid={pid},'", shell=True, capture_output=True, text=True, check=False).stdout.strip()
kids = subprocess.run(["ps", "-o", "args=", "--ppid", str(pid)], capture_output=True, text=True, check=False).stdout
return {
"threads": g("Threads"), "rss_mb": g("VmRSS") // 1024, "fds": len(os.listdir(f"/proc/{pid}/fd")),
"tcp": int(tcp or 0), "pyright": kids.count("pyright"), "kernels": kids.count("hermes_kernel_runner"),
+1
View File
@@ -164,6 +164,7 @@ class ControlledAgent(AIAgent):
[sys.executable, str(Path(__file__).with_name("foreign_writer.py")),
str(ROOT), str(HOME / "state.db"), sid],
capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=20,
check=False,
)
assert result.returncode == 0, result.stderr
faults.append("foreign-writer-committed")
+1 -1
View File
@@ -11,7 +11,7 @@ repo = Path(sys.argv[1]).resolve()
if len(sys.argv) == 2:
with tempfile.TemporaryDirectory(prefix="kanban-transport-") as home:
env = {"HOME": home, "HERMES_HOME": home + "/hermes", "PATH": os.environ["PATH"], "PYTHONDONTWRITEBYTECODE": "1"}
p = subprocess.run([sys.executable, __file__, str(repo), "isolated"], cwd=home, env=env, stdin=subprocess.DEVNULL)
p = subprocess.run([sys.executable, __file__, str(repo), "isolated"], cwd=home, env=env, stdin=subprocess.DEVNULL, check=False)
sys.exit(p.returncode)
sys.path.insert(0, str(repo))
from hermes_cli import kanban_db as kb
+1 -1
View File
@@ -152,7 +152,7 @@ def main():
code = ("import runpy; ns=runpy.run_path(" + repr(str(Path(__file__).resolve()))
+ "); ns['rows_worker']()")
proc = subprocess.run([sys.executable, "-c", code], cwd=repo, env=env,
stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=45)
stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=45, check=False)
transcript = proc.stdout + proc.stderr
if proc.returncode:
raise RuntimeError(transcript)
+2 -2
View File
@@ -150,7 +150,7 @@ from hermes_cli.main import main
main()
''', "mcp", *command]
result = subprocess.run(argv,
cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=40)
cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=40, check=False)
token_path = home / "mcp-tokens" / "fixture.json"
refresh_output = None
if token_path.exists() and not previous:
@@ -158,7 +158,7 @@ main()
tokens["expires_at"] = time.time() - 60
token_path.write_text(json.dumps(tokens))
refreshed = subprocess.run([sys.executable, "-m", "hermes_cli.main", "mcp", "test", "fixture"],
cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30)
cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30, check=False)
refresh_output = refreshed.stdout + refreshed.stderr
return {"mode": mode, "returncode": result.returncode, "output": result.stdout + result.stderr,
"token_persisted": token_path.exists(), "refresh_output": refresh_output, "wire": wire,
@@ -245,7 +245,7 @@ def main() -> int:
(tmp / "home").mkdir(parents=True)
import subprocess
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace").stdout.strip()
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False).stdout.strip()
wire = _FakeResponses()
try:
result = {
+2 -2
View File
@@ -21,7 +21,7 @@ from pathlib import Path
def _git(repo: str, *args: str) -> str:
return subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True, encoding="utf-8", errors="replace").stdout
return subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True, encoding="utf-8", errors="replace", check=False).stdout
def main(argv=None) -> int:
@@ -42,7 +42,7 @@ def main(argv=None) -> int:
surface_line = "(scripts/ci/check_public_surface.py not found; merge #103541 or pass a checkout that has it)"
else:
surface = subprocess.run([sys.executable, str(checker), "--base", a.base, "--head", a.open, "--json", str(out / "surface_at_open.json")],
cwd=a.repo, capture_output=True, text=True, encoding="utf-8", errors="replace")
cwd=a.repo, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False)
surface_line = (surface.stdout.splitlines() or [f"(check_public_surface failed: {surface.stderr.strip()[:120]})"])[0]
log = _git(a.repo, "log", "--no-merges", "--format=%H%x00%s", f"{a.open}..{a.head}")
@@ -40,7 +40,7 @@ t0 = time.time()
r = ag.run_conversation(task)
print("final:", (r.get("final_response") or "")[:80], "| wall", round(time.time() - t0, 1), "s")
time.sleep(1)
log = subprocess.run(f"grep -h '\\[{sid}\\]' ~/.hermes/logs/agent.log | grep 'API call #'", shell=True, capture_output=True).stdout.decode("utf-8", "replace")
log = subprocess.run(f"grep -h '\\[{sid}\\]' ~/.hermes/logs/agent.log | grep 'API call #'", shell=True, capture_output=True, check=False).stdout.decode("utf-8", "replace")
rows = re.findall(r"API call #(\d+): .*in=(\d+) out=(\d+) .*cache=(\d+)/(\d+) \((\d+)%\)", log)
tot_in = tot_c = 0
for n, i, o, c, ct, p in rows:
@@ -57,7 +57,7 @@ task = (f"Work in {work} (create it). Before EACH tool call, think carefully for
"5) read it back; then reply DONE.")
ag.run_conversation(task)
time.sleep(1)
log = subprocess.run(f"grep -h '\\[{sid}\\]' ~/.hermes/logs/agent.log | grep 'API call #'", shell=True, capture_output=True).stdout.decode("utf-8", "replace")
log = subprocess.run(f"grep -h '\\[{sid}\\]' ~/.hermes/logs/agent.log | grep 'API call #'", shell=True, capture_output=True, check=False).stdout.decode("utf-8", "replace")
rows = re.findall(r"API call #(\d+): .*in=(\d+) out=(\d+) .*cache=(\d+)/(\d+) \((\d+)%\)", log)
for r in rows: print(f" call {r[0]:>2} in={int(r[1]):>6} out={int(r[2]):>5} cached={int(r[3]):>6} ({r[5]}%) uncached={int(r[1])-int(r[3])}")
print(f"ARM {arm}: captured {len(captured)} payloads")
@@ -37,19 +37,19 @@ print(json.dumps({'pattern_guard':a.check_dangerous_command(cmd,'local',approval
# execution and marker contents, never invoke a host shutdown binary.
marker=home/'marker'
prefix=r+'(){ printf executed > "$MARKER"; }; '
p=subprocess.run(['/bin/bash','--noprofile','--norc','-c',prefix+cmd],env={'PATH':'/usr/bin:/bin','HOME':str(home),'MARKER':str(marker)},capture_output=True,text=True, encoding='utf-8', errors='replace',timeout=5)
p=subprocess.run(['/bin/bash','--noprofile','--norc','-c',prefix+cmd],env={'PATH':'/usr/bin:/bin','HOME':str(home),'MARKER':str(marker)},capture_output=True,text=True, encoding='utf-8', errors='replace',timeout=5, check=False)
assert marker.read_text(encoding='utf-8')=='executed'
print(json.dumps({'case':'newline safe execution','exit':p.returncode,'marker':marker.read_text(encoding='utf-8'),'stdout':p.stdout,'stderr':p.stderr}))
# Prove the benign backtick argument really is well-formed and matches input.
f=home/'f';f.write_text('needle\n', encoding='utf-8')
cmd='grep -e `printf needle` '+str(f)
p=subprocess.run(['/bin/bash','--noprofile','--norc','-c',cmd],env={'PATH':'/usr/bin:/bin','HOME':str(home)},capture_output=True,text=True, encoding='utf-8', errors='replace',timeout=5)
p=subprocess.run(['/bin/bash','--noprofile','--norc','-c',cmd],env={'PATH':'/usr/bin:/bin','HOME':str(home)},capture_output=True,text=True, encoding='utf-8', errors='replace',timeout=5, check=False)
assert p.returncode==0 and p.stdout=='needle\n'
print(json.dumps({'case':'backtick argument','command':cmd,'baseline':(b.detect_hardline_command(cmd) if b else None),'head':d.detect_hardline_command(cmd),'tokens':d._shell_tokens_with_spans(cmd,0),'exit':p.returncode,'stdout':p.stdout}))
# Reporter's exact spelling, fixture makes the sed address meaningful.
with tempfile.TemporaryDirectory() as tmp:
Path(tmp,'f').write_text('X\ny\nz\nw\n', encoding='utf-8')
cmd='sed -n "$(grep -n X f | cut -d: -f1),+3p" f'
p=subprocess.run(['/bin/bash','--noprofile','--norc','-c',cmd],cwd=tmp,env={'PATH':'/usr/bin:/bin','HOME':str(home)},capture_output=True,text=True, encoding='utf-8', errors='replace',timeout=5)
p=subprocess.run(['/bin/bash','--noprofile','--norc','-c',cmd],cwd=tmp,env={'PATH':'/usr/bin:/bin','HOME':str(home)},capture_output=True,text=True, encoding='utf-8', errors='replace',timeout=5, check=False)
assert p.returncode==0 and p.stdout=='X\ny\nz\nw\n'
print(json.dumps({'case':'reported real fixture','baseline':(b.detect_hardline_command(cmd) if b else None),'head':d.detect_hardline_command(cmd),'exit':p.returncode,'stdout':p.stdout}))
+1 -1
View File
@@ -48,7 +48,7 @@ def run_probe(script: str, args: list[str], repo: str, timeout: int = 240) -> tu
env["PYTHONPATH"] = repo + os.pathsep + env.get("PYTHONPATH", "")
cmd = [sys.executable, str(HERE / script), *[a.format(repo=repo) for a in args]]
try:
p = subprocess.run(cmd, cwd=repo, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, env=env)
p = subprocess.run(cmd, cwd=repo, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, env=env, check=False)
return p.returncode, (p.stdout + "\n" + p.stderr)
except subprocess.TimeoutExpired:
return 124, "TIMEOUT"
+1 -1
View File
@@ -43,7 +43,7 @@ def run(code, *args, profile=None):
child_env['HERMES_HOME'] = str(out / profile)
result = subprocess.run([sys.executable, '-c', code, *args], cwd=repo,
env=child_env, stdin=subprocess.DEVNULL,
capture_output=True, text=True, encoding='utf-8', timeout=45)
capture_output=True, text=True, encoding='utf-8', timeout=45, check=False)
if result.returncode:
raise RuntimeError(result.stdout + result.stderr)
return json.loads(result.stdout.splitlines()[-1])
+1
View File
@@ -62,6 +62,7 @@ def extract_arm(ref: str, workdir: Path, name: str) -> Path:
out = subprocess.run(
["git", "show", f"{ref}:tools/session_search_tool.py"],
cwd=REPO_ROOT, capture_output=True, text=True,
check=False,
)
if out.returncode != 0:
raise SystemExit(f"git show {ref}: {out.stderr.strip()}")
@@ -157,7 +157,7 @@ def run(out_path: str) -> dict:
tail = walk_history[cut:]
tail_real = _count_images(tail) * IMAGE_REAL + len(tail) * TEXT_PER_TURN // 2
result = {
"head": subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace").stdout.strip(),
"head": subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False).stdout.strip(),
"image_real_cost": IMAGE_REAL, "context_length": CONTEXT_LENGTH, "threshold": THRESHOLD,
"learned_image_cost_after": learned_image_token_cost("vision-local-ab", wire.base_url),
"provider_overflows": wire.overflows, "compress_calls": compress_calls, "per_turn": per_turn,
+1 -1
View File
@@ -276,7 +276,7 @@ def main() -> int:
tmp = Path(tempfile.mkdtemp(prefix="ab-token-accounting-"))
os.environ["HERMES_HOME"] = str(tmp / "home")
(tmp / "home").mkdir(parents=True)
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace").stdout.strip()
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False).stdout.strip()
wire = _FakeChat()
result: dict = {"checkout": str(ROOT), "head": head,
"compressor_sha256": hashlib.sha256(
@@ -84,6 +84,7 @@ if len(sys.argv) == 1:
stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
timeout=150,
check=False,
)
print(name, "exit", p.returncode, flush=True)
if p.returncode:
+1 -1
View File
@@ -201,7 +201,7 @@ mode = "overwrite"
[sys.executable, "-m", "hermes_cli.main", "chat", "--query", q,
"--quiet", "--max-turns", "30", "--accept-hooks", "--model", model],
cwd=work, env=env, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=600)
encoding="utf-8", errors="replace", timeout=600, check=False)
out = (p.stdout or "").strip()
rc = p.returncode
except subprocess.TimeoutExpired:
+1 -1
View File
@@ -31,7 +31,7 @@ if catchup:
unit = f"hermes-serve-audit-089c35aa-{tag}"
cmd = ["systemctl", "--user"]
def run(args):
return subprocess.run(args, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=65)
return subprocess.run(args, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=65, check=False)
def pid():
return run(cmd + ["show", unit, "--property=MainPID", "--value"]).stdout.strip()
record: dict[str, object] = {"repo": repo, "tag": tag, "unit": unit, "home": str(home), "tier": "native disposable systemd service"}
+1 -1
View File
@@ -29,7 +29,7 @@ import aiohttp
def _git_head() -> str:
return subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True,
stdin=subprocess.DEVNULL).stdout.strip()
stdin=subprocess.DEVNULL, check=False).stdout.strip()
def _sign(secret: str, msg_id: str, ts: str, body: bytes) -> str:
+1 -1
View File
@@ -74,7 +74,7 @@ def transcode_to_ogg_opus(path: str, *, bitrate: str = "32k", timeout: int = 60,
[ffmpeg, "-v", "error", "-y", "-i", str(path),
"-acodec", "libopus", "-ac", "1", "-b:a", bitrate, "-vbr", "on",
"-application", "voip", "-compression_level", "10", "-f", "ogg", work_path],
capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL)
capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL, check=False)
if result.returncode == 0 and os.path.getsize(work_path) > 0:
if in_place:
os.replace(work_path, ogg_path)
+1 -1
View File
@@ -113,7 +113,7 @@ def _remux_aac_to_m4a(aac_data: bytes) -> Optional[tuple[bytes, str]]:
dst_path = src_path[:-4] + ".m4a"
try:
proc = subprocess.run([ffmpeg, "-y", "-loglevel", "error", "-i", src_path, "-c:a", "copy", "-movflags",
"+faststart", dst_path], capture_output=True, timeout=10)
"+faststart", dst_path], capture_output=True, timeout=10, check=False)
if proc.returncode != 0:
logger.warning("Signal: AAC→M4A remux failed (ffmpeg exit %d): %s",
proc.returncode, proc.stderr.decode("utf-8", "replace")[:300])
+1
View File
@@ -193,6 +193,7 @@ class WebhookRouteProcessor:
result = subprocess.run(
[interpreter, str(path)], input=json.dumps(payload), capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=self.script_timeout_seconds, cwd=str(path.parent), env=build_subprocess_env(), **popen_kwargs,
check=False,
)
except subprocess.TimeoutExpired:
logger.warning("[webhook] script timed out: %s", path)
+1
View File
@@ -232,6 +232,7 @@ def _systemd_timeout_stop_us(unit_name: str) -> Optional[int]:
result = subprocess.run(
["systemctl", *flag, "show", unit_name, "--property=TimeoutStopUSec"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=2.0,
check=False,
)
except (subprocess.TimeoutExpired, OSError):
continue
+2
View File
@@ -429,6 +429,7 @@ def terminate_pid(
result = subprocess.run(
["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=10, creationflags=windows_hide_flags(),
check=False,
)
except FileNotFoundError:
os.kill(pid, signal.SIGTERM)
@@ -528,6 +529,7 @@ def _read_process_cmdline(pid: int) -> Optional[str]:
result = subprocess.run(
["ps", "-p", str(pid), "-o", "command="],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
check=False,
)
if result.returncode == 0 and result.stdout.strip():
return result.stdout.strip()
+4 -4
View File
@@ -336,7 +336,7 @@ def _is_git(path: str) -> bool:
try:
probe = subprocess.run([path, "--version"], capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=30, stdin=subprocess.DEVNULL,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), check=False)
except (OSError, ValueError, subprocess.SubprocessError):
return False
return probe.returncode == 0 and probe.stdout.startswith("git version")
@@ -753,7 +753,7 @@ def _held_open_lsof(path: Path, root: Path | None, any_git: bool) -> _Holder | b
try:
found = subprocess.run([lsof, "-F", "pc", "--", str(path)], capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=20,
stdin=subprocess.DEVNULL)
stdin=subprocess.DEVNULL, check=False)
except (OSError, subprocess.SubprocessError):
return None
pids = [line[1:] for line in found.stdout.splitlines() if line.startswith("p")]
@@ -769,7 +769,7 @@ def _lsof_cwds(lsof: str, pids: list[str]) -> dict[str, str] | None:
"""Each pid's cwd from ``lsof -d cwd`` (macOS has no /proc); None when lsof cannot answer."""
try:
out = subprocess.run([lsof, "-a", "-d", "cwd", "-F", "pn", "-p", ",".join(pids)], capture_output=True,
text=True, encoding="utf-8", errors="replace", timeout=20, stdin=subprocess.DEVNULL)
text=True, encoding="utf-8", errors="replace", timeout=20, stdin=subprocess.DEVNULL, check=False)
except (OSError, subprocess.SubprocessError):
return None
if out.returncode not in (0, 1):
@@ -813,7 +813,7 @@ def _ps_git_holder(lsof: str, git_dir: Path, root: Path, any_git: bool) -> _Hold
try:
out = subprocess.run([ps_bin, "-A", *(f"-o{c}=" for c in columns)], capture_output=True,
text=True, encoding="utf-8", errors="replace",
timeout=20, stdin=subprocess.DEVNULL)
timeout=20, stdin=subprocess.DEVNULL, check=False)
except (OSError, subprocess.SubprocessError):
return None
if out.returncode != 0:
+1 -1
View File
@@ -156,7 +156,7 @@ def _run_in_custody(argv: list, root: Path, **kwargs) -> int:
# the rest of a committed update (sync + build), and it holds the checkout lock fd, so the
# checkout stays locked for as long as it runs, even past this updater's death.
# health: allow HX006 -- unbounded by design: the update's own supervised child
return subprocess.run(argv, **kwargs, **_checkout_custody(root)).returncode
return subprocess.run(argv, **kwargs, **_checkout_custody(root)).returncode # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
proc = subprocess.Popen(argv, creationflags=lock.CREATE_SUSPENDED, **kwargs)
try:
refusal = bind(proc)
+1 -1
View File
@@ -95,7 +95,7 @@ def cli_version(binary: Path) -> str:
"""Return the first line of ``<binary> --version`` or ``"version unknown"``."""
try:
res = subprocess.run([str(binary), "--version"], capture_output=True, text=True, encoding='utf-8',
errors='replace', timeout=5)
errors='replace', timeout=5, check=False)
if res.returncode == 0:
return (res.stdout or res.stderr).strip().splitlines()[0]
except (OSError, subprocess.TimeoutExpired):
+1
View File
@@ -615,6 +615,7 @@ def posix_is_zombie(pid: int) -> bool:
r = subprocess.run(
["ps", "-o", "state=", "-p", str(pid)],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
check=False,
)
return r.returncode == 0 and r.stdout.strip().startswith("Z")
except Exception:
+2 -1
View File
@@ -88,7 +88,7 @@ def _head_sha(root: Path) -> str:
try:
head = subprocess.run([_git_executable(), "-C", str(root), "rev-parse", "HEAD"], capture_output=True,
text=True, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, timeout=60)
text=True, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, timeout=60, check=False)
if head.returncode == 0 and _SHA.fullmatch(head.stdout.strip()):
return head.stdout.strip()
except (OSError, subprocess.SubprocessError):
@@ -139,6 +139,7 @@ def main() -> int:
return subprocess.run(
[str(python), "-I", "-B", "-X", "utf8", str(root / "hermes_cli/update_serve_resume.py"),
str(context), str(result)], cwd=root, env=activation_environment(root),
check=False,
).returncode
from hermes_cli import update_receipt
from hermes_cli.update_lock import UpdateLock, describe_holder
+1 -1
View File
@@ -285,7 +285,7 @@ def _detect_default_windows() -> str | None:
def _run_stdout(argv: list[str]) -> str | None:
try:
return subprocess.run(argv, capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=5).stdout
errors="replace", timeout=5, check=False).stdout
except Exception:
return None
+1 -1
View File
@@ -146,7 +146,7 @@ def _detect_openclaw_processes() -> list[str]:
def _posix_probe(cmd: list[str], timeout: int):
try:
return subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8',
errors='replace', timeout=timeout)
errors='replace', timeout=timeout, check=False)
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
+1 -1
View File
@@ -1203,7 +1203,7 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
try:
result = subprocess.run(
["git", "worktree", "list"], capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=10, cwd=repo_root)
errors="replace", timeout=10, cwd=repo_root, check=False)
out = result.stdout.strip() if result.returncode == 0 else ""
except Exception:
out = ""
+8 -8
View File
@@ -30,7 +30,7 @@ def _probe(argv: list, timeout: int, ok, *, missing: str | None = None) -> bool:
"""Run a text-mode probe; True when it ran and ``ok(result)`` holds.
A missing executable logs *missing* (when given); every other failure is silent."""
try:
return bool(ok(subprocess.run(argv, timeout=timeout, **_TEXT)))
return bool(ok(subprocess.run(argv, timeout=timeout, **_TEXT, check=False)))
except FileNotFoundError:
if missing:
logger.debug(missing)
@@ -110,7 +110,7 @@ def write_clipboard_text(text: str) -> bool:
for argv, kw in _write_clipboard_commands(text.encode("utf-8")):
try:
if subprocess.run(argv, timeout=10, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, **kw).returncode == 0:
stderr=subprocess.DEVNULL, **kw, check=False).returncode == 0:
return True
except (OSError, subprocess.SubprocessError):
continue
@@ -122,7 +122,7 @@ def write_clipboard_text(text: str) -> bool:
def _osascript(expr: str, timeout: int = 3) -> str:
"""stdout of an osascript expression; "" on any failure."""
try:
r = subprocess.run(["osascript", "-e", expr], timeout=timeout, **_TEXT)
r = subprocess.run(["osascript", "-e", expr], timeout=timeout, **_TEXT, check=False)
return r.stdout if r.returncode == 0 else ""
except Exception as e:
logger.debug("osascript probe failed: %s", e)
@@ -163,7 +163,7 @@ def _macos_save_file_image(dest: Path) -> bool:
def _macos_pngpaste(dest: Path) -> bool:
"""pngpaste (brew install pngpaste) — fastest, cleanest."""
try:
r = subprocess.run(["pngpaste", str(dest)], stdin=subprocess.DEVNULL, capture_output=True, timeout=3)
r = subprocess.run(["pngpaste", str(dest)], stdin=subprocess.DEVNULL, capture_output=True, timeout=3, check=False)
return r.returncode == 0 and _nonempty(dest)
except FileNotFoundError:
pass # pngpaste not installed
@@ -188,7 +188,7 @@ on error
end try
'''
try:
r = subprocess.run(["osascript", "-e", script], timeout=5, **_TEXT)
r = subprocess.run(["osascript", "-e", script], timeout=5, **_TEXT, check=False)
return r.returncode == 0 and "fail" not in r.stdout and _nonempty(dest)
except Exception as e:
logger.debug("osascript clipboard extract failed: %s", e)
@@ -255,7 +255,7 @@ def _ps_clipboard(exe: str, timeout: int, label: str, dest: Path | None = None)
for check, extract in _PS_IMAGE_STRATEGIES:
try:
argv = [exe, *_PS_FLAGS, "-Command", check if dest is None else extract]
r = subprocess.run(argv, timeout=timeout, **_TEXT)
r = subprocess.run(argv, timeout=timeout, **_TEXT, check=False)
if dest is None:
if r.returncode == 0 and "True" in r.stdout:
return True
@@ -327,7 +327,7 @@ def _wayland_has_image() -> bool:
def _wayland_save(dest: Path) -> bool:
try:
types_r = subprocess.run(_WL_LIST_TYPES, timeout=3, **_TEXT)
types_r = subprocess.run(_WL_LIST_TYPES, timeout=3, **_TEXT, check=False)
types = types_r.stdout.splitlines() if types_r.returncode == 0 else ()
mime = next((m for m in _WAYLAND_MIME_PREFERENCE if m in types), None) # PNG preferred
if not mime:
@@ -361,7 +361,7 @@ def _convert_to_png(path: Path) -> bool:
try:
path.rename(tmp)
r = subprocess.run(["convert", str(tmp), "png:" + str(path)], stdin=subprocess.DEVNULL, capture_output=True,
timeout=5)
timeout=5, check=False)
if r.returncode == 0 and _nonempty(path):
tmp.unlink(missing_ok=True)
return True
+1 -1
View File
@@ -383,7 +383,7 @@ class SlashCommandCompleter(Completer):
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=2,
cwd=cwd, encoding="utf-8", errors="replace")
cwd=cwd, encoding="utf-8", errors="replace", check=False)
except (subprocess.TimeoutExpired, OSError):
continue
if proc.returncode != 0 or not proc.stdout.strip():
+1 -1
View File
@@ -3188,7 +3188,7 @@ def edit_config():
return
print(f"Opening {config_path} in {editor}...")
subprocess.run([editor, str(config_path)])
subprocess.run([editor, str(config_path)], check=False)
def _default_value_for_key(dotted_key: str):
+1 -1
View File
@@ -139,7 +139,7 @@ def _probe_gh_cli_token() -> Optional[str]:
try:
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8',
errors='replace', timeout=5, env=clean_env,
stdin=subprocess.DEVNULL, **_popen_kwargs)
stdin=subprocess.DEVNULL, **_popen_kwargs, check=False)
except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
logger.debug("gh CLI token lookup failed (%s): %s", gh_path, exc)
continue
+4 -4
View File
@@ -57,7 +57,7 @@ def _iter_process_table() -> list[tuple[int, str]]:
_append_row(rows, line[len("ProcessId=") :], current_cmd)
return rows
# ps, not `pgrep -f "hermes.*dashboard"` (greedy regex; consistent with gateway pid scan).
result = subprocess.run(["ps", "-A", "-o", "pid=,command="], timeout=10, **_PS_RUN_KWARGS)
result = subprocess.run(["ps", "-A", "-o", "pid=,command="], timeout=10, **_PS_RUN_KWARGS, check=False)
if result.returncode == 0:
for line in getattr(result, "stdout", "").split("\n"):
parts = line.strip().split(None, 1)
@@ -440,7 +440,7 @@ def _kill_pids_windows(pids: list[int], killed: list[int], failed: list[tuple[in
result = subprocess.run(
["taskkill", "/PID", str(pid), "/F"], capture_output=True,
stdin=subprocess.DEVNULL, text=True, encoding="utf-8",
errors="replace", timeout=10, creationflags=windows_hide_flags())
errors="replace", timeout=10, creationflags=windows_hide_flags(), check=False)
if result.returncode == 0:
killed.append(pid)
else:
@@ -500,7 +500,7 @@ def _posix_descendants(roots: list[int]) -> dict[int, tuple[int, int | None]]:
"""
from gateway.status import get_process_start_time
try:
result = subprocess.run(["ps", "-A", "-o", "pid=,ppid=,tty="], timeout=10, **_PS_RUN_KWARGS)
result = subprocess.run(["ps", "-A", "-o", "pid=,ppid=,tty="], timeout=10, **_PS_RUN_KWARGS, check=False)
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
return {}
children: dict[int, list[tuple[int, str]]] = {}
@@ -814,7 +814,7 @@ def _process_ppid(pid: int) -> int | None:
try:
if sys.platform == "win32":
return None
result = subprocess.run(["ps", "-o", "ppid=", "-p", str(pid)], timeout=5, **_PS_RUN_KWARGS)
result = subprocess.run(["ps", "-o", "ppid=", "-p", str(pid)], timeout=5, **_PS_RUN_KWARGS, check=False)
if result.returncode != 0 or not result.stdout:
return None
return int(result.stdout.strip().split()[0])
+2 -1
View File
@@ -330,7 +330,7 @@ def _macos_desktop_dr(app: Path) -> str | None:
"""Return the bundle's designated requirement string, or None on failure (a hanging codesign must never abort doctor)."""
codesign = shutil.which("codesign")
try:
proc = subprocess.run([codesign, "-d", "--requirements", "-", str(app)], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15) if codesign else None
proc = subprocess.run([codesign, "-d", "--requirements", "-", str(app)], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15, check=False) if codesign else None
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
return None if proc is None or proc.returncode != 0 else (proc.stdout or "") + (proc.stderr or "")
@@ -549,6 +549,7 @@ def _check_web_dashboard_import(should_fix: bool, f: Finding) -> None:
timeout=120,
cwd=str(PROJECT_ROOT),
env=env,
check=False,
)
except subprocess.TimeoutExpired:
_fail_and_issue(
+2 -1
View File
@@ -145,6 +145,7 @@ def check_legacy_desktop_checkout() -> None:
return subprocess.run(
["git", "-C", str(checkout), *args],
capture_output=True, text=True, encoding="utf-8", timeout=10,
check=False,
)
except (OSError, subprocess.SubprocessError):
return None
@@ -566,7 +567,7 @@ def _gh_authenticated() -> bool:
if not _safe_which("gh"):
return False
try:
result = subprocess.run(["gh", "auth", "status"], capture_output=True, timeout=10)
result = subprocess.run(["gh", "auth", "status"], capture_output=True, timeout=10, check=False)
return result.returncode == 0
except (OSError, subprocess.TimeoutExpired):
return False
+2 -2
View File
@@ -77,7 +77,7 @@ def _doctor_tool(name: str) -> tuple[str | None, str]:
def _run_ok(cmd: list[str], timeout: int, **kw) -> bool:
"""True when *cmd* exits 0 within *timeout*; a timeout counts as failure."""
try:
return subprocess.run(cmd, capture_output=True, timeout=timeout, **kw).returncode == 0
return subprocess.run(cmd, capture_output=True, timeout=timeout, **kw).returncode == 0 # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
except subprocess.TimeoutExpired:
return False
@@ -431,7 +431,7 @@ def _audit_one(npm_bin: str, npm_dir, label: str, audit_extra: list[str], issues
# Resolved absolute path so Windows can execute npm.cmd (CreateProcessW can't run bare .cmd names).
audit_result = subprocess.run([npm_bin, "audit", "--json", *audit_extra], cwd=str(npm_dir),
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=30,
env=with_hermes_node_path())
env=with_hermes_node_path(), check=False)
audit_data = json.loads(audit_result.stdout) if audit_result.stdout.strip() else {}
counts = audit_data.get("metadata", {}).get("vulnerabilities", {})
critical, high, moderate = (counts.get(k, 0) for k in ("critical", "high", "moderate"))
+1 -1
View File
@@ -45,7 +45,7 @@ def _git_output(project_root: Path, *args: str) -> str:
"""Stripped stdout of ``git <args>`` run in *project_root*, or '' on any failure."""
try:
result = subprocess.run(["git", *args], capture_output=True, text=True, encoding='utf-8', errors='replace',
timeout=5, cwd=str(project_root))
timeout=5, cwd=str(project_root), check=False)
return result.stdout.strip() if result.returncode == 0 else ""
except Exception:
return ""
+10 -7
View File
@@ -148,6 +148,7 @@ def _get_service_pids(all_profiles: bool = False) -> set:
+ ["list-units", pattern, "--plain", "--no-legend", "--no-pager"],
timeout=5,
**_CAPTURE_TEXT,
check=False,
)
for line in result.stdout.strip().splitlines():
parts = line.split()
@@ -159,6 +160,7 @@ def _get_service_pids(all_profiles: bool = False) -> set:
scope_args + ["show", svc, "--property=MainPID", "--value"],
timeout=5,
**_CAPTURE_TEXT,
check=False,
)
pid = int(show.stdout.strip())
if pid > 0:
@@ -189,7 +191,7 @@ def _get_service_pids(all_profiles: bool = False) -> set:
# Prefix scan also catches ai.hermes.gateway* agents the label derivation can't map
# (renamed profiles, other installs). Over-inclusion is safe: PIDs are only protected.
try:
result = subprocess.run(["launchctl", "list"], timeout=5, **_CAPTURE_TEXT)
result = subprocess.run(["launchctl", "list"], timeout=5, **_CAPTURE_TEXT, check=False)
if result.returncode == 0:
for line in result.stdout.strip().splitlines():
parts = line.split()
@@ -221,7 +223,7 @@ def _get_parent_pid(pid: int) -> int | None:
if is_windows() or not shutil.which("ps"):
return None
try:
result = subprocess.run(["ps", "-o", "ppid=", "-p", str(pid)], timeout=5, **_CAPTURE_TEXT)
result = subprocess.run(["ps", "-o", "ppid=", "-p", str(pid)], timeout=5, **_CAPTURE_TEXT, check=False)
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
raw = result.stdout.strip()
@@ -646,7 +648,7 @@ def _scan_gateway_pids(
if not _found_via_proc:
# ``-Aww`` not ``-A eww``: BSD/macOS ps rejects ``e``; ``-ww`` = unlimited width.
result = subprocess.run(["ps", "-Aww", "-o", "pid=,command="], timeout=10, **_CAPTURE_TEXT)
result = subprocess.run(["ps", "-Aww", "-o", "pid=,command="], timeout=10, **_CAPTURE_TEXT, check=False)
if result.returncode != 0:
return []
for line in result.stdout.split("\n"):
@@ -1490,7 +1492,7 @@ def _launchd_print_service_pid(domain: str, label: str) -> tuple[bool, int | Non
launchctl call must be reported, not read as "unloaded").
"""
try:
result = subprocess.run(["launchctl", "print", f"{domain}/{label}"], timeout=5, **_CAPTURE_TEXT)
result = subprocess.run(["launchctl", "print", f"{domain}/{label}"], timeout=5, **_CAPTURE_TEXT, check=False)
except FileNotFoundError:
return (False, None)
if result.returncode != 0:
@@ -1501,7 +1503,7 @@ def _launchd_print_service_pid(domain: str, label: str) -> tuple[bool, int | Non
def _launchd_service_registered(label: str, *, timeout: int = 5) -> bool:
"""True when launchd knows ``label`` (``launchctl list`` exit 0). Domain-agnostic, so still true on
macOS 26+ hosts whose per-user domains reject management. FileNotFoundError/TimeoutExpired propagate."""
result = subprocess.run(["launchctl", "list", label], timeout=timeout, **_CAPTURE_TEXT)
result = subprocess.run(["launchctl", "list", label], timeout=timeout, **_CAPTURE_TEXT, check=False)
return result.returncode == 0
@@ -2125,6 +2127,7 @@ def _windows_scheduled_task_state(task_name: str) -> str | None:
[powershell, "-NoProfile", "-NonInteractive", "-Command", ps_cmd],
capture_output=True, text=True, encoding="utf-8", errors="ignore", timeout=10,
creationflags=windows_hide_flags(),
check=False,
)
if result.returncode != 0:
return None
@@ -2533,7 +2536,7 @@ def _run_systemctl(args: list[str], *, system: bool = False, **kwargs) -> subpro
"""Run systemctl; raise RuntimeError (not raw FileNotFoundError) if missing, for callers bypassing
``supports_systemd_services()``."""
try:
return subprocess.run(_systemctl_cmd(system) + args, **kwargs)
return subprocess.run(_systemctl_cmd(system) + args, **kwargs) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
except FileNotFoundError:
from hermes_cli.gateway_command_errors import SystemctlUnavailableError
raise SystemctlUnavailableError() from None
@@ -3964,7 +3967,7 @@ def systemd_status(deep: bool = False, system: bool = False, full: bool = False)
log_cmd = ["journalctl"] + ([] if system else ["--user"]) + ["-u", svc, "-n", "20", "--no-pager"]
if full:
log_cmd.append("-l")
subprocess.run(log_cmd, timeout=10)
subprocess.run(log_cmd, timeout=10, check=False)
# =============================================================================
+3 -3
View File
@@ -49,7 +49,7 @@ def _probe_launchd_domain_for_label(label: str) -> str:
# Not loaded anywhere: Aqua → gui/<uid>; anything else (Background, loginwindow) → user/<uid>,
# the pre-probing default and the recommended domain on macOS 26+.
try:
result = subprocess.run(["launchctl", "managername"], timeout=5, **_gw()._CAPTURE_TEXT)
result = subprocess.run(["launchctl", "managername"], timeout=5, **_gw()._CAPTURE_TEXT, check=False)
if "Aqua" in (result.stdout or ""):
return gui_domain
except launchctl_errors:
@@ -902,7 +902,7 @@ def launchd_status(deep: bool = False):
plist_path = _gw().get_launchd_plist_path()
label = _gw().get_launchd_label()
try:
result = subprocess.run(["launchctl", "list", label], timeout=10, **_gw()._CAPTURE_TEXT)
result = subprocess.run(["launchctl", "list", label], timeout=10, **_gw()._CAPTURE_TEXT, check=False)
service_listed = result.returncode == 0
list_output = result.stdout
except subprocess.TimeoutExpired:
@@ -960,4 +960,4 @@ def launchd_status(deep: bool = False):
if log_file.exists():
print()
print("Recent logs:")
subprocess.run(["tail", "-20", str(log_file)], timeout=10)
subprocess.run(["tail", "-20", str(log_file)], timeout=10, check=False)
+1
View File
@@ -195,6 +195,7 @@ def _exec_schtasks(args: list[str]) -> tuple[int, str, str]:
proc = subprocess.run(
[schtasks, *args], capture_output=True, text=False,
timeout=_SCHTASKS_TIMEOUT_S, creationflags=windows_hide_flags(),
check=False,
)
return (
proc.returncode,
+5 -5
View File
@@ -72,7 +72,7 @@ def _gh_cli_token() -> Optional[str]:
env.pop(var, None)
result = subprocess.run(
[gh, "auth", "token"], capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=10, stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags())
timeout=10, stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags(), check=False)
except (OSError, subprocess.TimeoutExpired) as exc:
logger.debug("gh auth token lookup failed: %s", exc)
return None
@@ -102,7 +102,7 @@ def _credential_fill(origin: str) -> Optional[tuple[str, str]]:
result = subprocess.run(
[git, "-c", "core.askPass=", "credential", "fill"], input=request, capture_output=True,
text=True, encoding="utf-8", errors="replace", timeout=15, env=env,
creationflags=windows_hide_flags())
creationflags=windows_hide_flags(), check=False)
except (OSError, subprocess.TimeoutExpired) as exc:
logger.debug("git credential fill failed for %s: %s", origin, exc)
return None
@@ -204,13 +204,13 @@ def run_git_with_credential_fallback(
# Username", and the refusal below is never classified. Same drop _credential_fill does.
env.pop("GIT_ASKPASS", None)
env.pop("SSH_ASKPASS", None)
result = subprocess.run(argv, env=env, **run_kwargs)
result = subprocess.run(argv, env=env, **run_kwargs) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
if result.returncode == 0 or not url or not is_credential_required_error(result):
return result
auth_env = with_git_auth(env, url)
if auth_env == env:
return result
result = subprocess.run(argv, env=auth_env, **run_kwargs)
result = subprocess.run(argv, env=auth_env, **run_kwargs) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
sent = _auth_headers(auth_env)
rejected: list[str] = []
# The first credential (typically GITHUB_TOKEN from .env) was refused too: it is stale or
@@ -226,7 +226,7 @@ def run_git_with_credential_fallback(
sent |= headers
logger.warning("%s rejected the credential from %s; retrying with %s",
_https_origin(url), ", ".join(rejected) or "the stored credential", source)
result = subprocess.run(argv, env=candidate_env, **run_kwargs)
result = subprocess.run(argv, env=candidate_env, **run_kwargs) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
if result.returncode != 0 and is_credential_required_error(result):
rejected.append(source)
if result.returncode != 0 and "GITHUB_TOKEN/GH_TOKEN" in rejected and isinstance(result.stderr, str):
+1
View File
@@ -49,6 +49,7 @@ def _gh_cli_token() -> Optional[str]:
result = subprocess.run(
["gh", "auth", "token"], capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=_GH_CLI_TIMEOUT_SECONDS, stdin=subprocess.DEVNULL, creationflags=windows_hide_flags(),
check=False,
)
except (OSError, subprocess.SubprocessError) as exc:
logger.debug("gh CLI token lookup unavailable: %s", exc)
+4 -4
View File
@@ -52,7 +52,7 @@ def _git_proc_running() -> bool:
argv = (["tasklist", "/FI", "IMAGENAME eq git.exe", "/FO", "CSV"]
if os.name == "nt" else ["pgrep", "-x", "git"])
proc = subprocess.run(argv, capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=10, creationflags=windows_hide_flags())
timeout=10, creationflags=windows_hide_flags(), check=False)
if os.name == "nt":
return "git.exe" in proc.stdout.lower()
return proc.returncode == 0
@@ -761,7 +761,7 @@ def convert_treeless_checkout(repo_root: Path, **run_kwargs) -> bool:
"--filter=blob:none", "origin"]
fetch_kwargs = dict(cwd=str(repo_root), capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=900, **run_kwargs)
subprocess.run([*refetch, "--tags"], check=False, **fetch_kwargs)
subprocess.run([*refetch, "--tags"], **fetch_kwargs) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **run_kwargs; an explicit check=False would raise TypeError
if not _history_trees_complete(repo_root, **run_kwargs):
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(repo_root), check=True,
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30,
@@ -800,7 +800,7 @@ def convert_treeless_checkout_first(repo_root: Path) -> None:
def _git_version(**run_kwargs) -> tuple:
out = subprocess.run(["git", "--version"], capture_output=True, text=True, encoding="utf-8",
out = subprocess.run(["git", "--version"], capture_output=True, text=True, encoding="utf-8", # noqa: PLW1510 -- forwarding wrapper: **run_kwargs may carry check
errors="replace", timeout=30, **run_kwargs).stdout
match = re.search(r"(\d+)\.(\d+)", out)
return (int(match[1]), int(match[2])) if match else (0, 0)
@@ -809,7 +809,7 @@ def _git_version(**run_kwargs) -> tuple:
def _history_trees_complete(repo_root: Path, **run_kwargs) -> bool:
"""Whether every tree in HEAD's history is local, checked without fetching (~3 s for this repo)."""
env = dict(run_kwargs.pop("env", None) or noninteractive_git_env(), GIT_NO_LAZY_FETCH="1")
walk = subprocess.run(
walk = subprocess.run( # noqa: PLW1510 -- forwarding wrapper: **run_kwargs may carry check
["git", "rev-list", "--objects", "--filter=blob:none", "--missing=print", "HEAD"],
cwd=str(repo_root), capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=300, env=env, **run_kwargs,
+1
View File
@@ -403,6 +403,7 @@ def run_gate(gate: GoalGate, *, cwd: Optional[str] = None) -> tuple[bool, int, s
proc = subprocess.run(
gate.command, shell=True, capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=max(1, int(gate.timeout_seconds)), cwd=cwd or None,
check=False,
)
combined = (proc.stdout or "") + (("\n" + proc.stderr) if proc.stderr else "")
return proc.returncode == 0, proc.returncode, combined[-_GATE_OUTPUT_TAIL_CHARS:]
+2 -1
View File
@@ -550,9 +550,10 @@ def _cleanup_worker_tmux(conn: sqlite3.Connection, task_id: str) -> None:
out = subprocess.run(
["tmux", "list-panes", "-t", session, "-F", "#{pane_dead}"],
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5,
check=False,
)
if out.stdout.strip() == "1":
subprocess.run(["tmux", "kill-session", "-t", session], capture_output=True, timeout=5)
subprocess.run(["tmux", "kill-session", "-t", session], capture_output=True, timeout=5, check=False)
_kb._log.debug("Killed stale tmux session: %s", session)
except Exception:
pass # best-effort — never block completion
+1 -1
View File
@@ -29,7 +29,7 @@ def probe_devices(engine_dir: Path, backend: str) -> list[dict]:
out = subprocess.run(
[sys.executable, "-I", str(Path(__file__).resolve()), str(engine_dir), backend],
stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=20, creationflags=windows_hide_flags())
errors="replace", timeout=20, creationflags=windows_hide_flags(), check=False)
devices = json.loads(out.stdout) if out.returncode == 0 else []
except (OSError, ValueError, subprocess.TimeoutExpired):
return []
+4 -3
View File
@@ -67,7 +67,8 @@ _DEVICE_LINE_RE = re.compile(r"CUDA\d+:.*\((\d+)\s*MiB,\s*\d+\s*MiB free\)\s*$")
def _stdout(*argv: str) -> str:
return subprocess.run(
list(argv), capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5
list(argv), capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
check=False,
).stdout
@@ -244,7 +245,7 @@ def _cached_nvidia_gpu_query(ttl_s: float = _GPU_QUERY_TTL_S) -> dict | None:
[exe, "--query-gpu=memory.total,memory.free,name,pci.device_id,memory.used,utilization.gpu",
"--format=csv,noheader,nounits"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10,
creationflags=windows_hide_flags())
creationflags=windows_hide_flags(), check=False)
if out.returncode != 0 or not out.stdout.strip():
_gpu_query_cache = (now, None)
return None
@@ -325,7 +326,7 @@ def _engine_device_pool() -> tuple[int, bool | None] | None:
return None
exe = engine.binary
out = subprocess.run([str(exe), "--list-devices"], capture_output=True,
text=True, encoding="utf-8", errors="replace", timeout=30, cwd=str(exe.parent))
text=True, encoding="utf-8", errors="replace", timeout=30, cwd=str(exe.parent), check=False)
if out.returncode != 0:
return None
for line in (out.stdout + out.stderr).splitlines():
+1
View File
@@ -259,6 +259,7 @@ def _passes_boot_gate(staged: Path, venv_dir: Path) -> bool:
text=True, encoding="utf-8", errors="replace",
timeout=30,
env=env,
check=False,
)
except OSError as exc:
if exc.errno in (errno.ENOENT, errno.ENOEXEC):
+1
View File
@@ -1282,6 +1282,7 @@ def _resolve_workspace_key() -> Optional[str]:
result = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
check=False,
)
if result.returncode == 0 and result.stdout.strip():
return os.path.abspath(result.stdout.strip())
+1 -1
View File
@@ -15,7 +15,7 @@ def _probe_container(cmd: list, backend: str, via_sudo: bool = False):
all other exceptions propagate naturally.
"""
try:
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15)
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15, check=False)
except subprocess.TimeoutExpired:
label = f"sudo {backend}" if via_sudo else backend
print(
+1 -1
View File
@@ -87,7 +87,7 @@ _SYSTEMCTL_ERRORS = (FileNotFoundError, subprocess.TimeoutExpired, OSError)
def _run_probe(cmd: list[str], *, timeout: int) -> subprocess.CompletedProcess:
"""Captured, text-decoded ``subprocess.run`` for short local probes (systemctl, ps)."""
return subprocess.run(
cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, check=False)
def _restart_managed_dashboard_service(reason: str, unit: str = _DASHBOARD_SYSTEMD_UNIT) -> bool:
+2 -2
View File
@@ -607,7 +607,7 @@ def _desktop_macos_local_signing_identity() -> Optional[str]:
def _codesign_verify(codesign: str, app: Path, **kwargs) -> subprocess.CompletedProcess:
return subprocess.run(
return subprocess.run( # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
[codesign, "--verify", "--deep", "--strict", str(app)], capture_output=True, **kwargs)
@@ -1288,7 +1288,7 @@ def _detect_linux_password_store() -> str | None:
"org.freedesktop.DBus.Peer.Ping",
],
capture_output=True,
timeout=5)
timeout=5, check=False)
if result.returncode == 0:
return "gnome-libsecret"
return None
+2 -2
View File
@@ -114,7 +114,7 @@ def _whatsapp_install_bridge(bridge_dir) -> bool:
result = subprocess.run(
[npm, "install", "--no-fund", "--no-audit", "--progress=false"],
cwd=str(bridge_dir), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True,
encoding="utf-8", errors="replace", env=env)
encoding="utf-8", errors="replace", env=env, check=False)
except (pm.InstallError, OSError) as exc:
print(f" ✗ Bridge dependency preparation failed: {exc}")
return False
@@ -200,7 +200,7 @@ def cmd_whatsapp(args):
with contextlib.suppress(KeyboardInterrupt):
subprocess.run(
[node, str(bridge_script), "--pair-only", "--session", str(session_dir)],
cwd=str(bridge_dir), env=with_hermes_node_path())
cwd=str(bridge_dir), env=with_hermes_node_path(), check=False)
print()
if not (session_dir / "creds.json").exists():
+1 -1
View File
@@ -431,7 +431,7 @@ def _run_bootstrap(cwd: Path, commands: list[str]) -> None:
"""Execute bootstrap commands in *cwd*. Raise CatalogError on first failure."""
for cmd in commands:
_say(f" $ {cmd}", Colors.DIM)
rc = subprocess.run(cmd, cwd=str(cwd), shell=True).returncode
rc = subprocess.run(cmd, cwd=str(cwd), shell=True, check=False).returncode
if rc != 0:
raise CatalogError(f"bootstrap step failed (exit {rc}): {cmd}", failure_class="bootstrap_failed")
+2 -1
View File
@@ -386,7 +386,8 @@ def _op_whoami(binary: Path, account: str, *, token_value: str = "") -> Optional
try:
res = subprocess.run(
cmd, env=env, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=10
encoding="utf-8", errors="replace", timeout=10,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
return None
+1
View File
@@ -353,6 +353,7 @@ def _run_capability_probe(
text=True, encoding="utf-8", errors="replace",
timeout=_PROBE_TIMEOUT,
env=env,
check=False,
)
except subprocess.TimeoutExpired:
return None, f"capability probe timed out after {_PROBE_TIMEOUT}s"
+1
View File
@@ -71,6 +71,7 @@ def _git_origin_url(plugin_dir: Path) -> Optional[str]:
capture_output=True,
text=True, encoding="utf-8", errors="replace",
timeout=10,
check=False,
)
if proc.returncode == 0:
url = (proc.stdout or "").strip()
+1
View File
@@ -395,6 +395,7 @@ def default_ls_remote(source: str) -> str:
capture_output=True,
text=True, encoding="utf-8", errors="replace",
timeout=30,
check=False,
)
if proc.returncode != 0:
raise RuntimeError((proc.stderr or "ls-remote failed").strip()[:200])
+2
View File
@@ -514,6 +514,7 @@ def check_alias_collision(name: str) -> Optional[str]:
result = subprocess.run(
["where" if sys.platform == "win32" else "which", canon],
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5,
check=False,
)
if result.returncode == 0:
existing_path = result.stdout.strip().splitlines()[0]
@@ -1544,6 +1545,7 @@ def seed_profile_skills(profile_dir: Path, quiet: bool = False) -> Optional[dict
env={**os.environ, "HERMES_HOME": str(profile_dir)},
cwd=str(project_root),
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=60,
check=False,
)
if result.returncode == 0 and result.stdout.strip():
return json.loads(result.stdout.strip())
+1 -1
View File
@@ -124,7 +124,7 @@ def relaunch(
if sys.platform == "win32":
import subprocess
try:
result = subprocess.run(new_argv)
result = subprocess.run(new_argv, check=False)
sys.exit(result.returncode)
except KeyboardInterrupt:
sys.exit(130)
+1 -1
View File
@@ -447,7 +447,7 @@ def _list_projects(
try:
res = subprocess.run(
[str(binary), "project", "list", "--output", "json"],
env=env, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=15)
env=env, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=15, check=False)
except (OSError, subprocess.TimeoutExpired) as exc:
console.print(f" [red]Couldn't list projects: {exc}[/red]")
return None
+2 -2
View File
@@ -103,7 +103,7 @@ class LostAndFoundError(RuntimeError):
def _parse_sqlite3_cli_version(binary: str) -> Optional[tuple[int, int, int]]:
"""Version of the sqlite3 CLI at *binary* via ``--version``, or None when it cannot run or be parsed."""
try:
probe = subprocess.run([binary, "--version"], capture_output=True, timeout=30)
probe = subprocess.run([binary, "--version"], capture_output=True, timeout=30, check=False)
except (OSError, subprocess.SubprocessError):
return None
if probe.returncode != 0:
@@ -172,7 +172,7 @@ def _cli_supports_recover(binary: str) -> bool:
conn.commit()
finally:
conn.close()
probe = subprocess.run([binary, "-readonly", str(scratch), ".recover"], capture_output=True, timeout=30)
probe = subprocess.run([binary, "-readonly", str(scratch), ".recover"], capture_output=True, timeout=30, check=False)
return probe.returncode == 0 and b"sqlite_dbpage" not in probe.stderr
except (OSError, subprocess.SubprocessError, sqlite3.Error):
return False
+1 -1
View File
@@ -243,7 +243,7 @@ def _setup_backend_ssh(config: dict) -> None:
import subprocess
ssh_cmd = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", *(["-i", ssh_key] if ssh_key else []),
*(["-p", port] if port and port != "22" else []), f"{user}@{host}" if user else host, "echo ok"]
result = subprocess.run(ssh_cmd, timeout=10, **_RUN_KW)
result = subprocess.run(ssh_cmd, timeout=10, **_RUN_KW, check=False)
if result.returncode == 0:
_setup.print_success(" SSH connection successful!")
else:
+1 -1
View File
@@ -65,7 +65,7 @@ def _git_run(args: list[str], *, cwd: Optional[Path] = None, timeout: int = 5, t
try:
return subprocess.run(
[git, *args], capture_output=True, timeout=timeout, cwd=str(cwd) if cwd is not None else None,
**(_GIT_TEXT_KW if text else {}), **kwargs)
**(_GIT_TEXT_KW if text else {}), **kwargs, check=False)
except Exception:
return None
+2
View File
@@ -31,6 +31,7 @@ def source_repository(git_cmd=None, cwd=None) -> str:
[*git_cmd, "config", "--get", "remote.origin.url"], cwd=cwd,
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10,
stdin=subprocess.DEVNULL, env=source_git_env(),
check=False,
)
match = _GITHUB_ORIGIN.fullmatch(result.stdout.strip())
if result.returncode == 0 and match:
@@ -193,6 +194,7 @@ def _refuse_retirement_downgrade(request: dict, terminal: dict, git_cmd, cwd) ->
[*git_cmd, "rev-list", "--ancestry-path", f"{request['commit']}..HEAD"], cwd=cwd,
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10,
stdin=subprocess.DEVNULL, env=source_git_env(),
check=False,
)
# The target need not exist locally before the updater's fetch. When it
# does, any descendants prove that this pinned build would roll us back.
+2 -2
View File
@@ -17,7 +17,7 @@ from hermes_cli.cli_output import (
def _run_text(cmd: list, *, timeout, capture_output: bool = True,
**kwargs) -> subprocess.CompletedProcess:
"""Run a text subprocess with consistent decoding."""
return subprocess.run(cmd, capture_output=capture_output, text=True, encoding="utf-8",
return subprocess.run(cmd, capture_output=capture_output, text=True, encoding="utf-8", # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
errors="replace", timeout=timeout, **kwargs)
@@ -180,7 +180,7 @@ def _cua_driver_autostart_registered_windows(binary: Optional[str] = None) -> bo
try:
result = subprocess.run(
["schtasks.exe", "/Query", "/TN", "cua-driver-serve", "/XML"],
capture_output=True, timeout=10, creationflags=_post_setup_no_window_flags())
capture_output=True, timeout=10, creationflags=_post_setup_no_window_flags(), check=False)
if result.returncode:
return False
# Parse bytes: schtasks' XML declaration carries the output encoding.
+1 -1
View File
@@ -125,7 +125,7 @@ def _run_fresh_recovery_process(
else:
kwargs["start_new_session"] = True
try:
return subprocess.run(command, **kwargs)
return subprocess.run(command, **kwargs, check=False)
except (OSError, subprocess.TimeoutExpired) as exc:
logger.warning("Fresh gateway restart recovery failed: %s", exc)
return None
+1 -1
View File
@@ -425,7 +425,7 @@ def _run_python(argv: list[str], *, stdin: str = "", timeout: float = 60) -> str
try:
done = subprocess.run(argv, input=stdin, capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=timeout, creationflags=windows_hide_flags())
errors="replace", timeout=timeout, creationflags=windows_hide_flags(), check=False)
except (OSError, subprocess.SubprocessError):
return None
return done.stdout if done.returncode == 0 else None
+4 -4
View File
@@ -691,7 +691,7 @@ def _run_pending_fleet_restart() -> bool:
def _systemctl(cmd: list, *, timeout: float):
"""Run a systemctl (or sudo systemctl) invocation, capturing utf-8 text with a timeout."""
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, check=False)
def _systemctl_reset_and_restart(manage_cmd: list, svc_name: str, *, scope_cmd: list | None = None):
@@ -917,7 +917,7 @@ def _restart_macos_launchd_gateways(
_locate_launchd_gateway_service, _wait_for_launchd_service_pid,
)
if require_supervision:
listing = subprocess.run(["launchctl", "list"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
listing = subprocess.run(["launchctl", "list"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10, check=False)
if listing.returncode != 0:
failed_or_stale_units.append("launchd (listing failed)")
return
@@ -1160,10 +1160,10 @@ def _sudo_noninteractive_ok(targeted_probe: list) -> bool:
are about to elevate.
"""
try:
if subprocess.run(["sudo", "-n", "true"], capture_output=True, timeout=5).returncode == 0:
if subprocess.run(["sudo", "-n", "true"], capture_output=True, timeout=5, check=False).returncode == 0:
return True
# Blanket sudo refused — a targeted NOPASSWD sudoers entry may still work.
return subprocess.run(["sudo", "-n", *targeted_probe], capture_output=True, timeout=5).returncode == 0
return subprocess.run(["sudo", "-n", *targeted_probe], capture_output=True, timeout=5, check=False).returncode == 0
except (OSError, subprocess.TimeoutExpired):
return False
+1
View File
@@ -584,6 +584,7 @@ def _ensure_fhs_path_guard() -> None:
capture_output=True,
text=True, encoding="utf-8", errors="replace",
timeout=10,
check=False,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return # no bash or probe hung — don't block update on this
+1 -1
View File
@@ -50,7 +50,7 @@ def _critical_module_import_failures(
try:
result = subprocess.run(
runtime_command(Path(root), code=probe), cwd=str(root), capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=120)
encoding="utf-8", errors="replace", timeout=120, check=False)
except subprocess.TimeoutExpired:
return _probe_failure("TimeoutExpired", "timed out before reporting import health")
except (OSError, subprocess.SubprocessError):
+1 -1
View File
@@ -259,7 +259,7 @@ def run(argv: Sequence[str], *, inherit_lock: bool = False, **kwargs) -> subproc
parent-death signal for the rest); on Windows inside an update it is the same contract over
:func:`popen`, so the child is job-bound before it runs."""
if not (sys.platform == "win32" and _held() is not None):
return subprocess.run(list(argv), **_custody_kwargs(inherit_lock, kwargs))
return subprocess.run(list(argv), **_custody_kwargs(inherit_lock, kwargs)) # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
input, timeout = kwargs.pop("input", None), kwargs.pop("timeout", None)
check = kwargs.pop("check", False)
if input is not None:
+1
View File
@@ -208,6 +208,7 @@ def _stdlib_create_time(pid: int) -> float | None:
["ps", "-o", "lstart=", "-p", str(pid)], capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=5, stdin=subprocess.DEVNULL,
env={"PATH": os.environ.get("PATH") or "/bin:/usr/bin", "LC_ALL": "C", "TZ": "UTC0"},
check=False,
).stdout.strip()
if not out:
return None
+3 -1
View File
@@ -61,7 +61,8 @@ def _derived_version(
def _run_git(repo_dir: Path, *args: str) -> str | None:
try:
result = subprocess.run(
["git", *args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=3, cwd=str(repo_dir)
["git", *args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=3, cwd=str(repo_dir),
check=False,
)
except (OSError, subprocess.SubprocessError):
return None
@@ -230,6 +231,7 @@ def _git_version_info(repo_dir: Path, *, include_untracked: bool = False) -> Ver
text=True, encoding="utf-8", errors="replace",
timeout=3,
cwd=str(repo_dir),
check=False,
)
dirty = dirty_result.returncode == 0 and bool((dirty_result.stdout or "").strip())
except (OSError, subprocess.SubprocessError):
+1
View File
@@ -39,6 +39,7 @@ def _run(argv: list[str], cwd: str, timeout: int, env: dict) -> subprocess.Compl
return subprocess.run(
argv, cwd=cwd, capture_output=True, text=True, encoding='utf-8',
errors='replace', timeout=timeout, stdin=subprocess.DEVNULL, env=env,
check=False,
)
except (OSError, subprocess.SubprocessError):
return None
Loaded 100 of 563 files, more files were not shown because too many files have changed in this diff. Show more