[feat] The asyncio_event_loop mark specifying an optional event loop policy.

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

Author: seifertm

docs/source/reference/markers.rst

@@ -107,5 +107,26 @@ Similarly, a module-scoped loop is provided when adding the `asyncio_event_loop`
             global loop
             assert asyncio.get_running_loop() is loop
 
+The `asyncio_event_loop` mark supports an optional `policy` keyword argument to set the asyncio event loop policy.
+
+.. code-block:: python
+
+    import asyncio
+
+    import pytest
+
+
+    class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+        pass
+
+
+    @pytest.mark.asyncio_event_loop(policy=CustomEventLoopPolicy())
+    class TestUsesCustomEventLoopPolicy:
+        @pytest.mark.asyncio
+        async def test_uses_custom_event_loop_policy(self):
+            assert isinstance(asyncio.get_event_loop_policy(), CustomEventLoopPolicy)
+
+If no explicit policy is provided, the mark will use the loop policy returned by ``asyncio.get_event_loop_policy()``.
+
 .. |pytestmark| replace:: ``pytestmark``
 .. _pytestmark: http://doc.pytest.org/en/latest/example/markers.html#marking-whole-classes-or-modules

pytest_asyncio/plugin.py

@@ -31,6 +31,7 @@
     FixtureRequest,
     Function,
     Item,
+    Metafunc,
     Parser,
     PytestPluginManager,
     Session,
@@ -367,6 +368,7 @@ def pytest_collectstart(collector: pytest.Collector):
     for mark in marks:
         if not mark.name == "asyncio_event_loop":
             continue
+        event_loop_policy = mark.kwargs.get("policy", asyncio.get_event_loop_policy())
 
         # There seem to be issues when a fixture is shadowed by another fixture
         # and both differ in their params.
@@ -381,13 +383,23 @@ def pytest_collectstart(collector: pytest.Collector):
         @pytest.fixture(
             scope="class" if isinstance(collector, pytest.Class) else "module",
             name=event_loop_fixture_id,
+            params=(event_loop_policy,),
+            ids=(type(event_loop_policy).__name__,),
         )
         def scoped_event_loop(
             *args,  # Function needs to accept "cls" when collected by pytest.Class
+            request,
         ) -> Iterator[asyncio.AbstractEventLoop]:
-            loop = asyncio.get_event_loop_policy().new_event_loop()
+            new_loop_policy = request.param
+            old_loop_policy = asyncio.get_event_loop_policy()
+            old_loop = asyncio.get_event_loop()
+            asyncio.set_event_loop_policy(new_loop_policy)
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
             yield loop
             loop.close()
+            asyncio.set_event_loop_policy(old_loop_policy)
+            asyncio.set_event_loop(old_loop)
 
         # @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
@@ -430,6 +442,33 @@ def _hypothesis_test_wraps_coroutine(function: Any) -> bool:
     return _is_coroutine(function.hypothesis.inner_test)
 
 
+@pytest.hookimpl(tryfirst=True)
+def pytest_generate_tests(metafunc: Metafunc) -> None:
+    for event_loop_provider_node, _ in metafunc.definition.iter_markers_with_node(
+        "asyncio_event_loop"
+    ):
+        event_loop_fixture_id = event_loop_provider_node.stash.get(
+            _event_loop_fixture_id, None
+        )
+        if event_loop_fixture_id:
+            event_loop_fixture_id = event_loop_provider_node.stash[
+                _event_loop_fixture_id
+            ]
+            fixturemanager = metafunc.config.pluginmanager.get_plugin("funcmanage")
+            if "event_loop" in metafunc.fixturenames:
+                raise MultipleEventLoopsRequestedError(
+                    _MULTIPLE_LOOPS_REQUESTED_ERROR
+                    % (metafunc.definition.nodeid, event_loop_provider_node.nodeid),
+                )
+            # Add the scoped event loop fixture to Metafunc's list of fixture names and
+            # fixturedefs and leave the actual parametrization to pytest
+            metafunc.fixturenames.insert(0, event_loop_fixture_id)
+            metafunc._arg2fixturedefs[
+                event_loop_fixture_id
+            ] = fixturemanager._arg2fixturedefs[event_loop_fixture_id]
+            break
+
+
 @pytest.hookimpl(hookwrapper=True)
 def pytest_fixture_setup(
     fixturedef: FixtureDef, request: SubRequest
@@ -609,18 +648,12 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
         return
     event_loop_fixture_id = "event_loop"
     for node, mark in item.iter_markers_with_node("asyncio_event_loop"):
-        scoped_event_loop_provider_node = node
         event_loop_fixture_id = node.stash.get(_event_loop_fixture_id, None)
         if event_loop_fixture_id:
             break
     fixturenames = item.fixturenames  # type: ignore[attr-defined]
     # inject an event loop fixture for all async tests
     if "event_loop" in fixturenames:
-        if event_loop_fixture_id != "event_loop":
-            raise MultipleEventLoopsRequestedError(
-                _MULTIPLE_LOOPS_REQUESTED_ERROR
-                % (item.nodeid, scoped_event_loop_provider_node.nodeid),
-            )
         fixturenames.remove("event_loop")
     fixturenames.insert(0, event_loop_fixture_id)
     obj = getattr(item, "obj", None)

tests/markers/test_class_marker.py

@@ -126,3 +126,38 @@ async def test_remember_loop(self, event_loop):
     result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(errors=1)
     result.stdout.fnmatch_lines("*MultipleEventLoopsRequestedError: *")
+
+
+def test_asyncio_event_loop_mark_allows_specifying_the_loop_policy(
+    pytester: pytest.Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+                pass
+
+            @pytest.mark.asyncio_event_loop(policy=CustomEventLoopPolicy())
+            class TestUsesCustomEventLoopPolicy:
+
+                @pytest.mark.asyncio
+                async def test_uses_custom_event_loop_policy(self):
+                    assert isinstance(
+                        asyncio.get_event_loop_policy(),
+                        CustomEventLoopPolicy,
+                    )
+
+            @pytest.mark.asyncio
+            async def test_does_not_use_custom_event_loop_policy():
+                assert not isinstance(
+                    asyncio.get_event_loop_policy(),
+                    CustomEventLoopPolicy,
+                )
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)

tests/markers/test_module_marker.py

@@ -135,3 +135,53 @@ async def test_remember_loop(event_loop):
     result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(errors=1)
     result.stdout.fnmatch_lines("*MultipleEventLoopsRequestedError: *")
+
+
+def test_asyncio_event_loop_mark_allows_specifying_the_loop_policy(
+    pytester: Pytester,
+):
+    pytester.makepyfile(
+        __init__="",
+        custom_policy=dedent(
+            """\
+            import asyncio
+
+            class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+                pass
+            """
+        ),
+        test_uses_custom_policy=dedent(
+            """\
+            import asyncio
+            import pytest
+
+            from .custom_policy import CustomEventLoopPolicy
+
+            pytestmark = pytest.mark.asyncio_event_loop(policy=CustomEventLoopPolicy())
+
+            @pytest.mark.asyncio
+            async def test_uses_custom_event_loop_policy():
+                assert isinstance(
+                    asyncio.get_event_loop_policy(),
+                    CustomEventLoopPolicy,
+                )
+            """
+        ),
+        test_does_not_use_custom_policy=dedent(
+            """\
+            import asyncio
+            import pytest
+
+            from .custom_policy import CustomEventLoopPolicy
+
+            @pytest.mark.asyncio
+            async def test_does_not_use_custom_event_loop_policy():
+                assert not isinstance(
+                    asyncio.get_event_loop_policy(),
+                   CustomEventLoopPolicy,
+                )
+            """
+        ),
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)