[feat!] Provide a class-scoped asyncio event loop when a class has the asyncio mark.

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

Author: seifertm

Diff for: pytest_asyncio/plugin.py

@@ -26,6 +26,7 @@
 )
 
 import pytest
+from _pytest.mark.structures import get_unpacked_marks
 from pytest import (
     Config,
     FixtureRequest,
@@ -339,6 +340,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":
+            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:

Diff for: 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,52 @@ async def inc():
 @pytest.fixture
 def sample_fixture():
     return None
+
+
+def test_asyncio_mark_provides_class_scoped_loop(pytester: pytest.Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio
+            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=strict")
+    result.assert_outcomes(passed=2)
+
+
+def test_asyncio_mark_is_inherited_to_subclasses(pytester: pytest.Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio
+            class TestSuperClassWithMark:
+                pass
+
+            class TestWithoutMark(TestSuperClassWithMark):
+                loop: asyncio.AbstractEventLoop
+
+                async def test_remember_loop(self):
+                    TestWithoutMark.loop = asyncio.get_running_loop()
+
+                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)