Implement pytestmark.

Author: Tinche

README.rst

@@ -69,6 +69,10 @@ 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.
 
+Simply using this fixture will not set the generated event loop as the
+default asyncio event loop, or change the asyncio event loop policy in any way.
+Use ``pytest.mark.asyncio`` for this purpose.
+
 .. code-block:: python
 
     def test_http_client(event_loop):
@@ -89,11 +93,12 @@ event loop. This will take effect even if you're using the
         yield loop
         loop.close()
 
-A special pytest hook will ensure the produced loop is either set as the
-default global loop, or a special, error-throwing event loop policy is installed
-as the default policy (depending on the ``forbid_global_loop`` parameter).
-Fixtures depending on the ``event_loop`` fixture can expect the policy to be
-properly modified when they run.
+If the ``pytest.mark.asyncio`` marker is applied, a pytest hook will
+ensure the produced loop is either set as the default global loop, or a special,
+error-throwing event loop policy is installed as the default policy (depending
+on the ``forbid_global_loop`` parameter). Fixtures depending on the
+``event_loop`` fixture can expect the policy to be properly modified when they
+run.
 
 ``event_loop_process_pool``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -132,6 +137,26 @@ The event loop used can be overriden by overriding the ``event_loop`` fixture
 If ``forbid_global_loop`` is true, ``asyncio.get_event_loop()`` will result
 in exceptions, ensuring your tests are always passing the event loop explicitly.
 
+In order to make your test code a little more concise, the pytest |pytestmark|_
+feature can be used to mark entire modules or classes with this marker.
+Only test coroutines will be affected (by default, coroutines prefixed by
+``test_``), so, for example, fixtures are safe to define.
+
+.. code-block:: python
+
+    import asyncio
+    import pytest
+
+    # All test coroutines will be treated as marked.
+    pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
+
+    async def test_example(event_loop):
+        """No marker!"""
+        await asyncio.sleep(0, loop=event_loop)
+
+.. |pytestmark| replace:: ``pytestmark``
+.. _pytestmark: http://doc.pytest.org/en/latest/example/markers.html#marking-whole-classes-or-modules
+
 ``pytest.mark.asyncio_process_pool(forbid_global_loop=False)``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 The ``asyncio_process_pool`` marker is almost identical to the ``asyncio``
@@ -141,6 +166,10 @@ marker, except the event loop used will have a
 Changelog
 ---------
 
+0.6.0 (UNRELEASED)
+~~~~~~~~~~~~~~~~~~
+- ``pytestmark`` now works on both module and class level.
+
 0.5.0 (2016-09-07)
 ~~~~~~~~~~~~~~~~~~
 - Introduced a changelog.

pytest_asyncio/plugin.py

@@ -1,10 +1,14 @@
 import asyncio
-from concurrent.futures import ProcessPoolExecutor
-from contextlib import closing
 import inspect
 import socket
+
+from concurrent.futures import ProcessPoolExecutor
+from contextlib import closing
+
 import pytest
 
+from _pytest.python import transfer_markers
+
 
 class ForbiddenEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
     """An event loop policy that raises errors on any operation."""
@@ -30,10 +34,18 @@ def pytest_configure(config):
 
 @pytest.mark.tryfirst
 def pytest_pycollect_makeitem(collector, name, obj):
+    """A pytest hook to collect asyncio coroutines."""
     if collector.funcnamefilter(name) and _is_coroutine(obj):
         item = pytest.Function(name, parent=collector)
+
+        # Due to how pytest test collection works, module-level pytestmarks
+        # are applied after the collection step. Since this is the collection
+        # step, we look ourselves.
+        transfer_markers(obj, item.cls, item.module)
+        item = pytest.Function(name, parent=collector)  # To reload keywords.
+
         if ('asyncio' in item.keywords or
-           'asyncio_process_pool' in item.keywords):
+            'asyncio_process_pool' in item.keywords):
             return list(collector._genfunctions(name, obj))
 
 

