Add `SIGINT` cleanup to `spawn` fixture in `devx/conftest`
Convert `spawn` fixture to a generator and add post-test graceful subproc cleanup via `SIGINT`/`SIGKILL` to avoid leaving stale `pexpect` child procs around between test runs as well as any UDS-tpt socket files under the system runtime-dir. Deats, - convert `return _spawn` -> `yield _spawn` to enable post-yield teardown logic. - add a new `nonlocal spawned` ref so teardown logic can access the last spawned child from outside the delivered spawner fn-closure. - add `SIGINT`-loop after yield with 5s timeout, then `SIGKILL` if proc still alive. - add masked `breakpoint()` and TODO about UDS path cleanup (this commit msg was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-codetpt_tolerance
parent
f8e25688c7
commit
7145fa364f
|
|
@ -4,6 +4,7 @@
|
||||||
'''
|
'''
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
import time
|
import time
|
||||||
|
import signal
|
||||||
from typing import (
|
from typing import (
|
||||||
Callable,
|
Callable,
|
||||||
TYPE_CHECKING,
|
TYPE_CHECKING,
|
||||||
|
|
@ -69,12 +70,15 @@ def spawn(
|
||||||
import os
|
import os
|
||||||
os.environ['PYTHON_COLORS'] = '0'
|
os.environ['PYTHON_COLORS'] = '0'
|
||||||
|
|
||||||
|
spawned: PexpectSpawner|None = None
|
||||||
|
|
||||||
def _spawn(
|
def _spawn(
|
||||||
cmd: str,
|
cmd: str,
|
||||||
**mkcmd_kwargs,
|
**mkcmd_kwargs,
|
||||||
) -> pty_spawn.spawn:
|
) -> pty_spawn.spawn:
|
||||||
|
nonlocal spawned
|
||||||
unset_colors()
|
unset_colors()
|
||||||
return testdir.spawn(
|
spawned = testdir.spawn(
|
||||||
cmd=mk_cmd(
|
cmd=mk_cmd(
|
||||||
cmd,
|
cmd,
|
||||||
**mkcmd_kwargs,
|
**mkcmd_kwargs,
|
||||||
|
|
@ -84,9 +88,32 @@ def spawn(
|
||||||
# ^TODO? get `pytest` core to expose underlying
|
# ^TODO? get `pytest` core to expose underlying
|
||||||
# `pexpect.spawn()` stuff?
|
# `pexpect.spawn()` stuff?
|
||||||
)
|
)
|
||||||
|
return spawned
|
||||||
|
|
||||||
# such that test-dep can pass input script name.
|
# such that test-dep can pass input script name.
|
||||||
return _spawn # the `PexpectSpawner`, type alias.
|
yield _spawn # the `PexpectSpawner`, type alias.
|
||||||
|
|
||||||
|
if (
|
||||||
|
spawned
|
||||||
|
and
|
||||||
|
(ptyproc := spawned.ptyproc)
|
||||||
|
):
|
||||||
|
start: float = time.time()
|
||||||
|
timeout: float = 5
|
||||||
|
while (
|
||||||
|
spawned
|
||||||
|
and
|
||||||
|
spawned.isalive()
|
||||||
|
and
|
||||||
|
(_time_took := (time.time() - start) < timeout)
|
||||||
|
):
|
||||||
|
ptyproc.kill(signal.SIGINT)
|
||||||
|
time.sleep(0.01)
|
||||||
|
else:
|
||||||
|
ptyproc.kill(signal.SIGKILL)
|
||||||
|
|
||||||
|
# TODO? ensure we've cleaned up any UDS-paths?
|
||||||
|
# breakpoint()
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(
|
@pytest.fixture(
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue