test+harden(cli): cover parent-chain walk in concurrent-instance detection

Follow-up to @Strontvod's fix.

Tests:
- Five new tests in test_update_concurrent_quarantine.py cover the parent-
  chain exclusion: the .exe launcher is excluded, an unrelated sibling
  hermes.exe is still reported, multi-level ancestry is fully excluded,
  PID cycles in the parent chain don't hang, and a partially-stubbed
  psutil (no Process attribute) degrades gracefully instead of crashing.
- New _fake_psutil_with_parent_chain helper builds a fuller stand-in
  (Process / NoSuchProcess / AccessDenied + process_iter) than the
  process_iter-only SimpleNamespace the older tests use.

Hardening:
- Broaden the except in the parent-walk to bare Exception. The original
  fix listed (NoSuchProcess, AccessDenied, ValueError), but those names
  are evaluated lazily during exception matching — if psutil is a partial
  stub without the attribute, the exception handler itself raises
  AttributeError that escapes. The function is documented as 'never raises'
  (the surrounding update flow depends on it), so the broader catch keeps
  the contract regardless of how the dependency is shaped.

AUTHOR_MAP:
- Map schepers.zander1@gmail.com -> Strontvod so the salvaged commit
  resolves to @Strontvod in the release notes.

All 18 detect_concurrent + quarantine tests pass.
This commit is contained in:
Teknium 2026-05-24 19:41:23 -07:00
parent 323cce7e94
commit 46f8948bad
3 changed files with 185 additions and 2 deletions

View file

