forked from goodboy/tractor
1
0
Fork 0

Move `mk_cmd()` to `._testing`

Since we're going to need it more generally for `.devx` sub-sys tooling
tests.

Also, up the sync-pause ctl-c delay another 10ms..
aio_abandons
Tyler Goodlet 2024-07-10 15:40:44 -04:00
parent fc95c6719f
commit 131e3e8157
2 changed files with 38 additions and 17 deletions

View File

@ -13,7 +13,6 @@ TODO:
from functools import partial from functools import partial
import itertools import itertools
import platform import platform
import pathlib
import time import time
import pytest import pytest
@ -29,7 +28,7 @@ from tractor.devx._debug import (
_repl_fail_msg, _repl_fail_msg,
) )
from tractor._testing import ( from tractor._testing import (
examples_dir, mk_cmd,
) )
from conftest import ( from conftest import (
_ci_env, _ci_env,
@ -52,15 +51,6 @@ if platform.system() == 'Windows':
) )
def mk_cmd(ex_name: str) -> str:
'''
Generate a command suitable to pass to ``pexpect.spawn()``.
'''
script_path: pathlib.Path = examples_dir() / 'debugging' / f'{ex_name}.py'
return ' '.join(['python', str(script_path)])
# TODO: was trying to this xfail style but some weird bug i see in CI # TODO: was trying to this xfail style but some weird bug i see in CI
# that's happening at collect time.. pretty soon gonna dump actions i'm # that's happening at collect time.. pretty soon gonna dump actions i'm
# thinkin... # thinkin...
@ -84,19 +74,31 @@ def spawn(
start_method, start_method,
testdir, testdir,
reg_addr, reg_addr,
) -> 'pexpect.spawn':
) -> 'pexpect.spawn':
'''
Use the `pexpect` module shipped via `testdir.spawn()` to
run an `./examples/..` script by name.
'''
if start_method != 'trio': if start_method != 'trio':
pytest.skip( pytest.skip(
"Debugger tests are only supported on the trio backend" '`pexpect` based tests only supported on `trio` backend'
) )
def _spawn(cmd): def _spawn(
cmd: str,
**mkcmd_kwargs,
):
return testdir.spawn( return testdir.spawn(
cmd=mk_cmd(cmd), cmd=mk_cmd(
cmd,
**mkcmd_kwargs,
),
expect_timeout=3, expect_timeout=3,
) )
# such that test-dep can pass input script name.
return _spawn return _spawn
@ -1121,7 +1123,7 @@ def test_pause_from_sync(
# `subactor` suffers a race where the root/parent # `subactor` suffers a race where the root/parent
# sends an actor-cancel prior to it hitting its pause # sends an actor-cancel prior to it hitting its pause
# point; by def the value is 0.1 # point; by def the value is 0.1
delay=0.3, delay=0.4,
) )
# XXX, fwiw without a brief sleep here the SIGINT might actually # XXX, fwiw without a brief sleep here the SIGINT might actually
@ -1190,7 +1192,7 @@ def test_pause_from_sync(
child, child,
patt=attach_key, patt=attach_key,
# NOTE same as comment above # NOTE same as comment above
delay=0.3, delay=0.4,
) )
child.sendline('c') child.sendline('c')

View File

@ -54,6 +54,25 @@ def examples_dir() -> pathlib.Path:
return repodir() / 'examples' return repodir() / 'examples'
def mk_cmd(
ex_name: str,
exs_subpath: str = 'debugging',
) -> str:
'''
Generate a shell command suitable to pass to ``pexpect.spawn()``.
'''
script_path: pathlib.Path = (
examples_dir()
/ exs_subpath
/ f'{ex_name}.py'
)
return ' '.join([
'python',
str(script_path)
])
@acm @acm
async def expect_ctxc( async def expect_ctxc(
yay: bool, yay: bool,