Async fixtures, explicit user-defined loops and more

README.rst

@@ -1,12 +1,8 @@
 pytest-asyncio: pytest support for asyncio
 ==========================================
 
-.. image:: https://img.shields.io/pypi/v/pytest-asyncio.svg
-    :target: https://pypi.python.org/pypi/pytest-asyncio
-.. image:: https://travis-ci.org/pytest-dev/pytest-asyncio.svg?branch=master
-    :target: https://travis-ci.org/pytest-dev/pytest-asyncio
-.. image:: https://coveralls.io/repos/pytest-dev/pytest-asyncio/badge.svg
-    :target: https://coveralls.io/r/pytest-dev/pytest-asyncio
+.. image:: https://travis-ci.org/malinoff/pytest-asyncio.svg?branch=master
+    :target: https://travis-ci.org/malinoff/pytest-asyncio
 
 pytest-asyncio is an Apache2 licensed library, written in Python, for testing
 asyncio code with pytest.
@@ -15,30 +11,115 @@ asyncio code is usually written in the form of coroutines, which makes it
 slightly more difficult to test using normal testing tools. pytest-asyncio
 provides useful fixtures and markers to make testing easier.
 
+Original readme can be found in original `pytest-asyncio`_ repository.
+
+This fork completely changes how ``pytest-asyncio`` works with coroutine-based
+tests and loops:
+
+    * No more closing the loop after each test coroutine function. It's up
+      to the developer to choose when the loop is closed.
+
+    * The use of global event loop is now forbidden by default. You can accept
+      it by providing ``accept_global_loop=True`` to ``@asyncio.mark.asyncio``
+
+    * No more implicit loops defined by the plugin. In order to use a loop,
+      you must explicitly create and request ``loop`` fixture, otherwise
+      the plugin will raise a ``MissingLoopFixture`` exception. This fixture
+      can be named anything, but requires to return an instance of
+      ``asyncio.AbstractEventLoop``. There is one exception: if
+      ``accept_global_loop`` is ``True`` AND a ``loop`` fixture is not requested,
+      the plugin will use the global loop.
+
+The advantages are:
+
+    * You do not rely on implicit event loops created by the plugin.
+      Want to use ``concurrent.futures.ThreadPoolExecutor``? Easy!
+
+      .. code-block:: python
+
+          @pytest.yield_fixture
+          def threadpooled_loop():
+              loop = asyncio.get_event_loop_policy().new_event_loop()
+              loop.set_default_executor(concurrent.futures.ThreadPoolExecutor())
+              yield loop
+              loop.close()
+
+    * Lifetimes of loop fixtures can be expanded to ``module`` or ``session``
+      scopes easily (in the original plugin it is not possible because the loop
+      closes after each test coroutine function):
+
+      .. code-block:: python
+
+          @pytest.yield_fixture(scope='module')
+          def loop():
+              ...
+
+You can also create coroutine-based fixtures:
+
+    .. code-block:: python
+
+        @pytest_asyncio.async_fixture
+        async def async_fixture(loop):
+            await asyncio.sleep(1, loop=loop)
+            return 'something'
+
+By default, using a global asyncio loop is forbidden by default, similarly
+to the test coroutine functions. To accept it, provide ``accept_global_loop=True``
+to the ``@pytest_asyncio.async_fixture`` decorator.
+
+Examples compared to the original examples:
+
 .. code-block:: python
 
+    # Original
     @pytest.mark.asyncio
     async def test_some_asyncio_code():
         res = await library.do_something()
         assert b'expected result' == res
 
+    # Fork
+    import asyncio
+    @pytest.yield_fixture
+    def loop():
+        loop = asyncio.get_event_loop_policy().new_event_loop()
+        yield loop
+        loop.close()
+
+    @pytest.mark.asyncio
+    async def test_some_asyncio_code(loop):
+        res = await library.do_something(loop=loop)
+        assert b'expected result' == res
+
 or, if you're using the pre-Python 3.5 syntax:
 
 .. code-block:: python
 
+    # Original
     @pytest.mark.asyncio
     def test_some_asyncio_code():
         res = yield from library.do_something()
         assert b'expected result' == res
 
+    # Fork
+    import asyncio
+    @pytest.fixture
+    def loop():
+        return asyncio.get_event_loop_policy().new_event_loop()
+
+    @pytest.mark.asyncio
+    async def test_some_asyncio_code(loop):
+        res = await library.do_something(loop=loop)
+        assert b'expected result' == res
+
 pytest-asyncio has been strongly influenced by pytest-tornado_.
 