tests/conftest.py

@@ -6,6 +6,8 @@
 collect_ignore = []
 if sys.version_info[:2] < (3, 5):
     collect_ignore.append("test_simple_35.py")
+    collect_ignore.append("markers/test_class_marker_35.py")
+    collect_ignore.append("markers/test_module_marker_35.py")
 
 
 @pytest.yield_fixture()

tests/markers/test_class_marker.py

@@ -0,0 +1,26 @@
+"""Test if pytestmark works when defined on a class."""
+import asyncio
+import pytest
+
+
+class TestPyTestMark:
+    pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
+
+    @asyncio.coroutine
+    def test_no_global_loop(self, event_loop, sample_fixture):
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop()
+        counter = 1
+
+        @asyncio.coroutine
+        def inc():
+            nonlocal counter
+            counter += 1
+            yield from asyncio.sleep(0, loop=event_loop)
+        yield from asyncio.ensure_future(inc(), loop=event_loop)
+        assert counter == 2
+
+
+@pytest.fixture
+def sample_fixture():
+    return None

tests/markers/test_class_marker_35.py

@@ -0,0 +1,23 @@
+"""Test if pytestmark works when defined on a class."""
+import asyncio
+import pytest
+
+
+class TestPyTestMark:
+    pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
+
+    async def test_no_global_loop(self, event_loop, sample_fixture):
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop()
+        counter = 1
+        async def inc():
+            nonlocal counter
+            counter += 1
+            await asyncio.sleep(0, loop=event_loop)
+        await asyncio.ensure_future(inc(), loop=event_loop)
+        assert counter == 2
+
+
+@pytest.fixture
+def sample_fixture():
+    return None

tests/markers/test_module_marker.py

@@ -0,0 +1,46 @@
+"""Test if pytestmark works when defined in a module.
+
+Pre-3.5 version.
+"""
+import asyncio
+
+import pytest
+
+pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
+
+
+class TestPyTestMark:
+    @asyncio.coroutine
+    def test_no_global_loop_method(self, event_loop, sample_fixture):
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop()
+        counter = 1
+
+        @asyncio.coroutine
+        def inc():
+            nonlocal counter
+            counter += 1
+            yield from asyncio.sleep(0, loop=event_loop)
+
+        yield from asyncio.async(inc(), loop=event_loop)
+        assert counter == 2
+
+@asyncio.coroutine
+def test_no_global_loop_coroutine(event_loop, sample_fixture):
+    with pytest.raises(NotImplementedError):
+        asyncio.get_event_loop()
+    counter = 1
+
+    @asyncio.coroutine
+    def inc():
+        nonlocal counter
+        counter += 1
+        yield from asyncio.sleep(0, loop=event_loop)
+
+    yield from asyncio.async(inc(), loop=event_loop)
+    assert counter == 2
+
+
+@pytest.fixture
+def sample_fixture():
+    return None

tests/markers/test_module_marker_35.py

@@ -0,0 +1,36 @@
+"""Test if pytestmark works when defined in a module."""
+import asyncio
+
+import pytest
+
+pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
+
+
+class TestPyTestMark:
+    async def test_no_global_loop_method(self, event_loop, sample_fixture):
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop()
+
+        counter = 1
+        async def inc():
+            nonlocal counter
+            counter += 1
+            await asyncio.sleep(0, loop=event_loop)
+        await asyncio.ensure_future(inc(), loop=event_loop)
+        assert counter == 2
+
+async def test_no_global_loop_coroutine(event_loop, sample_fixture):
+    with pytest.raises(NotImplementedError):
+        asyncio.get_event_loop()
+    counter = 1
+    async def inc():
+        nonlocal counter
+        counter += 1
+        await asyncio.sleep(0, loop=event_loop)
+    await asyncio.ensure_future(inc(), loop=event_loop)
+    assert counter == 2
+
+
+@pytest.fixture
+def sample_fixture():
+    return None