[feat] Introduce the event_loop_policy fixture.

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

Author: seifertm

docs/source/how-to-guides/uvloop.rst

@@ -2,12 +2,17 @@
 How to test with uvloop
 =======================
 
+Redefinig the *event_loop_policy* fixture will parametrize all async tests. The following example causes all async tests to run multiple times, once for each event loop in the fixture parameters:
 Replace the default event loop policy in your *conftest.py:*
 
 .. code-block:: python
 
-    import asyncio
-
+    import pytest
     import uvloop
 
-    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+
+    @pytest.fixture(scope="session")
+    def event_loop_policy():
+        return uvloop.EventLoopPolicy()
+
+You may choose to limit the scope of the fixture to *package,* *module,* or *class,* if you only want a subset of your tests to run with uvloop.

docs/source/reference/changelog.rst

@@ -5,6 +5,7 @@ Changelog
 0.23.0 (UNRELEASED)
 ===================
 - Removes pytest-trio from the test dependencies `#620 <https://github.com/pytest-dev/pytest-asyncio/pull/620>`_
+- Introduces the *event_loop_policy* fixture which allows testing with non-default or multiple event loops  `#662 <https://github.com/pytest-dev/pytest-asyncio/pull/662>`_
 
 0.22.0 (2023-10-31)
 ===================

docs/source/reference/fixtures/event_loop_policy_example.py

@@ -0,0 +1,17 @@
+import asyncio
+
+import pytest
+
+
+class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+    pass
+
+
+@pytest.fixture(scope="module")
+def event_loop_policy(request):
+    return CustomEventLoopPolicy()
+
+
+@pytest.mark.asyncio(scope="module")
+async def test_uses_custom_event_loop_policy():
+    assert isinstance(asyncio.get_event_loop_policy(), CustomEventLoopPolicy)

docs/source/reference/fixtures/event_loop_policy_parametrized_example.py

@@ -0,0 +1,23 @@
+import asyncio
+from asyncio import DefaultEventLoopPolicy
+
+import pytest
+
+
+class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+    pass
+
+
+@pytest.fixture(
+    params=(
+        DefaultEventLoopPolicy(),
+        CustomEventLoopPolicy(),
+    ),
+)
+def event_loop_policy(request):
+    return request.param
+
+
+@pytest.mark.asyncio
+async def test_uses_custom_event_loop_policy():
+    assert isinstance(asyncio.get_event_loop_policy(), DefaultEventLoopPolicy)

docs/source/reference/fixtures/index.rst

@@ -22,6 +22,24 @@ If you need to change the type of the event loop, prefer setting a custom event
 If the ``pytest.mark.asyncio`` decorator is applied to a test function, the ``event_loop``
 fixture will be requested automatically by the test function.
 
+event_loop_policy
+=================
+Returns the event loop policy used to create asyncio event loops.
+The default return value is *asyncio.get_event_loop_policy().*
+
+This fixture can be overridden when a different event loop policy should be used.
+
+.. include:: event_loop_policy_example.py
+    :code: python
+
+Multiple policies can be provided via fixture parameters.
+The fixture is automatically applied to all pytest-asyncio tests.
+Therefore, all tests managed by pytest-asyncio are run once for each fixture parameter.
+The following example runs the test with different event loop policies.
+
+.. include:: event_loop_policy_parametrized_example.py
+    :code: python
+
 unused_tcp_port
 ===============
 Finds and yields a single unused TCP port on the localhost interface. Useful for

docs/source/reference/markers/class_scoped_loop_custom_policies_strict_mode_example.py

@@ -3,12 +3,16 @@
 import pytest
 
 
