| 1 | """Tests for scripts/lib/subproc.py. |
| 2 | |
| 3 | Covers the process-group cleanup path, timeout behavior, success path, |
| 4 | PID callback wiring, and environment inheritance. |
| 5 | """ |
| 6 | |
| 7 | import builtins |
| 8 | import os as real_os |
| 9 | import platform |
| 10 | import unittest |
| 11 | from unittest.mock import patch |
| 12 | |
| 13 | from lib import subproc |
| 14 | |
| 15 | IS_WINDOWS = platform.system() == "Windows" |
| 16 | |
| 17 | def get_shell_cmd(cmd_str: str) -> list[str]: |
| 18 | if IS_WINDOWS: |
| 19 | if cmd_str == "echo hello": |
| 20 | return ["cmd", "/c", "echo hello"] |
| 21 | elif cmd_str == "exit 3": |
| 22 | return ["cmd", "/c", "exit 3"] |
| 23 | elif cmd_str == "echo err >&2": |
| 24 | return ["cmd", "/c", "echo err 1>&2"] |
| 25 | elif cmd_str in ("sleep 10", "sleep 10 & wait"): |
| 26 | return ["powershell", "-Command", "Start-Sleep 10"] |
| 27 | elif cmd_str == "echo $LAST30DAYS_TEST_VAR": |
| 28 | return ["cmd", "/c", "echo %LAST30DAYS_TEST_VAR%"] |
| 29 | elif cmd_str == "true": |
| 30 | return ["cmd", "/c", "exit 0"] |
| 31 | elif cmd_str == "echo ok": |
| 32 | return ["cmd", "/c", "echo ok"] |
| 33 | else: |
| 34 | raise ValueError(f"No Windows command mapping for: {cmd_str}") |
| 35 | return ["sh", "-c", cmd_str] |
| 36 | |
| 37 | |
| 38 | class TestRunWithTimeout(unittest.TestCase): |
| 39 | def test_success_returns_stdout(self): |
| 40 | result = subproc.run_with_timeout( |
| 41 | get_shell_cmd("echo hello"), |
| 42 | timeout=5, |
| 43 | ) |
| 44 | self.assertEqual(result.returncode, 0) |
| 45 | self.assertEqual(result.stdout.strip(), "hello") |
| 46 | self.assertEqual(result.stderr, "") |
| 47 | |
| 48 | def test_nonzero_exit_returns_returncode_not_exception(self): |
| 49 | result = subproc.run_with_timeout( |
| 50 | get_shell_cmd("exit 3"), |
| 51 | timeout=5, |
| 52 | ) |
| 53 | self.assertEqual(result.returncode, 3) |
| 54 | |
| 55 | def test_captures_stderr(self): |
| 56 | result = subproc.run_with_timeout( |
| 57 | get_shell_cmd("echo err >&2"), |
| 58 | timeout=5, |
| 59 | ) |
| 60 | self.assertEqual(result.stderr.strip(), "err") |
| 61 | |
| 62 | def test_timeout_raises_subproctimeout(self): |
| 63 | with self.assertRaises(subproc.SubprocTimeout): |
| 64 | subproc.run_with_timeout( |
| 65 | get_shell_cmd("sleep 10"), |
| 66 | timeout=1, |
| 67 | ) |
| 68 | |
| 69 | def test_timeout_kills_process_group(self): |
| 70 | """A slow child inside a shell should be killed when the group is signaled.""" |
| 71 | with self.assertRaises(subproc.SubprocTimeout): |
| 72 | # Parent shell spawns a child that sleeps long. |
| 73 | # Without process-group cleanup, the child would orphan. |
| 74 | subproc.run_with_timeout( |
| 75 | get_shell_cmd("sleep 10 & wait"), |
| 76 | timeout=1, |
| 77 | ) |
| 78 | |
| 79 | def test_missing_command_raises_oserror(self): |
| 80 | """Missing executables raise FileNotFoundError (or PermissionError on |
| 81 | some filesystems if a same-named junk file exists).""" |
| 82 | with self.assertRaises(OSError): |
| 83 | subproc.run_with_timeout( |
| 84 | ["/nonexistent-path/last30days-test-no-such-bin"], |
| 85 | timeout=5, |
| 86 | ) |
| 87 | |
| 88 | def test_env_is_passed_through(self): |
| 89 | import os |
| 90 | env = {"LAST30DAYS_TEST_VAR": "custom_value"} |
| 91 | if IS_WINDOWS: |
| 92 | for k in ("SystemRoot", "SystemDrive", "PATH", "COMSPEC", "TEMP", "TMP"): |
| 93 | if k in os.environ: |
| 94 | env[k] = os.environ[k] |
| 95 | else: |
| 96 | env["PATH"] = "/usr/bin:/bin" |
| 97 | result = subproc.run_with_timeout( |
| 98 | get_shell_cmd("echo $LAST30DAYS_TEST_VAR"), |
| 99 | timeout=5, |
| 100 | env=env, |
| 101 | ) |
| 102 | self.assertEqual(result.stdout.strip(), "custom_value") |
| 103 | |
| 104 | def test_on_pid_callback_receives_pid(self): |
| 105 | seen_pids = [] |
| 106 | subproc.run_with_timeout( |
| 107 | get_shell_cmd("true"), |
| 108 | timeout=5, |
| 109 | on_pid=lambda pid: seen_pids.append(pid), |
| 110 | ) |
| 111 | self.assertEqual(len(seen_pids), 1) |
| 112 | self.assertIsInstance(seen_pids[0], int) |
| 113 | self.assertGreater(seen_pids[0], 0) |
| 114 | |
| 115 | def test_timeout_falls_back_to_kill_when_killpg_unavailable(self): |
| 116 | """Simulate Windows (no killpg/getpgid) — should fall back to proc.kill().""" |
| 117 | real_hasattr = builtins.hasattr |
| 118 | |
| 119 | def selective_hasattr(obj, name): |
| 120 | if obj is real_os and name in ("killpg", "getpgid", "setsid"): |
| 121 | return False |
| 122 | return real_hasattr(obj, name) |
| 123 | |
| 124 | with patch.object(builtins, "hasattr", side_effect=selective_hasattr): |
| 125 | with self.assertRaises(subproc.SubprocTimeout): |
| 126 | subproc.run_with_timeout( |
| 127 | ["sh", "-c", "sleep 10"], |
| 128 | timeout=1, |
| 129 | ) |
| 130 | |
| 131 | def test_on_pid_callback_exceptions_are_suppressed(self): |
| 132 | """If the PID callback raises, the subprocess should still run to completion.""" |
| 133 | def raising_callback(pid): |
| 134 | raise RuntimeError("boom") |
| 135 | |
| 136 | # Should not raise, callback exception is swallowed. |
| 137 | result = subproc.run_with_timeout( |
| 138 | get_shell_cmd("echo ok"), |
| 139 | timeout=5, |
| 140 | on_pid=raising_callback, |
| 141 | ) |
| 142 | self.assertEqual(result.returncode, 0) |
| 143 | self.assertEqual(result.stdout.strip(), "ok") |
| 144 | |
| 145 | def test_sigterm_ignoring_child_is_sigkill_escalated(self): |
| 146 | """A child that ignores SIGTERM must be escalated to SIGKILL. |
| 147 | |
| 148 | Without escalation, ``proc.wait(timeout=5)`` raises |
| 149 | ``subprocess.TimeoutExpired`` past the ``except`` block instead of the |
| 150 | documented ``SubprocTimeout``, and the child stays alive. |
| 151 | """ |
| 152 | with self.assertRaises(subproc.SubprocTimeout): |
| 153 | subproc.run_with_timeout( |
| 154 | ["sh", "-c", "trap '' TERM; sleep 30"], |
| 155 | timeout=1, |
| 156 | ) |
| 157 | |
| 158 | def test_escalation_path_guards_killpg_attributeerror(self): |
| 159 | """The SIGKILL escalation must not crash if killpg is unavailable (Windows). |
| 160 | |
| 161 | Regression for the #588 class of bug on the escalation path added in |
| 162 | #433: os.killpg raising AttributeError must be caught and fall back to |
| 163 | proc.kill(), so the documented SubprocTimeout surfaces instead of a bare |
| 164 | AttributeError. The primary SIGTERM path was already guarded (#552); this |
| 165 | mirrors that guard on the escalation path. |
| 166 | """ |
| 167 | TimeoutExpired = subproc.subprocess.TimeoutExpired |
| 168 | |
| 169 | class _FakeProc: |
| 170 | def __init__(self): |
| 171 | self.pid = 4321 |
| 172 | self.kill_count = 0 |
| 173 | |
| 174 | def communicate(self, timeout=None): |
| 175 | raise TimeoutExpired(cmd="x", timeout=timeout) |
| 176 | |
| 177 | def wait(self, timeout=None): |
| 178 | # First wait (timeout=5) forces the SIGKILL escalation branch; |
| 179 | # the final bounded wait is swallowed if the process still |
| 180 | # refuses to exit. |
| 181 | if timeout is not None: |
| 182 | raise TimeoutExpired(cmd="x", timeout=timeout) |
| 183 | return 0 |
| 184 | |
| 185 | def kill(self): |
| 186 | self.kill_count += 1 |
| 187 | |
| 188 | fake = _FakeProc() |
| 189 | with patch.object(subproc.subprocess, "Popen", return_value=fake), \ |
| 190 | patch.object(subproc.os, "getpgid", lambda pid: pid), \ |
| 191 | patch.object(subproc.os, "killpg", side_effect=AttributeError("no killpg on Windows")): |
| 192 | with self.assertRaises(subproc.SubprocTimeout): |
| 193 | subproc.run_with_timeout(["x"], timeout=1) |
| 194 | # Both the primary and escalation paths must have fallen back to kill(). |
| 195 | self.assertGreaterEqual(fake.kill_count, 2) |
| 196 | |
| 197 | |
| 198 | if __name__ == "__main__": |
| 199 | unittest.main() |
| 200 |