forked from goodboy/tractor
1
0
Fork 0

Merge pull request #45 from tgoodlet/expose_tractor_test

Expose `tractor_test` for external use
loglevel_to_tractor_tests
goodboy 2018-11-26 14:42:24 -05:00 committed by GitHub
commit 58ebacf0f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 46 additions and 29 deletions

View File

@ -35,6 +35,7 @@ setup(
platforms=['linux'],
packages=[
'tractor',
'tractor.testing',
],
install_requires=['msgpack', 'trio>0.8', 'async_generator', 'colorlog'],
tests_require=['pytest'],

View File

@ -2,10 +2,10 @@
``tractor`` testing!!
"""
import random
from functools import partial, wraps
import pytest
import tractor
from tractor.testing import tractor_test
pytest_plugins = ['pytester']
@ -28,20 +28,3 @@ def loglevel(request):
@pytest.fixture(scope='session')
def arb_addr():
return _arb_addr
def tractor_test(fn):
"""
Use:
@tractor_test
async def test_whatever():
await ...
"""
@wraps(fn)
def wrapper(*args, **kwargs):
# __tracebackhide__ = True
return tractor.run(
partial(fn, *args, **kwargs), arbiter_addr=_arb_addr)
return wrapper

View File

@ -8,8 +8,6 @@ import subprocess
import pytest
import tractor
from conftest import tractor_test
@ -72,8 +70,9 @@ async def test_cancel_remote_arbiter(daemon, arb_addr):
pass
@tractor_test
async def test_register_duplicate_name(daemon):
async def test_register_duplicate_name(daemon, arb_addr):
async def main():
assert not tractor.current_actor().is_arbiter
async with tractor.open_nursery() as n:
p1 = await n.start_actor('doggy')
@ -83,3 +82,7 @@ async def test_register_duplicate_name(daemon):
assert portal.channel.uid in (p2.channel.uid, p1.channel.uid)
await n.cancel()
# run it manually since we want to start **after**
# the other "daemon" program
tractor.run(main, arb_addr=arbiter_addr)

View File

@ -0,0 +1 @@
from ._tractor_test import tractor_test

View File

@ -0,0 +1,29 @@
import inspect
from functools import partial, wraps
from .. import run
__all__ = ['tractor_test']
def tractor_test(fn):
"""
Use:
@tractor_test
async def test_whatever():
await ...
If an ``arb_addr`` (a socket addr tuple) is defined in the
`pytest` fixture space it will be automatically injected.
"""
@wraps(fn)
def wrapper(*args, arb_addr=None, **kwargs):
# __tracebackhide__ = True
if 'arb_addr' in inspect.signature(fn).parameters:
kwargs['arb_addr'] = arb_addr
return run(
partial(fn, *args, **kwargs), arbiter_addr=arb_addr)
return wrapper