-@pytest.mark.asyncio_event_loop(
-    policy=[
+@pytest.fixture(
+    params=[
         asyncio.DefaultEventLoopPolicy(),
         asyncio.DefaultEventLoopPolicy(),
     ]
 )
+def event_loop_policy(request):
+    return request.param
+
+
 class TestWithDifferentLoopPolicies:
     @pytest.mark.asyncio
     async def test_parametrized_loop(self):

docs/source/reference/markers/class_scoped_loop_custom_policy_strict_mode_example.py

@@ -7,7 +7,12 @@ class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
     pass
 
 
-@pytest.mark.asyncio_event_loop(policy=CustomEventLoopPolicy())
+@pytest.fixture(scope="class")
+def event_loop_policy(request):
+    return CustomEventLoopPolicy()
+
+
+@pytest.mark.asyncio_event_loop
 class TestUsesCustomEventLoopPolicy:
     @pytest.mark.asyncio
     async def test_uses_custom_event_loop_policy(self):

pytest_asyncio/plugin.py

@@ -6,6 +6,7 @@
 import inspect
 import socket
 import warnings
+from asyncio import AbstractEventLoopPolicy
 from textwrap import dedent
 from typing import (
     Any,
@@ -553,12 +554,6 @@ 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())
-        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.
@@ -573,14 +568,12 @@ def pytest_collectstart(collector: pytest.Collector):
         @pytest.fixture(
             scope="class" if isinstance(collector, pytest.Class) else "module",
             name=event_loop_fixture_id,
-            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
-            request,
+            event_loop_policy,
         ) -> Iterator[asyncio.AbstractEventLoop]:
-            new_loop_policy = request.param
+            new_loop_policy = event_loop_policy
             old_loop_policy = asyncio.get_event_loop_policy()
             old_loop = asyncio.get_event_loop()
             asyncio.set_event_loop_policy(new_loop_policy)
@@ -675,6 +668,7 @@ def pytest_fixture_setup(
         _add_finalizers(
             fixturedef,
             _close_event_loop,
+            _restore_event_loop_policy(asyncio.get_event_loop_policy()),
             _provide_clean_event_loop,
         )
         outcome = yield
@@ -749,6 +743,23 @@ def _close_event_loop() -> None:
         loop.close()
 
 
+def _restore_event_loop_policy(previous_policy) -> Callable[[], None]:
+    def _restore_policy():
+        # Close any event loop associated with the old loop policy
+        # to avoid ResourceWarnings in the _provide_clean_event_loop finalizer
+        try:
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore", DeprecationWarning)
+                loop = previous_policy.get_event_loop()
+        except RuntimeError:
+            loop = None
+        if loop:
+            loop.close()
+        asyncio.set_event_loop_policy(previous_policy)
+
+    return _restore_policy
+
+
 def _provide_clean_event_loop() -> None:
     # At this point, the event loop for the current thread is closed.
     # When a user calls asyncio.get_event_loop(), they will get a closed loop.
@@ -856,6 +867,8 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
 @pytest.fixture
 def event_loop(request: FixtureRequest) -> Iterator[asyncio.AbstractEventLoop]:
     """Create an instance of the default event loop for each test case."""
+    new_loop_policy = request.getfixturevalue(event_loop_policy.__name__)
+    asyncio.set_event_loop_policy(new_loop_policy)
     loop = asyncio.get_event_loop_policy().new_event_loop()
     # Add a magic value to the event loop, so pytest-asyncio can determine if the
     # event_loop fixture was overridden. Other implementations of event_loop don't
@@ -867,6 +880,12 @@ def event_loop(request: FixtureRequest) -> Iterator[asyncio.AbstractEventLoop]:
     loop.close()
 
 
+@pytest.fixture(scope="session", autouse=True)
+def event_loop_policy() -> AbstractEventLoopPolicy:
+    """Return an instance of the policy used to create asyncio event loops."""
+    return asyncio.get_event_loop_policy()
+
+
 def _unused_port(socket_type: int) -> int:
     """Find an unused localhost port from 1024-65535 and return it."""
     with contextlib.closing(socket.socket(type=socket_type)) as sock:

tests/markers/test_class_marker.py

@@ -140,8 +140,12 @@ def test_asyncio_event_loop_mark_allows_specifying_the_loop_policy(
             class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
                 pass
 
-            @pytest.mark.asyncio_event_loop(policy=CustomEventLoopPolicy())
-            class TestUsesCustomEventLoopPolicy:
+            @pytest.mark.asyncio_event_loop
+            class TestUsesCustomEventLoop:
+
+                @pytest.fixture(scope="class")
+                def event_loop_policy(self):
+                    return CustomEventLoopPolicy()
 
                 @pytest.mark.asyncio
                 async def test_uses_custom_event_loop_policy(self):
@@ -173,15 +177,18 @@ def test_asyncio_event_loop_mark_allows_specifying_multiple_loop_policies(
 
             import pytest
 
-            @pytest.mark.asyncio_event_loop(
-                policy=[
+            @pytest.fixture(
+                params=[
                     asyncio.DefaultEventLoopPolicy(),
                     asyncio.DefaultEventLoopPolicy(),
                 ]
             )
+            def event_loop_policy(request):
+                return request.param
+
             class TestWithDifferentLoopPolicies:
                 @pytest.mark.asyncio
-                async def test_parametrized_loop(self):
+                async def test_parametrized_loop(self, request):
                     pass
             """
         )