Implement event loop factory hook

Author: tjkuson

changelog.d/1164.added.rst

@@ -0,0 +1 @@
+Added the ``pytest_asyncio_loop_factories`` hook to parametrize asyncio tests with custom event loop factories.

docs/how-to-guides/custom_loop_factory.rst

@@ -0,0 +1,44 @@
+==================================================
+How to use custom event loop factories for tests
+==================================================
+
+pytest-asyncio can run asynchronous tests with custom event loop factories by defining a ``pytest_asyncio_loop_factories`` hook in ``conftest.py``. The hook returns the factories to use for the current test item:
+
+.. code-block:: python
+
+   import asyncio
+
+   import pytest
+
+
+   class CustomEventLoop(asyncio.SelectorEventLoop):
+       pass
+
+
+   def pytest_asyncio_loop_factories(config, item):
+       return [CustomEventLoop]
+
+When multiple factories are returned, each asynchronous test is run once per factory. Synchronous tests are not parametrized. The configured loop scope still determines how long each event loop instance is kept alive.
+
+Factories should be callables without required parameters and should return an ``asyncio.AbstractEventLoop`` instance. The hook must return a non-empty sequence for every asyncio test.
+
+To select different factories for specific tests, you can inspect ``item``:
+
+.. code-block:: python
+
+   import asyncio
+
+   import uvloop
+
+
+   def pytest_asyncio_loop_factories(config, item):
+       if item.get_closest_marker("uvloop"):
+           return [uvloop.new_event_loop]
+       else:
+           return [asyncio.new_event_loop]
+
+Factory selection can vary per test item, regardless of loop scope. In other words, with ``module``/``package``/``session`` loop scopes you can still choose different factories for different tests by inspecting ``item``.
+
+.. note::
+
+   When the hook is defined, async tests are parametrized, so factory names are appended to test IDs. For example, a test ``test_example`` with factory ``CustomEventLoop`` will appear as ``test_example[CustomEventLoop]`` in the test output.

docs/how-to-guides/index.rst

@@ -10,6 +10,7 @@ How-To Guides
   change_fixture_loop
   change_default_fixture_loop
   change_default_test_loop
+  custom_loop_factory
   run_class_tests_in_same_loop
   run_module_tests_in_same_loop
   run_package_tests_in_same_loop

docs/how-to-guides/uvloop.rst

@@ -2,8 +2,29 @@
 How to test with uvloop
 =======================
 
-Redefining 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:*
+Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that returns ``uvloop.new_event_loop`` as a loop factory:
+
+.. code-block:: python
+
+    import uvloop
+
+
+    def pytest_asyncio_loop_factories(config, item):
+        return [uvloop.new_event_loop]
+
+.. seealso::
+
+   :doc:`custom_loop_factory`
+      More details on the ``pytest_asyncio_loop_factories`` hook, including per-test factory selection and multiple factory parametrization.
+
+Using the event_loop_policy fixture
+------------------------------------
+
+.. note::
+
+   ``asyncio.AbstractEventLoopPolicy`` is deprecated as of Python 3.14 (removal planned for 3.16), and ``uvloop.EventLoopPolicy`` will be removed alongside it. Prefer the hook approach above.
+
+For older versions of Python and uvloop, you can override the *event_loop_policy* fixture in your *conftest.py:*
 
 .. code-block:: python
 

pytest_asyncio/plugin.py

@@ -27,6 +27,7 @@
     Any,
     Literal,
     ParamSpec,
+    TypeAlias,
     TypeVar,
     overload,
 )
@@ -63,6 +64,7 @@
 _R = TypeVar("_R", bound=Awaitable[Any] | AsyncIterator[Any])
 _P = ParamSpec("_P")
 FixtureFunction = Callable[_P, _R]
+LoopFactory: TypeAlias = Callable[[], AbstractEventLoop]
 
 
 class PytestAsyncioError(Exception):
@@ -74,6 +76,19 @@ class Mode(str, enum.Enum):
     STRICT = "strict"
 
 
