Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
42 kB · 863 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864"""Checks for the bounded guest DNS diagnostic.
The child-cleanup checks stop a real process tree through the diagnostic's owndeadline, so the group kill is exercised rather than described. A hosted runnerstarts that interpreter far more slowly than a warm local one, so nothing hereassumes a fixed startup cost:
- the diagnostic budget is large enough to contain interpreter startup, so the stop reason is decided by what the child does, not by how fast a cold interpreter runs,- the child reports its own pids before it sleeps and the test waits for that report inside a bounded window instead of reading the file once,- "is it gone" is answered from ``/proc``: a killed orphan nobody has reaped is still a live pid in state ``Z``. It holds no CPU and issues no requests, so it is not executing, while ``kill(pid, 0)`` alone would call it alive for ever."""
import importlib.utilimport osfrom pathlib import Pathimport signalimport socketimport structimport subprocessimport sysimport tempfileimport timeimport unittestfrom unittest import mock
ROOT = Path(__file__).resolve().parents[2]spec = importlib.util.spec_from_file_location("diagnose_ci_dns", ROOT / "tools/diagnose_ci_dns.py")diagnostic = importlib.util.module_from_spec(spec)spec.loader.exec_module(diagnostic)
# The diagnostic's own budget for a child it must stop. It covers interpreter# startup as well, so it must leave room for a cold start on a busy host.STARTUP_SECONDS = 5# How long the child's own pid report may take to appear, and how long a stopped# pid may take to stop executing. Both are bounded, monotonic windows.REPORT_SECONDS = 2STATE_SECONDS = 10PROC_ROOT = Path("/proc")STATE_GONE = "gone" # no status entry at all: the task has exitedSTATE_UNKNOWN = "unknown" # the status could not be read: never assume it is goneZOMBIE_STATES = frozenset({"Z", "X", "x"})
def process_state(pid, *, root=None): """Return the kernel state letter for ``pid``, or a 'gone'/'unknown' marker.
``stat`` is ``pid (comm) state ...`` and ``comm`` may itself contain spaces and closing parentheses, so the state is read after the last ``)``. """ base = PROC_ROOT if root is None else Path(root) try: text = (base / str(int(pid)) / "stat").read_text(encoding="utf-8", errors="replace") except FileNotFoundError: return STATE_GONE except OSError: return STATE_UNKNOWN try: return text[text.rindex(")") + 2] except (ValueError, IndexError): return STATE_UNKNOWN
def process_is_executing(pid, *, root=None): """True while ``pid`` can still run; False once it exited or is a zombie.
An unreaped orphan in state ``Z`` holds no CPU and touches no request, so it has stopped; a state that cannot be read counts as still executing, because nothing observed says otherwise. Where ``/proc`` is not mounted the signal-0 probe is the only evidence available and is used as the fallback. """ base = PROC_ROOT if root is None else Path(root) if not (base / "self" / "stat").exists(): try: os.kill(int(pid), 0) except ProcessLookupError: return False except PermissionError: return True return True state = process_state(pid, root=base) if state == STATE_GONE: return False return state not in ZOMBIE_STATES
def already_stopped(pid, *, root=None): """True when a fresh read shows ``pid`` has already stopped by itself.
A pin can fail because the task exited and was reaped between the check above it and the pin itself, and such a task owes no signal. The pin failure alone never says that: it is the same ``None`` for a live pid this process may not pin. Only this read decides, and only a missing status entry or an unreaped zombie counts -- a state that cannot be read is not evidence of anything. """ base = PROC_ROOT if root is None else Path(root) if not (base / "self" / "stat").exists(): # Without ``/proc`` a missing entry proves nothing about the pid. return False state = process_state(pid, root=base) return state == STATE_GONE or state in ZOMBIE_STATES
def process_identity(pid, *, root=None): """Return the kernel identity of ``pid``, or None when it has no status entry.
``stat`` is ``pid (comm) state ppid pgrp session ... starttime ...`` and ``comm`` may contain spaces and parentheses, so every field is read after the last ``)``. ``start_ticks`` is the kernel start time in clock ticks since boot: together with ``pgrp`` it names one process for as long as the boot lasts, which a bare pid does not. """ base = PROC_ROOT if root is None else Path(root) try: text = (base / str(int(pid)) / "stat").read_text(encoding="utf-8", errors="replace") except OSError: return None try: fields = text[text.rindex(")") + 2:].split() return {"state": fields[0], "pgrp": int(fields[2]), "start_ticks": fields[19]} except (ValueError, IndexError): return None
def open_identity_fd(pid): """Pin ``pid`` to the process it names right now, or return None.
A pidfd follows its process rather than its pid, so a signal sent through it cannot reach a stranger that inherited the pid afterwards. """ opener = getattr(os, "pidfd_open", None) if opener is None: # pragma: no cover - kernels before 5.3 return None try: return opener(int(pid), 0) except OSError: return None
def signal_identity_fd(fd, signum): """Send ``signum`` through a pidfd; True when the kernel accepted it.""" sender = getattr(signal, "pidfd_send_signal", None) if sender is None: # pragma: no cover - kernels before 5.3 return False try: sender(fd, signum) return True except OSError: return False
def wait_until(predicate, *, timeout, interval=0.02): """Poll ``predicate`` for a bounded window and return its last value.""" deadline = time.monotonic() + float(timeout) while True: if predicate(): return True if time.monotonic() >= deadline: return False time.sleep(interval)
def wait_until_not_executing(pid, *, timeout=STATE_SECONDS, interval=0.05, root=None): """Wait a bounded window for ``pid`` to stop executing; return its last state.""" wait_until(lambda: not process_is_executing(pid, root=root), timeout=timeout, interval=interval) return process_state(pid, root=root)
def reported_tree(pidfile): """Return the process tree the child reported, or None while it is incomplete.
Each line is ``role pid start_ticks pgrp``; the start ticks and group travel with the pid so a later cleanup can tell the reported process from an unrelated one that inherited the pid after it exited. """ try: text = pidfile.read_text(encoding="utf-8", errors="replace") except FileNotFoundError: return None entries = {} for line in text.splitlines(): parts = line.split() if len(parts) != 4: return None role, pid, start_ticks, pgrp = parts if role not in ("child", "grand") or role in entries: return None if not (pid.isdigit() and start_ticks.isdigit() and pgrp.isdigit()): return None if int(pid) < 2: return None entries[role] = {"pid": int(pid), "start_ticks": start_ticks, "pgrp": int(pgrp)} if set(entries) != {"child", "grand"}: return None return {"pids": [entries["child"]["pid"], entries["grand"]["pid"]], "entries": entries}
def reported_pids(pidfile): """Return the pids the child reported, or None while that report is incomplete.""" tree = reported_tree(pidfile) return None if tree is None else tree["pids"]
def child_source(pidfile, *, noisy=False, grand_pid=None): """Source for a child that reports identity for itself and its grandchild, then sleeps.
With ``grand_pid`` the child spawns nothing at all: the test owns that process and this child only records its identity, so ownership never depends on this child reaching the point where it writes the report. Without it the child starts its own grandchild in the same group, which a group signal covers. """ lines = [ "import os, pathlib, subprocess, sys, time\n", f"pidfile = pathlib.Path({str(pidfile)!r})\n", "def ident(pid):\n", " fields = pathlib.Path('/proc/%d/stat' % pid).read_text().rsplit(')', 1)[1].split()\n", " return fields[19], fields[2]\n", ] if grand_pid is None: lines.append("grand = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n") lines.append("grand_pid = grand.pid\n") else: lines.append(f"grand_pid = {int(grand_pid)}\n") lines.extend([ "here, there = ident(os.getpid()), ident(grand_pid)\n", "pidfile.write_text('child %d %s %s\\ngrand %d %s %s\\n'\n", " % (os.getpid(), here[0], here[1], grand_pid, there[0], there[1]))\n", "print('started', flush=True)\n", ]) if noisy: lines.append("print('x' * 20000, flush=True)\n") lines.append("time.sleep(60)\n") return "".join(lines)
def paused_child_source(): """Source for a child that stops itself before it can report anything.""" return "import os, signal, time\nos.kill(os.getpid(), signal.SIGSTOP)\ntime.sleep(60)\n"
class _PipeChild: """A stand-in child whose output is a real pipe, so the drain path is unchanged.
The pid is far outside any real pid range, so even an unpatched group signal could not reach another process; the tests record ``killpg`` themselves. """
def __init__(self, payload=b"", *, keep_open=False): read_fd, write_fd = os.pipe() if payload: os.write(write_fd, payload) if not keep_open: os.close(write_fd) write_fd = None self.write_fd = write_fd self.stdout = os.fdopen(read_fd, "rb", buffering=0) self.pid = 4194304 self.returncode = 0
def poll(self): return self.returncode
def wait(self, timeout=None): return self.returncode
def close(self): if self.write_fd is not None: os.close(self.write_fd) self.write_fd = None
class _UnkillableChild: """A real child presented with a process group production cannot reach.
``pid`` is outside any real pid range, so the group signals the diagnostic issues land on a group that does not exist while the real process tree keeps running. Everything else delegates, so the production code path is unchanged. The regression below needs exactly that: a stop that fails after the child has already reported its pids. """
def __init__(self, proc): self.proc = proc self.pid = 4194304 self.stdout = proc.stdout
@property def returncode(self): return self.proc.returncode
def wait(self, timeout=None): return self.proc.wait(timeout=timeout)
def poll(self): return self.proc.poll()
def stop_owned_tree(spawned, pidfile, *, timeout=STATE_SECONDS): """Stop whatever the diagnostic left behind; return ``(confirmed, detail)``.
The captured handles are the ownership token. A leader is stopped through its own group only while it is still executing, and reaped because it is this process's own child; a group whose leader has been reaped is never signalled, because that group id can already belong to someone else.
A descendant is never stopped by group at all. It is signalled through a pidfd taken now, or as a single pid whose kernel start time and group still match what the child reported. A pid that cannot be identified, a signal that cannot be delivered, and a pid that outlives the bound are all reported as unconfirmed: nothing here quietly waits on a survivor and calls it clean. """ tree = reported_tree(pidfile) entries = {} if tree is None else tree["entries"] unconfirmed = [] if tree is None and not spawned: # No report and no handle: this call observed nothing it could verify, so # it may not claim the tree is gone. return False, "no report and no owned handle: nothing could be verified" for proc in spawned: if process_is_executing(proc.pid): try: os.killpg(proc.pid, signal.SIGKILL) except OSError: pass for proc in spawned: try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: unconfirmed.append(f"the child {proc.pid} could not be reaped") leaders = {proc.pid for proc in spawned} for entry in entries.values(): pid = entry["pid"] if pid in leaders or not process_is_executing(pid): continue # Pin the process first, then read the identity the kernel holds for the # pinned pid: a check that ran before the pin could be stale by the time # any signal was sent, and a plain pid signal could reach a stranger. fd = open_identity_fd(pid) if fd is None: # The pin also fails when the pid is reaped between the check above and # the pin itself, and a task that already stopped owes no signal. Any # other failure -- no pidfd support, a live pid this process may not # pin -- still needed a stop it did not get, so a fresh read showing it # already stopped is the only thing that confirms it. if not already_stopped(pid): unconfirmed.append(f"pid {pid} could not be pinned, so it was not signalled") continue try: current = process_identity(pid) if current is None: # The same exit, now between the pin and this read: a status entry # that is already gone owes no signal either. if not already_stopped(pid): unconfirmed.append(f"pid {pid} identity could not be read after pinning") continue if current["start_ticks"] != entry["start_ticks"] or current["pgrp"] != entry["pgrp"]: unconfirmed.append(f"pid {pid} is no longer the process the child reported") continue if not signal_identity_fd(fd, signal.SIGKILL): unconfirmed.append(f"pid {pid} could not be signalled through its pidfd") finally: os.close(fd) for pid in (tree["pids"] if tree else []): if process_is_executing(pid): wait_until_not_executing(pid, timeout=timeout) if process_is_executing(pid): unconfirmed.append(f"pid {pid} was still executing {timeout:g}s after the cleanup") return not unconfirmed, "; ".join(unconfirmed)
class DiagnosticTests(unittest.TestCase): def test_resolver_overrides_are_fixed_and_child_only(self): original = diagnostic.os.environ.get("GODEBUG") for mode in ("go", "cgo"): result = diagnostic.run_command((sys.executable, "-c", "import os; print(os.environ['GODEBUG'])"), resolver=mode) self.assertEqual(result["output"].strip(), f"netdns={mode}+1") self.assertEqual(diagnostic.os.environ.get("GODEBUG"), original) with self.assertRaises(ValueError): diagnostic.run_command(("missing",), resolver="arbitrary") with self.assertRaises(ValueError): diagnostic.run_command(("missing",), timeout=31)
def test_missing_command_is_explicit(self): with mock.patch.object(diagnostic.shutil, "which", return_value=None): result = diagnostic.run_command(("missing",)) self.assertEqual(result["state"], "unavailable")
def test_command_output_is_bounded(self): result = diagnostic.run_command((sys.executable, "-c", "print('x'*20000)")) self.assertEqual(result["state"], "output_limit") self.assertLessEqual(len(result["output"]), 16384)
def _spy_popen(self, spawned, wrap=None): """Return a patch target that captures the real handle before production sees it.""" real_popen = subprocess.Popen
def spawn(argv, **kwargs): proc = real_popen(argv, **kwargs) spawned.append(proc) return proc if wrap is None else wrap(proc)
return spawn
def test_timeout_and_output_limit_kill_plugin_child(self): for noisy in (False, True): with self.subTest(noisy=noisy), tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" code = child_source(pidfile, noisy=noisy) spawned = [] started = time.monotonic() # The handle is captured at spawn, so the teardown in ``finally`` # owns the tree even when the child never gets as far as reporting # its pids, and even when an assertion fails first. with mock.patch.object(diagnostic.subprocess, "Popen", self._spy_popen(spawned)): try: result = diagnostic.run_command( (sys.executable, "-c", code, str(pidfile)), timeout=STARTUP_SECONDS ) elapsed = time.monotonic() - started # The reason is exact: output over the cap is not a # timeout, and silence at the deadline is not a limit. self.assertEqual( result["state"], "output_limit" if noisy else "timeout", result["output"][-400:] ) self.assertLess( elapsed, STARTUP_SECONDS + REPORT_SECONDS + STATE_SECONDS, "the stop is bounded" ) self.assertIn("started", result["output"]) # The child reports both pids before it sleeps, so an # incomplete report is a startup problem to name, not # something to read past. self.assertTrue( wait_until(lambda: reported_pids(pidfile) is not None, timeout=REPORT_SECONDS), f"the child never reported its pids: state={result['state']!r} after {elapsed:.2f}s", ) child_pid, grandchild_pid = reported_pids(pidfile) self.assertNotEqual(child_pid, grandchild_pid) for pid in (child_pid, grandchild_pid): state = wait_until_not_executing(pid) # An orphan killed with its group stays a zombie until # something adopts and reaps it: it no longer executes # anything, so it has not outlived the stop. self.assertFalse( process_is_executing(pid), f"pid {pid} was still executing (state {state!r}) {STATE_SECONDS:g}s after the stop", ) finally: cleanup = stop_owned_tree(spawned, pidfile) # the teardown must be confirmed, not merely attempted self.assertTrue(cleanup[0], cleanup[1])
def test_the_owned_cleanup_stops_the_tree_when_the_production_stop_fails(self): """A stop that never reaches the group is still fully cleaned up.
The child reports its pids first; then the diagnostic's group signals are redirected at a group that does not exist, so the tree survives the production stop and ``run_command`` gives up with its bounded timeout exception. The assertions below state that the stop failed; the teardown in ``finally`` must still leave nothing executing. """ with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" code = child_source(pidfile) spawned = [] with mock.patch.object( diagnostic.subprocess, "Popen", self._spy_popen(spawned, wrap=_UnkillableChild) ): try: with self.assertRaises(subprocess.TimeoutExpired): diagnostic.run_command((sys.executable, "-c", code, str(pidfile)), timeout=1) # the pids were reported before the stop was attempted self.assertTrue( wait_until(lambda: reported_pids(pidfile) is not None, timeout=REPORT_SECONDS), "the child never reported its pids", ) child_pid, grandchild_pid = reported_pids(pidfile) # and that stop did not reach either of them self.assertTrue(process_is_executing(child_pid), "the production stop reached the group") self.assertTrue(process_is_executing(grandchild_pid), "the production stop reached the group") finally: cleanup = stop_owned_tree(spawned, pidfile) self.assertTrue(cleanup[0], cleanup[1]) for pid in reported_pids(pidfile): self.assertFalse(process_is_executing(pid), f"pid {pid} survived the owned cleanup")
def _start_owned_sleep(self): """Start the process this test owns; its handle exists before anything else.""" return subprocess.Popen( [sys.executable, "-c", "import time; time.sleep(60)"], start_new_session=True )
def test_the_owned_cleanup_stops_an_owned_process_the_group_could_not_reach(self): """A process this test owns outlives the leader's group and is still stopped.
The test starts the detached sleep itself, so its handle exists before the child runs and ownership never depends on a report; the child only records its identity. The production stop reaches the child's group, which this process is not part of, so the cleanup has to stop it from the handle. """ detached = self._start_owned_sleep() spawned = [detached] try: with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" code = child_source(pidfile, grand_pid=detached.pid) with mock.patch.object(diagnostic.subprocess, "Popen", self._spy_popen(spawned)): try: result = diagnostic.run_command( (sys.executable, "-c", code, str(pidfile)), timeout=2 ) self.assertEqual(result["state"], "timeout", result["output"][-400:]) self.assertTrue( wait_until(lambda: reported_pids(pidfile) is not None, timeout=REPORT_SECONDS), "the child never reported its pids", ) child_pid, reported_pid = reported_pids(pidfile) self.assertEqual(reported_pid, detached.pid) self.assertFalse(process_is_executing(child_pid), "the leader should have been stopped") self.assertTrue( process_is_executing(detached.pid), "the owned process should have survived" ) # it leads its own group, so the leader's group signal never reached it self.assertEqual(process_identity(detached.pid)["pgrp"], detached.pid) finally: cleanup = stop_owned_tree(spawned, pidfile) self.assertTrue(cleanup[0], cleanup[1]) self.assertFalse(process_is_executing(detached.pid), "the owned process outlived its cleanup") finally: if detached.poll() is None: detached.kill() detached.wait(timeout=STATE_SECONDS)
def test_the_owned_cleanup_stops_an_owned_process_when_the_child_never_reported(self): """A child that pauses before reporting leaves no report; ownership survives.
The child stops itself before it writes anything, so there is no identity to compare and no pid to read. The process this test owns is still stopped from the handle captured before the child ran, and the cleanup is never confirmed on the strength of a missing report alone. """ detached = self._start_owned_sleep() spawned = [detached] try: with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" with mock.patch.object(diagnostic.subprocess, "Popen", self._spy_popen(spawned)): try: result = diagnostic.run_command( (sys.executable, "-c", paused_child_source(), str(pidfile)), timeout=2 ) self.assertEqual(result["state"], "timeout", result["output"][-400:]) # the child was paused before it could report anything at all self.assertIsNone(reported_pids(pidfile)) self.assertTrue( process_is_executing(detached.pid), "the owned process should have survived" ) finally: cleanup = stop_owned_tree(spawned, pidfile) self.assertTrue(cleanup[0], cleanup[1]) self.assertFalse(process_is_executing(detached.pid), "the owned process outlived its cleanup") finally: if detached.poll() is None: detached.kill() detached.wait(timeout=STATE_SECONDS)
def test_regular_command_result(self): result = diagnostic.run_command((sys.executable, "-c", "print('ok')")) self.assertEqual(result["exit_code"], 0) self.assertEqual(result["output"].strip(), "ok")
def test_loopback_queries_never_use_external_resolver(self): sock = mock.MagicMock() sock.recv.return_value = struct.pack("!HHHHHH", 123, 0x8180, 1, 2, 0, 0) with mock.patch.object(socket, "socket") as factory, mock.patch("secrets.randbelow", return_value=123): factory.return_value.__enter__.return_value = sock result = diagnostic.query_loopback(1) sock.connect.assert_called_once_with(("127.0.0.1", 53)) sock.settimeout.assert_called_once_with(3) self.assertEqual(result["state"], "response") self.assertEqual(result["answers"], 2) self.assertEqual(result["rcode"], 0)
def test_timeout_and_invalid_packet_are_explicit(self): for data in (b"short", struct.pack("!HHHHHH", 999, 0x8180, 1, 0, 0, 0)): with self.subTest(data=data), mock.patch.object(socket, "socket") as factory, mock.patch("secrets.randbelow", return_value=123): factory.return_value.__enter__.return_value.recv.return_value = data self.assertEqual(diagnostic.query_loopback(1)["state"], "invalid_response") with mock.patch.object(socket, "socket", side_effect=TimeoutError): self.assertEqual(diagnostic.query_loopback(1)["state"], "failed")
def test_states_are_read_from_the_field_after_the_last_paren(self): with tempfile.TemporaryDirectory() as directory: root = self._proc_tree(directory, { 41: "41 (python) S 1 41 41", 42: "42 (python) Z 1 42 42", 43: "43 (worker ) Z) R 1 43 43", 44: "44 (dead) X 1 44 44", }) self.assertEqual(process_state(41, root=root), "S") self.assertEqual(process_state(42, root=root), "Z") # a comm containing ')' and a space must not shift the state letter self.assertEqual(process_state(43, root=root), "R") self.assertEqual(process_state(44, root=root), "X")
def test_a_zombie_has_stopped_even_though_its_pid_is_still_there(self): with tempfile.TemporaryDirectory() as directory: root = self._proc_tree(directory, { 41: "41 (python) S 1 41 41", 42: "42 (python) Z 1 42 42", }) self.assertTrue(process_is_executing(41, root=root)) # kill(pid, 0) would still succeed here: the state letter decides self.assertFalse(process_is_executing(42, root=root)) self.assertEqual(process_state(9999, root=root), STATE_GONE) self.assertFalse(process_is_executing(9999, root=root))
def test_an_unreadable_state_is_not_called_gone(self): with tempfile.TemporaryDirectory() as directory: root = self._proc_tree(directory, {41: "not a stat line"}) self.assertEqual(process_state(41, root=root), STATE_UNKNOWN) self.assertTrue( process_is_executing(41, root=root), "only a missing entry or a zombie state means stopped", )
def test_the_state_wait_is_bounded_and_reports_the_last_state(self): with tempfile.TemporaryDirectory() as directory: root = self._proc_tree(directory, {}) states = iter(["S", "S", "Z"]) with mock.patch.object(sys.modules[__name__], "process_state", lambda pid, **_: next(states, "Z")): self.assertEqual(wait_until_not_executing(1, timeout=1, interval=0, root=root), "Z") # a predicate that never holds still returns, on its own bound started = time.monotonic() self.assertFalse(wait_until(lambda: False, timeout=0)) self.assertLess(time.monotonic() - started, 5) self.assertTrue(wait_until(lambda: True, timeout=0))
def test_an_incomplete_pid_report_is_not_trusted(self): with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" self.assertIsNone(reported_pids(pidfile)) pidfile.write_text("", encoding="utf-8") self.assertIsNone(reported_pids(pidfile)) # a report missing either process, or missing its identity, is unused pidfile.write_text("child 4321 91 4321\n", encoding="utf-8") self.assertIsNone(reported_pids(pidfile)) pidfile.write_text("child 4321 91\ngrand 4322 93\n", encoding="utf-8") self.assertIsNone(reported_pids(pidfile)) pidfile.write_text("child 4321 91 4321\ngrand 4322 93 4321\n", encoding="utf-8") self.assertEqual(reported_pids(pidfile), [4321, 4322]) # the identity travels with each pid tree = reported_tree(pidfile) self.assertEqual(tree["entries"]["grand"], {"pid": 4322, "start_ticks": "93", "pgrp": 4321})
def test_a_pid_that_no_longer_matches_the_report_is_never_signalled(self): """A reused pid is refused rather than killed: the identity has to match.
Both entries name this very process but claim a later start time, so a cleanup that signalled on the pid alone would kill the test run. The refusal has to be reported as unconfirmed instead. """ with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" me = os.getpid() mine = process_identity(me) self.assertIsNotNone(mine) later = int(mine["start_ticks"]) + 1 pidfile.write_text( f"child {me} {later} {mine['pgrp']}\ngrand {me} {later} {mine['pgrp']}\n", encoding="utf-8", ) confirmed, detail = stop_owned_tree([], pidfile, timeout=0.2) self.assertFalse(confirmed) self.assertIn("no longer the process the child reported", detail) self.assertTrue(process_is_executing(me), "a pid that cannot be identified is never signalled")
def test_an_identity_swapped_while_pinning_is_not_signalled(self): """The identity is read after the pin, so a swap is refused without a signal.
The pin itself succeeds, so only the identity read can show that the pid no longer names the reported process. Nothing may be signalled then, and the pinned descriptors have to be closed rather than leaked. """ with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" me = os.getpid() mine = process_identity(me) later = int(mine["start_ticks"]) + 1 pidfile.write_text( f"child {me} {later} {mine['pgrp']}\ngrand {me} {later} {mine['pgrp']}\n", encoding="utf-8", ) pinned, signalled = [], []
def pin(pid): read_fd, write_fd = os.pipe() os.close(write_fd) pinned.append(read_fd) return read_fd
before = len(os.listdir("/proc/self/fd")) with mock.patch.object(sys.modules[__name__], "open_identity_fd", pin), \ mock.patch.object(sys.modules[__name__], "signal_identity_fd", lambda fd, sig: signalled.append((fd, sig)) or True): confirmed, detail = stop_owned_tree([], pidfile, timeout=0.2) self.assertFalse(confirmed) self.assertIn("no longer the process the child reported", detail) self.assertEqual(len(pinned), 2, "each reported pid is pinned once") self.assertEqual(signalled, [], "a pid whose identity changed is never signalled") self.assertEqual(len(os.listdir("/proc/self/fd")), before, "the pinned fds were closed, not leaked") self.assertTrue(process_is_executing(me))
def test_a_pid_that_cannot_be_pinned_is_never_signalled_numerically(self): """Without a pidfd nothing proves which process the pid names, so nothing is sent.
The report matches this very process, so a numeric fallback signal would end the test run. The refusal has to be reported instead. """ with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" me = os.getpid() mine = process_identity(me) pidfile.write_text( f"child {me} {mine['start_ticks']} {mine['pgrp']}\n" f"grand {me} {mine['start_ticks']} {mine['pgrp']}\n", encoding="utf-8", ) kills = [] with mock.patch.object(sys.modules[__name__], "open_identity_fd", lambda pid: None), \ mock.patch.object(os, "kill", lambda pid, sig: kills.append(pid)): confirmed, detail = stop_owned_tree([], pidfile, timeout=0.2) self.assertFalse(confirmed) self.assertIn("could not be pinned", detail) self.assertEqual(kills, [], "there is no numeric fallback signal") self.assertTrue(process_is_executing(me))
def _reaping_window(self, pidfile, first_checks, *, state): """Report this very process, checked for a stop first and read as ``state``.
The report names this process because nothing is ever signalled here: a signal addressed to these pids would end the test run. ``first_checks`` True values put the pin inside the window where the pid still looks executing, which is the window the cleanup has to survive. """ me = os.getpid() mine = process_identity(me) pidfile.write_text( f"child {me} {mine['start_ticks']} {mine['pgrp']}\n" f"grand {me} {mine['start_ticks']} {mine['pgrp']}\n", encoding="utf-8", ) checks = [True] * first_checks # after the scripted checks the pid keeps answering exactly as ``state`` # says, so the cleanup's own bounded confirming wait reads the same process live = state != STATE_GONE and state not in ZOMBIE_STATES signalled = []
def still_executing(pid, *, root=None): # every pid still looks like it needs a stop until the checks run out, # which is when the cleanup makes its own confirming read return checks.pop(0) if checks else live
patches = ( mock.patch.object(sys.modules[__name__], "process_is_executing", still_executing), mock.patch.object(sys.modules[__name__], "open_identity_fd", lambda pid: None), mock.patch.object(sys.modules[__name__], "process_state", lambda pid, **kwargs: state), mock.patch.object( sys.modules[__name__], "signal_identity_fd", lambda fd, sig: signalled.append((fd, sig)) or True, ), ) return checks, signalled, patches
def test_a_pid_reaped_before_its_pin_is_confirmed_stopped(self): """A descendant that exits between the check and the pin needs no signal.
This is the observed cleanup failure: the group stop had already reached the descendant, and it was reaped in the window between "still executing" and the pin. That pin fails exactly like an unpinnable live process, so only the fresh state read may decide -- and a pid with no status entry has stopped, which confirms the cleanup instead of reporting a survivor. """ with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" checks, signalled, patches = self._reaping_window(pidfile, 2, state=STATE_GONE) with patches[0], patches[1], patches[2], patches[3]: confirmed, detail = stop_owned_tree([], pidfile, timeout=0.2) self.assertEqual(checks, [], "each reported pid was checked for a stop before the read") self.assertTrue(confirmed, detail) self.assertEqual(signalled, [], "a pid that already stopped is never signalled") self.assertTrue(process_is_executing(os.getpid()))
def test_a_live_pid_that_cannot_be_pinned_stays_unconfirmed(self): """The same pin failure stays unconfirmed while the pid can still execute.
Only the read after the pin separates the two cases. Here it reports a live process, so the exemption that covers an already reaped pid cannot be claimed for a survivor that simply could not be pinned. """ with tempfile.TemporaryDirectory() as directory: pidfile = Path(directory) / "child.pid" checks, signalled, patches = self._reaping_window(pidfile, 2, state="S") with patches[0], patches[1], patches[2], patches[3]: confirmed, detail = stop_owned_tree([], pidfile, timeout=0.2) self.assertEqual(checks, [], "each reported pid was checked for a stop before the read") self.assertFalse(confirmed) self.assertIn("could not be pinned", detail) self.assertIn("was still executing", detail) self.assertEqual(signalled, [], "there is no numeric fallback signal") self.assertTrue(process_is_executing(os.getpid()))
def test_no_report_and_no_handle_cannot_be_called_clean(self): """Nothing observed means nothing confirmed, even though nothing was left.""" with tempfile.TemporaryDirectory() as directory: confirmed, detail = stop_owned_tree([], Path(directory) / "missing.pid", timeout=0.2) self.assertFalse(confirmed) self.assertIn("nothing could be verified", detail)
def test_each_stop_reason_signals_the_stopped_group(self): """The state decides the stop reason; both reasons signal that group.""" for expected, child in ( ("output_limit", _PipeChild(b"x" * 20000)), ("timeout", _PipeChild(keep_open=True)), ): with self.subTest(state=expected), \ mock.patch.object(diagnostic.shutil, "which", return_value="/usr/bin/python3"), \ mock.patch.object(diagnostic.subprocess, "Popen", lambda *args, **kwargs: child), \ mock.patch.object(diagnostic.os, "killpg") as killpg: try: result = diagnostic.run_command(("python3", "-c", "child"), timeout=0.2) finally: child.close() self.assertEqual(result["state"], expected) self.assertEqual(result["exit_code"], 0) self.assertGreaterEqual(killpg.call_count, 1, "the stopped group is signalled") # every signal addressed the child's own group, never the caller's self.assertEqual({call.args[0] for call in killpg.call_args_list}, {child.pid}) self.assertLessEqual(len(result["output"]), 16384)
def _proc_tree(self, directory, entries): """Build a stand-in /proc tree so the state probes read real files.""" root = Path(directory) (root / "self").mkdir() (root / "self" / "stat").write_text("1 (python) S 0 1\n", encoding="utf-8") for pid, text in entries.items(): (root / str(pid)).mkdir() (root / str(pid) / "stat").write_text(text + "\n", encoding="utf-8") return root
def test_commands_do_not_modify_or_read_environment(self): for _, argv in diagnostic.COMMANDS: self.assertNotIn(argv[0], ("docker", "sudo", "su", "env", "printenv")) self.assertNotIn("restart", argv) self.assertNotIn("--property=Environment", argv) self.assertEqual(diagnostic.ENV_KEYS, ("PATH", "HOME", "LANG", "LC_ALL", "TERM"))
if __name__ == "__main__": unittest.main()