MCPcopy Create free account
hub / github.com/agentforce314/clawcodex / _run_in_subprocess

Function _run_in_subprocess

tests/test_init_integration.py:262–292  ·  view source on GitHub ↗

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,
)

Source from the content-addressed store, hash-verified

260
261
262def _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
295class TestSigtermPathRunsCleanups(unittest.TestCase):

Calls 3

getMethod · 0.45
communicateMethod · 0.45
killMethod · 0.45

Tested by

no test coverage detected