+.. _pytest-asyncio: https://github.com/pytest-dev/pytest-asyncio/blob/master/README.rst
 .. _pytest-tornado: https://github.com/eugeniy/pytest-tornado
 
 Features
 --------
 
-- fixtures for creating and injecting versions of the asyncio event loop
+- pluggable fixtures of the asyncio event loops
 - fixtures for injecting unused tcp ports
 - pytest markers for treating tests as asyncio coroutines
 - easy testing with non-default event loops
@@ -51,48 +132,13 @@ To install pytest-asyncio, simply:
 
 .. code-block:: bash
 
-    $ pip install pytest-asyncio
+    $ pip install git+https://github.com/malinoff/pytest-asyncio
 
 This is enough for pytest to pick up pytest-asyncio.
 
 Fixtures
 --------
 
-``event_loop``
-~~~~~~~~~~~~~~
-Creates and injects a new instance of the default asyncio event loop. The loop
-will be closed at the end of the test.
-
-Note that just using the ``event_loop`` fixture won't make your test function
-a coroutine. You'll need to interact with the event loop directly, using methods
-like ``event_loop.run_until_complete``. See the ``pytest.mark.asyncio`` marker
-for treating test functions like coroutines.
-
-.. code-block:: python
-
-    def test_http_client(event_loop):
-        url = 'http://httpbin.org/get'
-        resp = event_loop.run_until_complete(http_client(url))
-        assert b'HTTP/1.1 200 OK' in resp
-
-This fixture can be easily overridden in any of the standard pytest locations
-(e.g. directly in the test file, or in ``conftest.py``) to use a non-default
-event loop. This will take effect even if you're using the
-``pytest.mark.asyncio`` marker and not the ``event_loop`` fixture directly.
-
-.. code-block:: python
-
-    @pytest.fixture()
-    def event_loop():
-        return MyCustomLoop()
-
-
-``event_loop_process_pool``
-~~~~~~~~~~~~~~~~~~~~~~~~~~~
-The ``event_loop_process_pool`` fixture is almost identical to the
-``event_loop`` fixture, except the created event loop will have a
-``concurrent.futures.ProcessPoolExecutor`` set as the default executor.
-
 ``unused_tcp_port``
 ~~~~~~~~~~~~~~~~~~~
 Finds and yields a single unused TCP port on the localhost interface. Useful for
@@ -112,25 +158,17 @@ when several unused TCP ports are required in a test.
 Markers
 -------
 
-``pytest.mark.asyncio(forbid_global_loop=False)``
+``pytest.mark.asyncio(accept_global_loop=False)``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Mark your test coroutine with this marker and pytest will execute it as an
-asyncio task using the event loop provided by the ``event_loop`` fixture. See
+asyncio task using the event loop provided by a ``loop`` fixture. See
 the introductory section for an example.
 
-The event loop used can be overriden by overriding the ``event_loop`` fixture
-(see above).
+A different event loop can be provided easily, see the introductory section.
 
-If ``forbid_global_loop`` is true, ``asyncio.get_event_loop()`` will result
+If ``accept_global_loop`` is false, ``asyncio.get_event_loop()`` will result
 in exceptions, ensuring your tests are always passing the event loop explicitly.
 
-``pytest.mark.asyncio_process_pool(forbid_global_loop=False)``
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-The ``asyncio_process_pool`` marker is almost identical to the ``asyncio``
-marker, except the event loop used will have a
-``concurrent.futures.ProcessPoolExecutor`` set as the default executor.
-
-
 Contributing
 ------------
 Contributions are very welcome. Tests can be run with ``tox``, please ensure

pytest_asyncio/__init__.py

@@ -1 +1,3 @@
+from .fixture import async_fixture
+
 __version__ = '0.4.0'

pytest_asyncio/exceptions.py

@@ -0,0 +1,3 @@
+class MissingLoopFixture(Exception):
+    """Raised if a test coroutine function does not request a loop fixture."""
+    pass

pytest_asyncio/fixture.py

