[feat] The "policy" keyword argument to asyncio_event_loop allows passing an iterable of policies.

This causes tests under the _asyncio_event_loop_ mark to be parametrized with the different loop policies.

Signed-off-by: Michael Seifert <m.seifert@digitalernachschub.de>

Author: seifertm

docs/source/reference/markers.rst

@@ -126,6 +126,30 @@ The `asyncio_event_loop` mark supports an optional `policy` keyword argument to
         async def test_uses_custom_event_loop_policy(self):
             assert isinstance(asyncio.get_event_loop_policy(), CustomEventLoopPolicy)
 
+
+The ``policy`` keyword argument may also take an iterable of event loop policies. This causes tests under by the `asyncio_event_loop` mark to be parametrized with different policies:
+
+.. code-block:: python
+
+    import asyncio
+
+    import pytest
+
+    import pytest_asyncio
+
+
+    @pytest.mark.asyncio_event_loop(
+        policy=[
+            asyncio.DefaultEventLoopPolicy(),
+            uvloop.EventLoopPolicy(),
+        ]
+    )
+    class TestWithDifferentLoopPolicies:
+        @pytest.mark.asyncio
+        async def test_parametrized_loop(self):
+            pass
+
+
 If no explicit policy is provided, the mark will use the loop policy returned by ``asyncio.get_event_loop_policy()``.
 
 .. |pytestmark| replace:: ``pytestmark``

pytest_asyncio/plugin.py

@@ -369,6 +369,11 @@ def pytest_collectstart(collector: pytest.Collector):
         if not mark.name == "asyncio_event_loop":
             continue
         event_loop_policy = mark.kwargs.get("policy", asyncio.get_event_loop_policy())
+        policy_params = (
+            event_loop_policy
+            if isinstance(event_loop_policy, Iterable)
+            else (event_loop_policy,)
+        )
 
         # There seem to be issues when a fixture is shadowed by another fixture
         # and both differ in their params.
@@ -383,8 +388,8 @@ 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__,),
+            params=policy_params,
+            ids=tuple(type(policy).__name__ for policy in policy_params),
         )
         def scoped_event_loop(
             *args,  # Function needs to accept "cls" when collected by pytest.Class

tests/markers/test_class_marker.py

@@ -161,3 +161,30 @@ async def test_does_not_use_custom_event_loop_policy():
     )
     result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(passed=2)
+
+
+def test_asyncio_event_loop_mark_allows_specifying_multiple_loop_policies(
+    pytester: pytest.Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+
+            import pytest
+
+            @pytest.mark.asyncio_event_loop(
+                policy=[
+                    asyncio.DefaultEventLoopPolicy(),
+                    asyncio.DefaultEventLoopPolicy(),
+                ]
+            )
+            class TestWithDifferentLoopPolicies:
+                @pytest.mark.asyncio
+                async def test_parametrized_loop(self):
+                    pass
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)

tests/markers/test_module_marker.py

@@ -185,3 +185,30 @@ async def test_does_not_use_custom_event_loop_policy():
     )
     result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(passed=2)
+
+
+def test_asyncio_event_loop_mark_allows_specifying_multiple_loop_policies(
+    pytester: Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+
+            import pytest
+
+            pytestmark = pytest.mark.asyncio_event_loop(
+                policy=[
+                    asyncio.DefaultEventLoopPolicy(),
+                    asyncio.DefaultEventLoopPolicy(),
+                ]
+            )
+
+            @pytest.mark.asyncio
+            async def test_parametrized_loop():
+                pass
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)