Run ``code`` in a fresh Python subprocess, optionally sending it ``signal_name`` after ``signal_after_ms`` ms. Returns (returncode, stdout, stderr).
(
code: str,
signal_after_ms: float | None = None,
signal_name: str = "SIGTERM",
timeout: float = 5.0,
)
| 260 | |
| 261 | |
| 262 | def _run_in_subprocess( |
| 263 | code: str, |
| 264 | signal_after_ms: float | None = None, |
| 265 | signal_name: str = "SIGTERM", |
| 266 | timeout: float = 5.0, |
| 267 | ) -> tuple[int, str, str]: |
| 268 | """Run ``code`` in a fresh Python subprocess, optionally sending it |
| 269 | ``signal_name`` after ``signal_after_ms`` ms. Returns |
| 270 | (returncode, stdout, stderr). |
| 271 | """ |
| 272 | import signal as _signal |
| 273 | env = dict(os.environ) |
| 274 | env["PYTHONPATH"] = str(WORKTREE_ROOT) + os.pathsep + env.get("PYTHONPATH", "") |
| 275 | proc = subprocess.Popen( |
| 276 | [sys.executable, "-c", code], |
| 277 | stdout=subprocess.PIPE, |
| 278 | stderr=subprocess.PIPE, |
| 279 | env=env, |
| 280 | ) |
| 281 | if signal_after_ms is not None: |
| 282 | time.sleep(signal_after_ms / 1000.0) |
| 283 | try: |
| 284 | proc.send_signal(getattr(_signal, signal_name)) |
| 285 | except ProcessLookupError: |
| 286 | pass # already exited |
| 287 | try: |
| 288 | out, err = proc.communicate(timeout=timeout) |
| 289 | except subprocess.TimeoutExpired: |
| 290 | proc.kill() |
| 291 | out, err = proc.communicate() |
| 292 | return proc.returncode, out.decode("utf-8", errors="replace"), err.decode("utf-8", errors="replace") |
| 293 | |
| 294 | |
| 295 | class TestSigtermPathRunsCleanups(unittest.TestCase): |
no test coverage detected