+hookspec = pluggy.HookspecMarker("pytest")
+
+
+class PytestAsyncioSpecs:
+    @hookspec
+    def pytest_asyncio_loop_factories(
+        self,
+        config: Config,
+        item: Item,
+    ) -> Iterable[LoopFactory]:
+        raise NotImplementedError  # pragma: no cover
+
+
 ASYNCIO_MODE_HELP = """\
 'auto' - for automatically handling all async functions by the plugin
 'strict' - for autoprocessing disabling (useful if different async frameworks \
@@ -83,6 +98,7 @@ class Mode(str, enum.Enum):
 
 
 def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None:
+    pluginmanager.add_hookspecs(PytestAsyncioSpecs)
     group = parser.getgroup("asyncio")
     group.addoption(
         "--asyncio-mode",
@@ -219,6 +235,45 @@ def _get_asyncio_debug(config: Config) -> bool:
         return val == "true"
 
 
+def _collect_hook_loop_factories(
+    config: Config,
+    item: Item,
+) -> tuple[LoopFactory, ...] | None:
+    hook_caller = config.hook.pytest_asyncio_loop_factories
+    hook_impls = hook_caller.get_hookimpls()
+    if not hook_impls:
+        return None
+    if len(hook_impls) > 1:
+        msg = (
+            "Multiple pytest_asyncio_loop_factories implementations found; please"
+            " provide a single hook implementation."
+        )
+        raise pytest.UsageError(msg)
+
+    results: list[Iterable[LoopFactory] | None] = hook_caller(config=config, item=item)
+    msg = "pytest_asyncio_loop_factories must return a non-empty sequence of callables."
+    if not results:
+        raise pytest.UsageError(msg)
+    result = results[0]
+    if result is None or not isinstance(result, Sequence):
+        raise pytest.UsageError(msg)
+    # Copy into an immutable snapshot so later mutations of the hook's
+    # original container do not affect stash state or parametrization.
+    factories = tuple(result)
+    if not factories or any(not callable(factory) for factory in factories):
+        raise pytest.UsageError(msg)
+    return factories
+
+
+def _get_item_loop_scope(item: Item, config: Config) -> _ScopeName:
+    marker = item.get_closest_marker("asyncio")
+    default_loop_scope = _get_default_test_loop_scope(config)
+    if marker is None:
+        return default_loop_scope
+    else:
+        return _get_marked_loop_scope(marker, default_loop_scope)
+
+
 _DEFAULT_FIXTURE_LOOP_SCOPE_UNSET = """\
 The configuration option "asyncio_default_fixture_loop_scope" is unset.
 The event loop scope for asynchronous fixtures will default to the "fixture" caching \
@@ -611,6 +666,27 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass(
     hook_result.force_result(updated_node_collection)
 
 
+@pytest.hookimpl(tryfirst=True)
+def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
+    if _get_asyncio_mode(
+        metafunc.config
+    ) == Mode.STRICT and not metafunc.definition.get_closest_marker("asyncio"):
+        return
+    if PytestAsyncioFunction.item_subclass_for(metafunc.definition) is None:
+        return
+    hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition)
+    if hook_factories is None:
+        return
+    metafunc.fixturenames.append("asyncio_loop_factory")
+    loop_scope = _get_item_loop_scope(metafunc.definition, metafunc.config)
+    metafunc.parametrize(
+        "asyncio_loop_factory",
+        hook_factories,
+        indirect=True,
+        scope=loop_scope,
+    )
+
+
 @contextlib.contextmanager
 def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
     old_loop_policy = _get_event_loop_policy()
@@ -798,12 +874,16 @@ def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable:
     )
     def _scoped_runner(
         event_loop_policy,
+        asyncio_loop_factory,
         request: FixtureRequest,
     ) -> Iterator[Runner]:
         new_loop_policy = event_loop_policy
         debug_mode = _get_asyncio_debug(request.config)
         with _temporary_event_loop_policy(new_loop_policy):
-            runner = Runner(debug=debug_mode).__enter__()
+            runner = Runner(
+                debug=debug_mode,
+                loop_factory=asyncio_loop_factory,
+            ).__enter__()
             try:
                 yield runner
             except Exception as e:
@@ -830,6 +910,11 @@ def _scoped_runner(
     )
 
 
+@pytest.fixture(scope="session")
+def asyncio_loop_factory(request: FixtureRequest) -> LoopFactory | None:
+    return getattr(request, "param", None)
+
+
 @pytest.fixture(scope="session", autouse=True)
 def event_loop_policy() -> AbstractEventLoopPolicy:
     """Return an instance of the policy used to create asyncio event loops."""