Drop `wrapt` for `tractor_test`, revert to `functools`

Realized a bit late that (pretty sure) i already tried this using
`wrapt` idea and waay back and found the same "issue" XD

The `wrapt.decorator` transparently proxies `__code__` from the async
test fn, fooling `pytest`'s coroutine detection into skipping wrapped
tests as "unhandled coroutines". `functools.wraps` preserves the sig for
fixture injection via `__wrapped__` without leaking the async nature.

So i let `claude` rework the latest code to go back to using the old
stdlib wrapping again..

Deats,
- `functools.partial` replaces `wrapt.PartialCallableObjectProxy`.
- wrapper takes plain `**kwargs`; runtime settings extracted via
  `kwargs.get()` in `_main()`.
- `iscoroutinefunction()` guard moved before wrapper definition.
- drop all `*args` passing (fixture kwargs only).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
subint_spawner_backend
Gud Boi 2026-04-10 12:08:46 -04:00
parent f47010d7e9
commit 452a32fb23
1 changed files with 43 additions and 53 deletions

View File

@ -21,18 +21,17 @@ and applications.
'''
from functools import (
partial,
wraps,
)
import inspect
import platform
from typing import (
Callable,
Type,
)
import pytest
import tractor
import trio
import wrapt
def tractor_test(
@ -89,26 +88,34 @@ def tractor_test(
'''
__tracebackhide__: bool = hide_tb
# handle the decorator not called with () case.
# i.e. in `wrapt` support a deco-with-optional-args,
# https://wrapt.readthedocs.io/en/master/decorators.html#decorators-with-optional-arguments
# handle @tractor_test (no parens) vs @tractor_test(timeout=10)
if wrapped is None:
return wrapt.PartialCallableObjectProxy(
return partial(
tractor_test,
timeout=timeout,
hide_tb=hide_tb
hide_tb=hide_tb,
)
@wrapt.decorator
def wrapper(
wrapped: Callable,
instance: object|Type|None,
args: tuple,
kwargs: dict,
):
funcname: str = wrapped.__name__
if not inspect.iscoroutinefunction(wrapped):
raise TypeError(
f'Test-fn {funcname!r} must be an async-function !!'
)
# NOTE: we intentionally use `functools.wraps` instead of
# `@wrapt.decorator` here bc wrapt's transparent proxy makes
# `inspect.iscoroutinefunction(wrapper)` return `True` (it
# proxies `__code__` from the wrapped async fn), which causes
# pytest to skip the test as an "unhandled coroutine".
# `functools.wraps` preserves the signature for fixture
# injection (via `__wrapped__`) without leaking the async
# nature.
@wraps(wrapped)
def wrapper(**kwargs):
__tracebackhide__: bool = hide_tb
# NOTE, ensure we inject any test-fn declared fixture names.
# NOTE, ensure we inject any test-fn declared fixture
# names.
for kw in [
'reg_addr',
'loglevel',
@ -121,7 +128,7 @@ def tractor_test(
assert kw in kwargs
start_method = kwargs.get('start_method')
if platform.system() == "Windows":
if platform.system() == 'Windows':
if start_method is None:
kwargs['start_method'] = 'trio'
elif start_method != 'trio':
@ -129,60 +136,43 @@ def tractor_test(
'ONLY the `start_method="trio"` is supported on Windows.'
)
# open a root-actor, passing certain
# `tractor`-runtime-settings, then invoke the test-fn body as
# the root-most task.
# Open a root-actor, passing certain runtime-settings
# extracted from the fixture kwargs, then invoke the
# test-fn body as the root-most task.
#
# https://wrapt.readthedocs.io/en/master/decorators.html#processing-function-arguments
async def _main(
*args,
# runtime-settings
loglevel:str|None = None,
reg_addr:tuple|None = None,
start_method: str|None = None,
debug_mode: bool = False,
tpt_proto: str|None = None,
**kwargs,
):
# NOTE: we use `kwargs.get()` (not named params) so that
# the fixture values remain in `kwargs` and are forwarded
# to `wrapped()` — the test fn may declare the same
# fixtures in its own signature.
async def _main(**kwargs):
__tracebackhide__: bool = hide_tb
reg_addr = kwargs.get('reg_addr')
with trio.fail_after(timeout):
async with tractor.open_root_actor(
registry_addrs=[reg_addr] if reg_addr else None,
loglevel=loglevel,
start_method=start_method,
registry_addrs=(
[reg_addr] if reg_addr else None
),
loglevel=kwargs.get('loglevel'),
start_method=kwargs.get('start_method'),
# TODO: only enable when pytest is passed --pdb
debug_mode=debug_mode,
# TODO: only enable when pytest is passed
# --pdb
debug_mode=kwargs.get('debug_mode', False),
):
# invoke test-fn body IN THIS task
await wrapped(
*args,
**kwargs,
)
funcname = wrapped.__name__
if not inspect.iscoroutinefunction(wrapped):
raise TypeError(
f"Test-fn {funcname!r} must be an async-function !!"
)
await wrapped(**kwargs)
# invoke runtime via a root task.
return trio.run(
partial(
_main,
*args,
**kwargs,
)
)
return wrapper(
wrapped,
)
return wrapper
def pytest_addoption(