[feat] Introduce the asyncio_event_loop mark which provides a class-scoped asyncio event loop when a class has the mark.

Signed-off-by: Michael Seifert <[email protected]>

Author: seifertm

docs/source/reference/markers.rst

@@ -30,5 +30,57 @@ In *auto* mode, the ``pytest.mark.asyncio`` marker can be omitted, the marker is
 automatically to *async* test functions.
 
 
+``pytest.mark.asyncio_event_loop``
+==================================
+Test classes with this mark provide a class-scoped asyncio event loop.
+
+This functionality is orthogonal to the `asyncio` mark.
+That means the presence of this mark does not imply that async test functions inside the class are collected by pytest-asyncio.
+The collection happens automatically in `auto` mode.
+However, if you're using strict mode, you still have to apply the `asyncio` mark to your async test functions.
+
+The following code example uses the `asyncio_event_loop` mark to provide a shared event loop for all tests in `TestClassScopedLoop`:
+
+.. code-block:: python
+
+    import asyncio
+
+    import pytest
+
+
+    @pytest.mark.asyncio_event_loop
+    class TestClassScopedLoop:
+        loop: asyncio.AbstractEventLoop
+
+        @pytest.mark.asyncio
+        async def test_remember_loop(self):
+            TestClassScopedLoop.loop = asyncio.get_running_loop()
+
+        @pytest.mark.asyncio
+        async def test_this_runs_in_same_loop(self):
+            assert asyncio.get_running_loop() is TestClassScopedLoop.loop
+
+In *auto* mode, the ``pytest.mark.asyncio`` marker can be omitted:
+
+.. code-block:: python
+
+    import asyncio
+
+    import pytest
+
+
+    @pytest.mark.asyncio_event_loop
+    class TestClassScopedLoop:
+        loop: asyncio.AbstractEventLoop
+
+        async def test_remember_loop(self):
+            TestClassScopedLoop.loop = asyncio.get_running_loop()
+
+        async def test_this_runs_in_same_loop(self):
+            assert asyncio.get_running_loop() is TestClassScopedLoop.loop
+
+
+
+
 .. |pytestmark| replace:: ``pytestmark``
 .. _pytestmark: http://doc.pytest.org/en/latest/example/markers.html#marking-whole-classes-or-modules

pytest_asyncio/plugin.py

@@ -26,6 +26,7 @@
 )
 
 import pytest
+from _pytest.mark.structures import get_unpacked_marks
 from pytest import (
     Config,
     FixtureRequest,
@@ -176,6 +177,11 @@ def pytest_configure(config: Config) -> None:
         "mark the test as a coroutine, it will be "
         "run using an asyncio event loop",
     )
+    config.addinivalue_line(
+        "markers",
+        "asyncio_event_loop: "
+        "Provides an asyncio event loop in the scope of the marked test class",
+    )
 
 
 @pytest.hookimpl(tryfirst=True)
@@ -339,6 +345,33 @@ def pytest_pycollect_makeitem(
     return None
 
 
+@pytest.hookimpl
+def pytest_collectstart(collector: pytest.Collector):
+    if not isinstance(collector, pytest.Class):
+        return
+    # pytest.Collector.own_markers is empty at this point,
+    # so we rely on _pytest.mark.structures.get_unpacked_marks
+    marks = get_unpacked_marks(collector.obj, consider_mro=True)
+    for mark in marks:
+        if not mark.name == "asyncio_event_loop":
+            continue
+
+        @pytest.fixture(
+            scope="class",
+            name="event_loop",
+        )
+        def scoped_event_loop(cls) -> Iterator[asyncio.AbstractEventLoop]:
+            loop = asyncio.get_event_loop_policy().new_event_loop()
+            yield loop
+            loop.close()
+
+        # @pytest.fixture does not register the fixture anywhere, so pytest doesn't
+        # know it exists. We work around this by attaching the fixture function to the
+        # collected Python class, where it will be picked up by pytest.Class.collect()
+        collector.obj.__pytest_asyncio_scoped_event_loop = scoped_event_loop
+        break
+
+
 def pytest_collection_modifyitems(
     session: Session, config: Config, items: List[Item]
 ) -> None:

tests/markers/test_class_marker.py

@@ -1,5 +1,6 @@
 """Test if pytestmark works when defined on a class."""
 import asyncio
+from textwrap import dedent
 
 import pytest
 
@@ -23,3 +24,83 @@ async def inc():
 @pytest.fixture
 def sample_fixture():
     return None
+
+
+def test_asyncio_event_loop_mark_provides_class_scoped_loop_strict_mode(
+    pytester: pytest.Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio_event_loop
+            class TestClassScopedLoop:
+                loop: asyncio.AbstractEventLoop
+
+                @pytest.mark.asyncio
+                async def test_remember_loop(self):
+                    TestClassScopedLoop.loop = asyncio.get_running_loop()
+
+                @pytest.mark.asyncio
+                async def test_this_runs_in_same_loop(self):
+                    assert asyncio.get_running_loop() is TestClassScopedLoop.loop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)
+
+
+def test_asyncio_event_loop_mark_provides_class_scoped_loop_auto_mode(
+    pytester: pytest.Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio_event_loop
+            class TestClassScopedLoop:
+                loop: asyncio.AbstractEventLoop
+
+                async def test_remember_loop(self):
+                    TestClassScopedLoop.loop = asyncio.get_running_loop()
+
+                async def test_this_runs_in_same_loop(self):
+                    assert asyncio.get_running_loop() is TestClassScopedLoop.loop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=auto")
+    result.assert_outcomes(passed=2)
+
+
+def test_asyncio_event_loop_mark_is_inherited_to_subclasses(pytester: pytest.Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio_event_loop
+            class TestSuperClassWithMark:
+                pass
+
+            class TestWithoutMark(TestSuperClassWithMark):
+                loop: asyncio.AbstractEventLoop
+
+                @pytest.mark.asyncio
+                async def test_remember_loop(self):
+                    TestWithoutMark.loop = asyncio.get_running_loop()
+
+                @pytest.mark.asyncio
+                async def test_this_runs_in_same_loop(self):
+                    assert asyncio.get_running_loop() is TestWithoutMark.loop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)