From f11bd95ac7e6be312434ae068a60339db7a40cab Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Fri, 1 May 2015 17:59:44 -0300 Subject: [PATCH] Add support for test methods - Inspired on the default implementation for pytest_pyfunc_call for normal test functions - Refactor pytest_pyfunc_call to remove duplication fixes #5 --- pytest_asyncio/plugin.py | 54 +++++++++++++++++++--------------------- tests/test_simple.py | 13 +++++++++- 2 files changed, 37 insertions(+), 30 deletions(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 8a3876d8..7e4a6be4 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -27,40 +27,36 @@ def pytest_pycollect_makeitem(collector, name, obj): return list(collector._genfunctions(name, obj)) -def _argnames(func): - spec = inspect.getfullargspec(func) - if spec.defaults: - return spec.args[:-len(spec.defaults)] - return spec.args - - @pytest.mark.tryfirst def pytest_pyfunc_call(pyfuncitem): - """A hook wrapper.""" - if 'asyncio_process_pool' in pyfuncitem.keywords: - event_loop = pyfuncitem.funcargs.get('event_loop_process_pool') - funcargs = dict((arg, pyfuncitem.funcargs[arg]) - for arg in _argnames(pyfuncitem.obj)) - event_loop.run_until_complete(asyncio.async(pyfuncitem.obj(**funcargs))) - # prevent other pyfunc calls from executing - return True - elif 'asyncio' in pyfuncitem.keywords: - event_loop = pyfuncitem.funcargs.get('event_loop') - funcargs = dict((arg, pyfuncitem.funcargs[arg]) - for arg in _argnames(pyfuncitem.obj)) - event_loop.run_until_complete(asyncio.async(pyfuncitem.obj(**funcargs))) - # prevent other pyfunc calls from executing - return True + """ + Run asyncio marked test functions in an event loop instead of a normal + function call. + """ + for marker_name, fixture_name in _markers_2_fixtures.items(): + if marker_name in pyfuncitem.keywords: + event_loop = pyfuncitem.funcargs[fixture_name] + funcargs = pyfuncitem.funcargs + testargs = {arg: funcargs[arg] + for arg in pyfuncitem._fixtureinfo.argnames} + event_loop.run_until_complete( + asyncio.async(pyfuncitem.obj(**testargs))) + return True def pytest_runtest_setup(item): - if 'asyncio' in item.keywords and 'event_loop' not in item.fixturenames: - # inject an event loop fixture for all async tests - item.fixturenames.append('event_loop') - - if ('asyncio_process_pool' in item.keywords and - 'event_loop_process_pool' not in item.fixturenames): - item.fixturenames.append('event_loop_process_pool') + for marker, fixture in _markers_2_fixtures.items(): + if marker in item.keywords and fixture not in item.fixturenames: + # inject an event loop fixture for all async tests + item.fixturenames.append(fixture) + + +# maps marker to the name of the event loop fixture that will be available +# to marked test functions +_markers_2_fixtures = { + 'asyncio': 'event_loop', + 'asyncio_process_pool': 'event_loop_process_pool', +} @pytest.fixture diff --git a/tests/test_simple.py b/tests/test_simple.py index 1be77aaf..a8bf370e 100644 --- a/tests/test_simple.py +++ b/tests/test_simple.py @@ -87,4 +87,15 @@ def closer(_, writer): port=unused_tcp_port) server1.close() - yield from server1.wait_closed() \ No newline at end of file + yield from server1.wait_closed() + + +class Test: + """Test that asyncio marked functions work in test methods.""" + + @pytest.mark.asyncio + def test_asyncio_marker_method(self): + """Test the asyncio pytest marker in a Test class.""" + url = 'http://httpbin.org/get' + resp = yield from simple_http_client(url) + assert b'HTTP/1.1 200 OK' in resp