From d3c835d14f52c6d86d164e149c79ec006f68e376 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sat, 7 Mar 2026 16:27:01 +0000 Subject: [PATCH 1/4] Implement event loop factory hook --- changelog.d/1164.added.rst | 1 + docs/how-to-guides/custom_loop_factory.rst | 44 +++ docs/how-to-guides/index.rst | 1 + docs/how-to-guides/uvloop.rst | 25 +- pytest_asyncio/plugin.py | 87 ++++- tests/test_loop_factory_parametrization.py | 380 +++++++++++++++++++++ 6 files changed, 535 insertions(+), 3 deletions(-) create mode 100644 changelog.d/1164.added.rst create mode 100644 docs/how-to-guides/custom_loop_factory.rst create mode 100644 tests/test_loop_factory_parametrization.py diff --git a/changelog.d/1164.added.rst b/changelog.d/1164.added.rst new file mode 100644 index 00000000..eb16c24c --- /dev/null +++ b/changelog.d/1164.added.rst @@ -0,0 +1 @@ +Added the ``pytest_asyncio_loop_factories`` hook to parametrize asyncio tests with custom event loop factories. diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst new file mode 100644 index 00000000..33a9d38e --- /dev/null +++ b/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. diff --git a/docs/how-to-guides/index.rst b/docs/how-to-guides/index.rst index 2dadc881..9f38b4f0 100644 --- a/docs/how-to-guides/index.rst +++ b/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 diff --git a/docs/how-to-guides/uvloop.rst b/docs/how-to-guides/uvloop.rst index a796bea7..83ca3cac 100644 --- a/docs/how-to-guides/uvloop.rst +++ b/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 diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 28c97cc9..7150d1d3 100644 --- a/pytest_asyncio/plugin.py +++ b/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.""" diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py new file mode 100644 index 00000000..ffe72fdf --- /dev/null +++ b/tests/test_loop_factory_parametrization.py @@ -0,0 +1,380 @@ +from __future__ import annotations + +from textwrap import dedent + +import pytest +from pytest import Pytester + + +def test_hook_factories_apply_to_async_tests(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoop] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_uses_custom_loop(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop" + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=1) + + +def test_hook_factories_parametrize_async_tests(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoopA(asyncio.SelectorEventLoop): + pass + + class CustomEventLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoopA, CustomEventLoopB] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_runs_once_per_factory(): + loop_name = type(asyncio.get_running_loop()).__name__ + assert loop_name in ("CustomEventLoopA", "CustomEventLoopB") + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_hook_factories_apply_to_async_fixtures(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoop] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture + async def loop_fixture(): + return asyncio.get_running_loop() + + @pytest.mark.asyncio + async def test_fixture_uses_custom_loop(loop_fixture): + assert type(loop_fixture).__name__ == "CustomEventLoop" + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop" + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=1) + + +def test_sync_tests_are_not_parametrized(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoopA(asyncio.SelectorEventLoop): + pass + + class CustomEventLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoopA, CustomEventLoopB] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + def test_sync(request): + assert "asyncio_loop_factory" not in request.fixturenames + + @pytest.mark.asyncio + async def test_async(request): + assert "asyncio_loop_factory" in request.fixturenames + loop_name = type(asyncio.get_running_loop()).__name__ + assert loop_name in ("CustomEventLoopA", "CustomEventLoopB") + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=3) + + +@pytest.mark.parametrize( + "hook_body", + ( + "return []", + "return (factory for factory in [CustomEventLoop])", + "return [CustomEventLoop, 1]", + "return None", + ), +) +def test_hook_requires_non_empty_sequence_of_callables( + pytester: Pytester, + hook_body: str, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent(f"""\ + import asyncio + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + {hook_body} + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_async(): + assert True + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(errors=1) + result.stdout.fnmatch_lines( + ["*pytest_asyncio_loop_factories must return a non-empty sequence*"] + ) + + +def test_hook_rejects_multiple_hook_implementations(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + pytest_plugins = ("extra_loop_factory_plugin",) + + class CustomEventLoopA(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoopA] + """)) + pytester.makepyfile( + extra_loop_factory_plugin=dedent("""\ + import asyncio + + class CustomEventLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoopB] + """), + test_hooks=dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_async(): + assert True + """), + ) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(errors=1) + result.stdout.fnmatch_lines( + ["*Multiple pytest_asyncio_loop_factories implementations found*"] + ) + + +def test_hook_accepts_tuple_return(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return (CustomEventLoop,) + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_uses_custom_loop(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop" + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=1) + + +@pytest.mark.parametrize("default_test_loop_scope", ("function", "module")) +def test_hook_factories_can_vary_per_test_with_default_loop_scope( + pytester: Pytester, + default_test_loop_scope: str, +) -> None: + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + f"asyncio_default_test_loop_scope = {default_test_loop_scope}" + ) + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoopA(asyncio.SelectorEventLoop): + pass + + class CustomEventLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + if item.name.endswith("a"): + return [CustomEventLoopA] + else: + return [CustomEventLoopB] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_a(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoopA" + + @pytest.mark.asyncio + async def test_b(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoopB" + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_hook_factories_can_vary_per_test_with_session_scope_across_modules( + pytester: Pytester, +) -> None: + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + "asyncio_default_test_loop_scope = session" + ) + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoopA(asyncio.SelectorEventLoop): + pass + + class CustomEventLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + if "test_a.py::" in item.nodeid: + return [CustomEventLoopA] + return [CustomEventLoopB] + """)) + pytester.makepyfile( + test_a=dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_a(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoopA" + """), + test_b=dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_b(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoopB" + """), + ) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_hook_factories_work_in_auto_mode(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [CustomEventLoop] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + + pytest_plugins = "pytest_asyncio" + + async def test_uses_custom_loop(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop" + """)) + result = pytester.runpytest("--asyncio-mode=auto") + result.assert_outcomes(passed=1) + + +def test_function_loop_scope_allows_per_test_factories_with_session_default( + pytester: Pytester, +) -> None: + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + "asyncio_default_test_loop_scope = session" + ) + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomEventLoopA(asyncio.SelectorEventLoop): + pass + + class CustomEventLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + if item.name.endswith("a"): + return [CustomEventLoopA] + else: + return [CustomEventLoopB] + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio(loop_scope="function") + async def test_a(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoopA" + + @pytest.mark.asyncio(loop_scope="function") + async def test_b(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoopB" + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) From 4de8c94defc813e65cb03333e5992fa72c2faf0b Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sun, 8 Mar 2026 14:34:11 +0000 Subject: [PATCH 2/4] Allow multiple hook implementations --- docs/how-to-guides/custom_loop_factory.rst | 2 + pytest_asyncio/plugin.py | 11 +---- tests/test_loop_factory_parametrization.py | 55 ++++++++++++++-------- 3 files changed, 39 insertions(+), 29 deletions(-) diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst index 33a9d38e..7225a545 100644 --- a/docs/how-to-guides/custom_loop_factory.rst +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -22,6 +22,8 @@ When multiple factories are returned, each asynchronous test is run once per fac 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. +When multiple ``pytest_asyncio_loop_factories`` implementations are present, pytest-asyncio uses the first non -``None`` result in pytest's normal hook dispatch order. + To select different factories for specific tests, you can inspect ``item``: .. code-block:: python diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 7150d1d3..db9556ba 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -85,7 +85,7 @@ def pytest_asyncio_loop_factories( self, config: Config, item: Item, - ) -> Iterable[LoopFactory]: + ) -> Iterable[LoopFactory] | None: raise NotImplementedError # pragma: no cover @@ -240,15 +240,8 @@ def _collect_hook_loop_factories( item: Item, ) -> tuple[LoopFactory, ...] | None: hook_caller = config.hook.pytest_asyncio_loop_factories - hook_impls = hook_caller.get_hookimpls() - if not hook_impls: + if not hook_caller.get_hookimpls(): 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." diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index ffe72fdf..75eb2690 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -163,44 +163,59 @@ async def test_async(): ) -def test_hook_rejects_multiple_hook_implementations(pytester: Pytester) -> None: +def test_nested_conftest_multiple_hook_implementations_are_allowed( + pytester: Pytester, +) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio - pytest_plugins = ("extra_loop_factory_plugin",) - - class CustomEventLoopA(asyncio.SelectorEventLoop): + class RootCustomEventLoop(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoopA] + return [RootCustomEventLoop] """)) - pytester.makepyfile( - extra_loop_factory_plugin=dedent("""\ - import asyncio + subdir = pytester.mkdir("subtests") + subdir.joinpath("conftest.py").write_text( + dedent("""\ + import asyncio - class CustomEventLoopB(asyncio.SelectorEventLoop): - pass + class SubCustomEventLoop(asyncio.SelectorEventLoop): + pass - def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoopB] - """), - test_hooks=dedent("""\ + def pytest_asyncio_loop_factories(config, item): + return [SubCustomEventLoop] + """), + ) + pytester.makepyfile( + test_root=dedent("""\ + import asyncio import pytest pytest_plugins = "pytest_asyncio" @pytest.mark.asyncio - async def test_async(): - assert True + async def test_uses_root_loop(): + loop_name = type(asyncio.get_running_loop()).__name__ + assert loop_name in ("RootCustomEventLoop", "SubCustomEventLoop") """), ) - result = pytester.runpytest("--asyncio-mode=strict") - result.assert_outcomes(errors=1) - result.stdout.fnmatch_lines( - ["*Multiple pytest_asyncio_loop_factories implementations found*"] + subdir.joinpath("test_sub.py").write_text( + dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_uses_sub_loop(): + loop_name = type(asyncio.get_running_loop()).__name__ + assert loop_name in ("RootCustomEventLoop", "SubCustomEventLoop") + """), ) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) def test_hook_accepts_tuple_return(pytester: Pytester) -> None: From 4971ae513cf2b041d65fb120bf60fba68d4eb548 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sun, 8 Mar 2026 16:33:39 +0000 Subject: [PATCH 3/4] Rename fixture and use symbolic references --- pytest_asyncio/plugin.py | 10 +++++----- tests/test_loop_factory_parametrization.py | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index db9556ba..b8e36f7c 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -670,10 +670,10 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition) if hook_factories is None: return - metafunc.fixturenames.append("asyncio_loop_factory") + metafunc.fixturenames.append(_asyncio_loop_factory.name) loop_scope = _get_item_loop_scope(metafunc.definition, metafunc.config) metafunc.parametrize( - "asyncio_loop_factory", + _asyncio_loop_factory.name, hook_factories, indirect=True, scope=loop_scope, @@ -867,7 +867,7 @@ def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable: ) def _scoped_runner( event_loop_policy, - asyncio_loop_factory, + _asyncio_loop_factory, request: FixtureRequest, ) -> Iterator[Runner]: new_loop_policy = event_loop_policy @@ -875,7 +875,7 @@ def _scoped_runner( with _temporary_event_loop_policy(new_loop_policy): runner = Runner( debug=debug_mode, - loop_factory=asyncio_loop_factory, + loop_factory=_asyncio_loop_factory, ).__enter__() try: yield runner @@ -904,7 +904,7 @@ def _scoped_runner( @pytest.fixture(scope="session") -def asyncio_loop_factory(request: FixtureRequest) -> LoopFactory | None: +def _asyncio_loop_factory(request: FixtureRequest) -> LoopFactory | None: return getattr(request, "param", None) diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index 75eb2690..3cbe02d3 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -112,11 +112,11 @@ def pytest_asyncio_loop_factories(config, item): pytest_plugins = "pytest_asyncio" def test_sync(request): - assert "asyncio_loop_factory" not in request.fixturenames + assert "_asyncio_loop_factory" not in request.fixturenames @pytest.mark.asyncio async def test_async(request): - assert "asyncio_loop_factory" in request.fixturenames + assert "_asyncio_loop_factory" in request.fixturenames loop_name = type(asyncio.get_running_loop()).__name__ assert loop_name in ("CustomEventLoopA", "CustomEventLoopB") """)) From 8f3d0f4a42685d2cb8b70ee42d357e8d7930dfba Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sun, 8 Mar 2026 19:13:13 +0000 Subject: [PATCH 4/4] Use __name__ instead of name The name attribute isn't present on older pytest versions. --- pytest_asyncio/plugin.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index b8e36f7c..a92fc23a 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -670,10 +670,10 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition) if hook_factories is None: return - metafunc.fixturenames.append(_asyncio_loop_factory.name) + metafunc.fixturenames.append(_asyncio_loop_factory.__name__) loop_scope = _get_item_loop_scope(metafunc.definition, metafunc.config) metafunc.parametrize( - _asyncio_loop_factory.name, + _asyncio_loop_factory.__name__, hook_factories, indirect=True, scope=loop_scope,