@@ -0,0 +1,44 @@
+import asyncio
+import itertools
+import functools
+
+import pytest  # noqa
+import _pytest.python
+
+from .utils import find_loop, maybe_accept_global_loop
+
+
+class AsyncFixtureFunctionMarker(_pytest.python.FixtureFunctionMarker):
+
+    def __init__(self, *args, accept_global_loop, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.accept_global_loop = accept_global_loop
+
+    def __call__(self, coroutine):
+        if not asyncio.iscoroutinefunction(coroutine):
+            raise ValueError('Only coroutine functions supported')
+
+        @functools.wraps(coroutine)
+        def inner(*args, **kwargs):
+            event_loop = find_loop(itertools.chain(args, kwargs.values()))
+            with maybe_accept_global_loop(
+                    event_loop, self.accept_global_loop) as loop:
+                return loop.run_until_complete(coroutine(*args, **kwargs))
+
+        inner._pytestfixturefunction = self
+        return inner
+
+
+def async_fixture(scope='function', params=None, autouse=False, ids=None,
+                  accept_global_loop=False):
+    if callable(scope) and params is None and not autouse:
+        # direct invocation
+        marker = AsyncFixtureFunctionMarker(
+            'function', params, autouse, accept_global_loop=accept_global_loop,
+        )
+        return marker(scope)
+    if params is not None and not isinstance(params, (list, tuple)):
+        params = list(params)
+    return AsyncFixtureFunctionMarker(
+        scope, params, autouse, ids=ids, accept_global_loop=accept_global_loop,
+    )

pytest_asyncio/plugin.py

@@ -1,14 +1,11 @@
+import socket
 import asyncio
-from concurrent.futures import ProcessPoolExecutor
-from contextlib import closing
 import inspect
-import socket
-import pytest
+import contextlib
 
+import pytest
 
-class ForbiddenEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
-    """An event loop policy that raises errors on any operation."""
-    pass
+from .utils import find_loop, maybe_accept_global_loop
 
 
 def _is_coroutine(obj):
@@ -32,8 +29,7 @@ def pytest_configure(config):
 def pytest_pycollect_makeitem(collector, name, obj):
     if collector.funcnamefilter(name) and _is_coroutine(obj):
         item = pytest.Function(name, parent=collector)
-        if ('asyncio' in item.keywords or
-           'asyncio_process_pool' in item.keywords):
+        if 'asyncio' in item.keywords:
             return list(collector._genfunctions(name, obj))
 
 
@@ -43,66 +39,26 @@ def pytest_pyfunc_call(pyfuncitem):
     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]
-
-            forbid_global_loop = pyfuncitem.keywords[marker_name].kwargs.get('forbid_global_loop')
-
-            policy = asyncio.get_event_loop_policy()
-            if forbid_global_loop:
-                asyncio.set_event_loop_policy(ForbiddenEventLoopPolicy())
-            else:
-                policy.set_event_loop(event_loop)
-
-            funcargs = pyfuncitem.funcargs
-            testargs = {arg: funcargs[arg]
-                        for arg in pyfuncitem._fixtureinfo.argnames}
-            try:
-                event_loop.run_until_complete(
-                    asyncio.async(pyfuncitem.obj(**testargs), loop=event_loop))
-            finally:
-                if forbid_global_loop:
-                    asyncio.set_event_loop_policy(policy)
-                event_loop.close()
-                return True
-
-
-def pytest_runtest_setup(item):
-    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',
-}
-
+    if 'asyncio' in pyfuncitem.keywords:
+        marker_kwargs = pyfuncitem.keywords['asyncio'].kwargs
+        accept_global_loop = marker_kwargs.get('accept_global_loop', False)
 
-@pytest.fixture
-def event_loop(request):
-    """Create an instance of the default event loop for each test case."""
-    policy = asyncio.get_event_loop_policy()
-    return policy.new_event_loop()
-
-
-@pytest.fixture
-def event_loop_process_pool(event_loop):
-    """Create a fresh instance of the default event loop.
+        event_loop = find_loop(pyfuncitem.funcargs.values())
 
-    The event loop will have a process pool set as the default executor."""
-    event_loop.set_default_executor(ProcessPoolExecutor())
-    return event_loop
+        funcargs = pyfuncitem.funcargs
+        testargs = {arg: funcargs[arg]
+                    for arg in pyfuncitem._fixtureinfo.argnames}
+        with maybe_accept_global_loop(
+                event_loop, accept_global_loop) as loop:
+            loop.run_until_complete(asyncio.async(pyfuncitem.obj(**testargs),
+                                                  loop=loop))
+            return True
 
 
 @pytest.fixture
 def unused_tcp_port():
     """Find an unused localhost TCP port from 1024-65535 and return it."""
-    with closing(socket.socket()) as sock:
+    with contextlib.closing(socket.socket()) as sock:
         sock.bind(('127.0.0.1', 0))
         return sock.getsockname()[1]