@ -7693,12 +7693,18 @@ def _detect_concurrent_hermes_instances(
exclude_pids: set[int] = {exclude_pid}
else:
exclude_pids = {os.getpid()}
# The parent-walk is best-effort: if psutil rejects a PID (NoSuchProcess /
# AccessDenied) we stop walking and use whatever we've collected so far.
# Broader Exception catch on the outer block guards against partially-
# stubbed psutil in unit tests (e.g. a SimpleNamespace lacking Process /
# NoSuchProcess) — the surrounding update flow documents this helper as
# "never raises".
try:
current = psutil.Process(next(iter(exclude_pids)))
while True:
try:
parent = current.parent()
except (psutil.NoSuchProcess, psutil.AccessDenied):
except Exception:
break
if parent is None or parent.pid <= 0:
break
@ -7706,7 +7712,7 @@ def _detect_concurrent_hermes_instances(
break # loop detected
exclude_pids.add(parent.pid)
current = parent
except (psutil.NoSuchProcess, psutil.AccessDenied, ValueError):
except Exception:
pass
# Resolve every shim path to its canonical form once for cheap comparison.

View file

@ -54,6 +54,7 @@ AUTHOR_MAP = {
"erhanyasarx@gmail.com": "erhnysr",
"30366221+WorldWriter@users.noreply.github.com": "WorldWriter",
"dafeng@DafengdeMacBook-Pro.local": "WorldWriter",
"schepers.zander1@gmail.com": "Strontvod",
"anadi.jaggia@gmail.com": "Jaggia",
"32201324+simpolism@users.noreply.github.com": "simpolism",
"simpolism@gmail.com": "simpolism",

View file

@ -118,6 +118,182 @@ def test_detect_concurrent_is_noop_off_windows(_winp, tmp_path):
assert cli_main._detect_concurrent_hermes_instances(tmp_path) == []
# ---------------------------------------------------------------------------
# Parent-chain exclusion (issue #30768 follow-up — the setuptools .exe
# launcher on Windows is a separate native process that spawns python.exe;
# excluding only ``os.getpid()`` flags the launcher as a concurrent instance.
# ---------------------------------------------------------------------------
def _fake_psutil_with_parent_chain(
parent_chain: list[int],
proc_iter_rows: list,
):
"""Build a psutil stand-in that has Process()/parent() AND process_iter().
``parent_chain`` is the list of PIDs returned by successive ``.parent()``
calls starting from the seed (``os.getpid()``); the last entry's
``.parent()`` returns ``None`` to terminate the walk.
"""
class _FakeProc:
def __init__(self, pid: int, chain: list[int]):
self.pid = pid
self._chain = chain
def parent(self):
if not self._chain:
return None
next_pid = self._chain[0]
return _FakeProc(next_pid, self._chain[1:])
class _NoSuchProcess(Exception):
pass
class _AccessDenied(Exception):
pass
def _process(pid):
return _FakeProc(pid, list(parent_chain))
return types.SimpleNamespace(
Process=_process,
NoSuchProcess=_NoSuchProcess,
AccessDenied=_AccessDenied,
process_iter=lambda attrs: iter(proc_iter_rows),
)
@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_excludes_parent_chain(_winp, tmp_path):
"""The .exe launcher (parent of os.getpid()) must NOT be flagged.
Simulates the real Windows topology: hermes.exe launcher (PID L) spawns
python.exe (PID os.getpid()). Both run from the same shim path. With the
old single-PID exclusion, L would be reported as a concurrent instance.
"""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")
me = os.getpid()
launcher_pid = me + 100 # the .exe launcher — our parent
rows = [
_make_proc(me, str(shim), "python.exe"),
_make_proc(launcher_pid, str(shim), "hermes.exe"),
]
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[launcher_pid],
proc_iter_rows=rows,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
# Both self AND the launcher are excluded; no false positive.
assert result == []
@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_still_finds_unrelated_other_hermes(_winp, tmp_path):
"""A sibling hermes.exe outside our ancestor chain must still be reported."""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")
me = os.getpid()
launcher_pid = me + 100 # our .exe launcher (parent — must be excluded)
sibling_pid = me + 200 # an UNRELATED hermes.exe (must still be reported)
rows = [
_make_proc(me, str(shim), "python.exe"),
_make_proc(launcher_pid, str(shim), "hermes.exe"),
_make_proc(sibling_pid, str(shim), "hermes.exe"),
]
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[launcher_pid],
proc_iter_rows=rows,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
assert result == [(sibling_pid, "hermes.exe")]
@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_parent_chain_walks_deep(_winp, tmp_path):
"""Multi-level ancestry (shell → launcher → python) is fully excluded."""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")
me = os.getpid()
parent_pid = me + 1
grandparent_pid = me + 2
greatgrandparent_pid = me + 3
rows = [
_make_proc(me, str(shim), "python.exe"),
_make_proc(parent_pid, str(shim), "hermes.exe"),
_make_proc(grandparent_pid, str(shim), "hermes.exe"),
_make_proc(greatgrandparent_pid, str(shim), "hermes.exe"),
]
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[parent_pid, grandparent_pid, greatgrandparent_pid],
proc_iter_rows=rows,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
assert result == []
@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_parent_walk_handles_cycle(_winp, tmp_path):
"""A PID cycle in the parent chain must not hang the walk."""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")
me = os.getpid()
bogus_loop_pid = me + 1
rows = [_make_proc(me, str(shim), "python.exe")]
# Chain that points back to ``me`` — the loop-detection branch must break.
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[bogus_loop_pid, me, bogus_loop_pid],
proc_iter_rows=rows,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
# No crash, no hang; self + bogus_loop_pid excluded; no others reported.
assert result == []
@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_parent_walk_handles_stub_without_process(_winp, tmp_path):
"""Partially-stubbed psutil (no Process attr) must NOT crash the helper.
The function documents itself as "never raises"; a unit-test stub that
only models ``process_iter`` must still complete cleanly with a sensible
result rather than escape ``AttributeError`` to the caller.
"""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")
me = os.getpid()
other_pid = me + 1
rows = [
_make_proc(me, str(shim), "hermes.exe"),
_make_proc(other_pid, str(shim), "hermes.exe"),
]
# SimpleNamespace with ONLY process_iter — no Process / NoSuchProcess.
fake_psutil = types.SimpleNamespace(process_iter=lambda attrs: iter(rows))
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
# Parent-walk silently failed; self still excluded; other still reported.
assert result == [(other_pid, "hermes.exe")]
# ---------------------------------------------------------------------------
# _format_concurrent_instances_message
# ---------------------------------------------------------------------------