Add fork-aware capture fixtures to `_testing.pytest`

Extend the pytest plugin with helpers that detect
and adapt to `--capture=sys` under fork-based
spawners (`main_thread_forkserver`, `mp_forkserver`)
where fd-capture causes hangs.

Deats,
- track `_cap_sys_passed_as_flag` + `_cap_fd_set`
  globals in `pytest_load_initial_conftests()`.
- add `@pytest.hookimpl(tryfirst=True)` + re-parse
  args after appending `--capture=sys`.
- `_is_forking_spawner()` predicate + fixture.
- `maybe_xfail_for_spawner()` — enalbes skipping tests that need capsys
  but weren't passed `--capture=sys`.
- `set_fork_aware_capture` fixture — returns the appropriate capture
  fixture per spawner backend based on `start_method: str` set via CLI.
- wire `set_fork_aware_capture` into `tractor_test`
  wrapper's fixture injection.

Also,
- add `alert_on_finish` session fixture (terminal
  bell on completion; tho not sure it works fully..)
- add `ids=` to `start_method` parametrize.
- restore `default=False` on `--enable-stackscope`.
- drop commented-out `--ll` option block; we will likely factor it to
  our plugin eventually however..

(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
subint_forkserver_backend
Gud Boi 2026-05-02 01:09:02 -04:00
parent 5a9926fc32
commit d549c72052
1 changed files with 148 additions and 13 deletions

View File

@ -52,16 +52,30 @@ pytest_plugins: tuple[str, ...] = (
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import Namespace from argparse import Namespace
_cap_sys_passed_as_flag: bool = False
_cap_fd_set: bool = False
# XXX REQUIRED in order to enforce `--capture=` flag # XXX REQUIRED in order to enforce `--capture=` flag
# pre test session. # pre test session.
# https://docs.pytest.org/en/stable/reference/reference.html#bootstrapping-hooks # https://docs.pytest.org/en/stable/reference/reference.html#bootstrapping-hooks
@pytest.hookimpl(tryfirst=True)
def pytest_load_initial_conftests( def pytest_load_initial_conftests(
early_config: pytest.Config, early_config: pytest.Config,
parser: pytest.Parser, parser: pytest.Parser,
args: list[str], args: list[str],
): ):
global _cap_sys_passed_as_flag, _cap_fd_set
opts: Namespace = early_config.option opts: Namespace = early_config.option
if opts.capture == 'fd':
_cap_fd_set = True
opts_w_args: Namespace = parser.parse_known_args(args) opts_w_args: Namespace = parser.parse_known_args(args)
if opts_w_args.capture == 'fd':
_cap_fd_set = True
if '--capture=sys' in args:
_cap_sys_passed_as_flag = True
# XXX, ALWAYS apply capsys for fork based spawners: # XXX, ALWAYS apply capsys for fork based spawners:
# * main_thread_forkserver # * main_thread_forkserver
@ -94,14 +108,23 @@ def pytest_load_initial_conftests(
(spawner := opts_w_args.spawn_backend) in [ (spawner := opts_w_args.spawn_backend) in [
'main_thread_forkserver', 'main_thread_forkserver',
] ]
and
opts.capture == 'fd'
): ):
print( print(
f'XXX SETTING CAPSYS due to spawning backend XXX\n' f'XXX SETTING CAPSYS due to spawning backend XXX\n'
f'--spawn-backend={spawner!r}\n' f'--spawn-backend={spawner!r}\n'
) )
opts.capture = 'sys' opts.capture = 'sys'
# ^TODO XXX?/
# seems like this doesn't get set by the above!?
args.append(
'--capture=sys',
)
out = parser.parse_known_and_unknown_args(
args,
early_config.option,
)
assert out[0].capture == 'sys'
# breakpoint()
# TODO, set various `$TRACTOR_X*` osenv vars here! # TODO, set various `$TRACTOR_X*` osenv vars here!
print( print(
@ -187,11 +210,17 @@ def tractor_test(
# injection (via `__wrapped__`) without leaking the async # injection (via `__wrapped__`) without leaking the async
# nature. # nature.
@wraps(wrapped) @wraps(wrapped)
def wrapper(**kwargs): def wrapper(
set_fork_aware_capture: pytest.CaptureFixture|None = None,
# ^NOTE when set, the decorated fn declared as fixture-param.
**kwargs,
):
__tracebackhide__: bool = hide_tb __tracebackhide__: bool = hide_tb
# NOTE, ensure we inject any test-fn declared fixture # NOTE, ensure we inject any test-fn declared fixture
# names. # names.
sig = inspect.signature(wrapped)
for kw in [ for kw in [
'reg_addr', 'reg_addr',
'loglevel', 'loglevel',
@ -200,9 +229,13 @@ def tractor_test(
'tpt_proto', 'tpt_proto',
'timeout', 'timeout',
]: ]:
if kw in inspect.signature(wrapped).parameters: if kw in sig.parameters:
assert kw in kwargs assert kw in kwargs
if 'set_fork_aware_capture' in sig.parameters:
assert set_fork_aware_capture
kwargs['set_fork_aware_capture'] = set_fork_aware_capture
# Extract runtime settings as locals for # Extract runtime settings as locals for
# `open_root_actor()`; these must NOT leak into # `open_root_actor()`; these must NOT leak into
# `kwargs` when the test fn doesn't declare them # `kwargs` when the test fn doesn't declare them
@ -245,7 +278,6 @@ def tractor_test(
# invoke test-fn body IN THIS task # invoke test-fn body IN THIS task
await wrapped(**kwargs) await wrapped(**kwargs)
# invoke runtime via a root task.
return trio.run( return trio.run(
partial( partial(
_main, _main,
@ -259,13 +291,6 @@ def tractor_test(
def pytest_addoption( def pytest_addoption(
parser: pytest.Parser, parser: pytest.Parser,
): ):
# parser.addoption(
# "--ll",
# action="store",
# dest='loglevel',
# default='ERROR', help="logging level to set when testing"
# )
parser.addoption( parser.addoption(
"--spawn-backend", "--spawn-backend",
action="store", action="store",
@ -291,7 +316,7 @@ def pytest_addoption(
"--enable-stackscope", "--enable-stackscope",
action="store_true", action="store_true",
dest='enable_stackscope', dest='enable_stackscope',
# default=False, default=False,
help=( help=(
'Install `stackscope` SIGUSR1 handler in pytest + ' 'Install `stackscope` SIGUSR1 handler in pytest + '
'every spawned subactor for live trio task-tree ' 'every spawned subactor for live trio task-tree '
@ -317,6 +342,13 @@ def pytest_addoption(
def pytest_configure( def pytest_configure(
config: pytest.Config, config: pytest.Config,
): ):
# opts: Namespace = config.option
# print(
# f'PYTEST_CONFIGURE\n'
# f'capture={opts.capture!r}\n'
# )
# breakpoint()
backend: str = config.option.spawn_backend backend: str = config.option.spawn_backend
from tractor.spawn._spawn import try_set_start_method from tractor.spawn._spawn import try_set_start_method
try: try:
@ -414,6 +446,25 @@ def pytest_collection_modifyitems(
break break
@pytest.fixture(
scope="session",
autouse=True,
)
def alert_on_finish():
'''
Ring a terminal notification on full test session
completion to alert any would be human.
'''
# TODO, check attached to tty or skip!
yield # run all tests
print("\a") # trigger terminal bell
# ?TODO, any other nice-tricks/specific tuis we could try?
# - supposedly works in many terminals:
# >> print("\033]5;Alert: Tests Finished\a")
# - sway/i3-nag?
@pytest.fixture(scope='session') @pytest.fixture(scope='session')
def debug_mode( def debug_mode(
request: pytest.FixtureRequest, request: pytest.FixtureRequest,
@ -538,6 +589,7 @@ def pytest_generate_tests(
"start_method", "start_method",
[spawn_backend], [spawn_backend],
scope='module', scope='module',
ids=lambda item: f'start_method={spawn_backend}',
) )
# TODO, parametrize any `tpt_proto: str` declaring tests! # TODO, parametrize any `tpt_proto: str` declaring tests!
@ -548,3 +600,86 @@ def pytest_generate_tests(
# proto_tpts, # TODO, double check this list usage! # proto_tpts, # TODO, double check this list usage!
# scope='module', # scope='module',
# ) # )
def _is_forking_spawner(
start_method: str,
) -> bool:
return start_method in [
'main_thread_forkserver',
'mp_forkserver',
]
@pytest.fixture
def is_forking_spawner(
start_method: str,
) -> bool:
'''
Is the `pytest` run using a `fork()`ing process spawning-backend?
'''
return _is_forking_spawner
def maybe_xfail_for_spawner(
start_method: str,
is_forking_spawner: bool,
) -> None:
'''
Fork based spawning backends caude issues with `pytest`'s
fd-capture mechanism and can cause various suites to hang.
Instead this helper allows skipping/xfailing from a test
when a certain spawner + CLI-flag input is detected.
'''
if (
not _cap_sys_passed_as_flag
and
is_forking_spawner
):
pytest.skip(
f'Spawner {start_method!r} requires the flag,\n'
f'--capture=sys or similar..\n'
)
def maybe_override_capture(
request: pytest.FixtureRequest,
start_method: bool,
):
if _is_forking_spawner(start_method):
return request.getfixturevalue('capsys')
return request.getfixturevalue(
request.config.option.capture
)
@pytest.fixture
def set_fork_aware_capture(
request: pytest.FixtureRequest,
start_method: str,
) -> pytest.CaptureFixture|str:
'''
Force `--capture=sys` method for tests using
a forking-spawner backend due to fd-copying issues
which can oddly make certain tests hang/fail.
'''
if _cap_sys_passed_as_flag:
return 'sys'
capsys: pytest.CaptureFixture = maybe_override_capture(
request=request,
start_method=start_method,
)
return capsys
# XXX reset?
# with capsys.disabled():
# pass
# return partial(
# maybe_override_capture,
# request=request,
# start_method=start_method,
# )