Prevent `test_breakpoint_hook_restored` subproc hangs

If the underlying example script fails (say due to a console output
pattern-mismatch, `AssertionError`) the `pexpect` managed subproc with
a `debug_mode=True` crash-handling-REPL engaged will ofc *not terminate*
due to any SIGINT sent by the test harnesss (since we shield from it as
part of normal sub-actor debugger operation). So instead always send
a 'continue' cmd to the active `PdbREPL`'s stdin so it deactivates and
allows the py-script-process to raise and terminate, unblocking the
`pexpect.spawn`'s internal subproc joiner (which would otherwise hang
without manual intervention, blocking downstream tests..).

Also, use the new `PexpectSpawner` type alias after actually importing
future annots.. XD
Tyler Goodlet 2025-06-11 19:32:56 -04:00
parent 6fd08c6e32
commit 37377e8220
2 changed files with 26 additions and 12 deletions

View File

@ -2,6 +2,7 @@
`tractor.devx.*` tooling sub-pkg test space.
'''
from __future__ import annotations
import time
from typing import (
Callable,

View File

@ -13,9 +13,13 @@ TODO:
when debugging a problem inside the stack vs. in their app.
'''
from __future__ import annotations
import os
import signal
import time
from typing import (
TYPE_CHECKING,
)
from .conftest import (
expect,
@ -29,9 +33,12 @@ from pexpect.exceptions import (
EOF,
)
if TYPE_CHECKING:
from ..conftest import PexpectSpawner
def test_shield_pause(
spawn,
spawn: PexpectSpawner,
):
'''
Verify the `tractor.pause()/.post_mortem()` API works inside an
@ -126,7 +133,7 @@ def test_shield_pause(
def test_breakpoint_hook_restored(
spawn,
spawn: PexpectSpawner,
):
'''
Ensures our actor runtime sets a custom `breakpoint()` hook
@ -140,16 +147,22 @@ def test_breakpoint_hook_restored(
child = spawn('restore_builtin_breakpoint')
child.expect(PROMPT)
assert_before(
child,
[
_pause_msg,
"<Task '__main__.main'",
"('root'",
"first bp, tractor hook set",
]
)
child.sendline('c')
try:
assert_before(
child,
[
_pause_msg,
"<Task '__main__.main'",
"('root'",
"first bp, tractor hook set",
]
)
# XXX if the above raises `AssertionError`, without sending
# the final 'continue' cmd to the REPL-active sub-process,
# we'll hang waiting for that pexpect instance to terminate..
finally:
child.sendline('c')
child.expect(PROMPT)
assert_before(
child,