From d3c835d14f52c6d86d164e149c79ec006f68e376 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sat, 7 Mar 2026 16:27:01 +0000 Subject: [PATCH 01/11] 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 02/11] 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 03/11] 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 04/11] 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, From 506eb278b57f2345c9a2c6c0c3062ba248d4adea Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Mon, 9 Mar 2026 00:16:13 +0000 Subject: [PATCH 05/11] Use mapping --- changelog.d/1164.added.rst | 2 + docs/how-to-guides/custom_loop_factory.rst | 36 +-- docs/how-to-guides/uvloop.rst | 8 +- pytest_asyncio/plugin.py | 163 ++++++++---- tests/markers/test_invalid_arguments.py | 39 ++- tests/test_loop_factory_parametrization.py | 289 ++++++++++++++++++--- 6 files changed, 429 insertions(+), 108 deletions(-) diff --git a/changelog.d/1164.added.rst b/changelog.d/1164.added.rst index eb16c24c..ccf2988a 100644 --- a/changelog.d/1164.added.rst +++ b/changelog.d/1164.added.rst @@ -1 +1,3 @@ Added the ``pytest_asyncio_loop_factories`` hook to parametrize asyncio tests with custom event loop factories. + +The hook now returns a mapping of factory names to loop factories, and ``pytest.mark.asyncio(loop_factories=[...])`` can be used to select a subset of configured factories per test. diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst index 7225a545..c52363fc 100644 --- a/docs/how-to-guides/custom_loop_factory.rst +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -1,8 +1,8 @@ -================================================== +================================================ 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: +pytest-asyncio can run asynchronous tests with custom event loop factories by implementing ``pytest_asyncio_loop_factories`` in ``conftest.py``. The hook returns a mapping from factory names to loop factory callables: .. code-block:: python @@ -16,31 +16,31 @@ pytest-asyncio can run asynchronous tests with custom event loop factories by de def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoop] + return { + "stdlib": asyncio.new_event_loop, + "custom": 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. +By default, each asynchronous test is run once per configured 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. +Factories should be callables without required parameters and should return an ``asyncio.AbstractEventLoop`` instance. The effective hook result must be a non-empty mapping of non-empty string names to callables. -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``: +To run a test with only a subset of configured factories, use the ``loop_factories`` argument of ``pytest.mark.asyncio``: .. code-block:: python - import asyncio + import pytest - import uvloop + @pytest.mark.asyncio(loop_factories=["custom"]) + async def test_only_with_custom_event_loop(): + pass - 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``. +If ``loop_factories`` contains unknown names, pytest-asyncio raises a ``pytest.UsageError`` during collection. + +When multiple ``pytest_asyncio_loop_factories`` implementations are present, pytest-asyncio uses the first non-``None`` result in pytest's hook dispatch order. .. 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. + When the hook is defined, async tests are parametrized via ``pytest.metafunc.parametrize``, and mapping keys are used as test IDs. For example, a test ``test_example`` with an event loop factory key ``foo`` will appear as ``test_example[foo]`` in test output. diff --git a/docs/how-to-guides/uvloop.rst b/docs/how-to-guides/uvloop.rst index 83ca3cac..99afd238 100644 --- a/docs/how-to-guides/uvloop.rst +++ b/docs/how-to-guides/uvloop.rst @@ -2,7 +2,7 @@ How to test with uvloop ======================= -Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that returns ``uvloop.new_event_loop`` as a loop factory: +Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that maps factory names to loop factories: .. code-block:: python @@ -10,7 +10,9 @@ Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that retur def pytest_asyncio_loop_factories(config, item): - return [uvloop.new_event_loop] + return { + "uvloop": uvloop.new_event_loop, + } .. seealso:: @@ -18,7 +20,7 @@ Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that retur 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:: diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index a92fc23a..e12dd89a 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -20,6 +20,7 @@ Generator, Iterable, Iterator, + Mapping, Sequence, ) from types import AsyncGeneratorType, CoroutineType @@ -80,12 +81,12 @@ class Mode(str, enum.Enum): class PytestAsyncioSpecs: - @hookspec + @hookspec(firstresult=True) def pytest_asyncio_loop_factories( self, config: Config, item: Item, - ) -> Iterable[LoopFactory] | None: + ) -> Mapping[str, LoopFactory] | None: raise NotImplementedError # pragma: no cover @@ -235,38 +236,34 @@ def _get_asyncio_debug(config: Config) -> bool: return val == "true" +_INVALID_LOOP_FACTORIES = ( + "pytest_asyncio_loop_factories must return a non-empty mapping of factory names " + "to callables." +) + + def _collect_hook_loop_factories( config: Config, item: Item, -) -> tuple[LoopFactory, ...] | None: +) -> dict[str, LoopFactory] | None: hook_caller = config.hook.pytest_asyncio_loop_factories if not hook_caller.get_hookimpls(): return None - 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) + result = hook_caller(config=config, item=item) + if result is None or not isinstance(result, Mapping): + raise pytest.UsageError(_INVALID_LOOP_FACTORIES) + # Copy into an isolated snapshot so later mutations of the hook's + # original container do not affect parametrization. + factories = dict(result) + if not factories or any( + not isinstance(name, str) or not name or not callable(factory) + for name, factory in factories.items() + ): + raise pytest.UsageError(_INVALID_LOOP_FACTORIES) 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 \ @@ -526,10 +523,14 @@ def _loop_scope(self) -> _ScopeName: marker is present, the the loop scope is determined by the configuration value of `asyncio_default_test_loop_scope`, instead. """ + default_loop_scope = _get_default_test_loop_scope(self.config) marker = self.get_closest_marker("asyncio") assert marker is not None - default_loop_scope = _get_default_test_loop_scope(self.config) - return _get_marked_loop_scope(marker, default_loop_scope) + loop_scope = marker.kwargs.get("loop_scope") or marker.kwargs.get("scope") + if loop_scope is None: + return default_loop_scope + else: + return loop_scope @property def _synchronization_target_attr(self) -> tuple[object, str]: @@ -618,6 +619,16 @@ def _synchronization_target_attr(self) -> tuple[object, str]: return self.obj.hypothesis, "inner_test" +def _resolve_asyncio_marker(item: Function) -> Mark | None: + marker = item.get_closest_marker("asyncio") + if marker is not None: + return marker + if _get_asyncio_mode(item.config) == Mode.AUTO: + item.add_marker("asyncio") + return item.get_closest_marker("asyncio") + return None + + # The function name needs to start with "pytest_" # see https://github.com/pytest-dev/pytest/issues/11307 @pytest.hookimpl(specname="pytest_pycollect_makeitem", hookwrapper=True) @@ -648,33 +659,63 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass( updated_item = node if isinstance(node, Function): specialized_item_class = PytestAsyncioFunction.item_subclass_for(node) - if specialized_item_class: - if _get_asyncio_mode( - node.config - ) == Mode.AUTO and not node.get_closest_marker("asyncio"): - node.add_marker("asyncio") - if node.get_closest_marker("asyncio"): - updated_item = specialized_item_class._from_function(node) + if ( + specialized_item_class is not None + and _resolve_asyncio_marker(node) is not None + ): + updated_item = specialized_item_class._from_function(node) updated_node_collection.append(updated_item) 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"): + specialized_item_class = PytestAsyncioFunction.item_subclass_for( + metafunc.definition + ) + if specialized_item_class is None: return - if PytestAsyncioFunction.item_subclass_for(metafunc.definition) is None: + + asyncio_marker = _resolve_asyncio_marker(metafunc.definition) + if asyncio_marker is None: return + marker_loop_scope, marker_selected_factory_names = _parse_asyncio_marker( + asyncio_marker + ) + hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition) if hook_factories is None: + if marker_selected_factory_names is not None: + raise pytest.UsageError( + "mark.asyncio 'loop_factories' requires at least one " + "pytest_asyncio_loop_factories hook implementation." + ) return + + if marker_selected_factory_names is None: + effective_factories = hook_factories + else: + missing_factory_names = tuple( + name for name in marker_selected_factory_names if name not in hook_factories + ) + if missing_factory_names: + msg = ( + f"Unknown factory name(s) {missing_factory_names}." + f" Available names: {', '.join(hook_factories)}." + ) + raise pytest.UsageError(msg) + # Build the mapping in marker order to preserve explicit user + # selection order in parametrization. + effective_factories = { + name: hook_factories[name] for name in marker_selected_factory_names + } metafunc.fixturenames.append(_asyncio_loop_factory.__name__) - loop_scope = _get_item_loop_scope(metafunc.definition, metafunc.config) + default_loop_scope = _get_default_test_loop_scope(metafunc.config) + loop_scope = marker_loop_scope or default_loop_scope metafunc.parametrize( _asyncio_loop_factory.__name__, - hook_factories, + effective_factories.values(), + ids=effective_factories.keys(), indirect=True, scope=loop_scope, ) @@ -823,15 +864,16 @@ def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None: Please use the "loop_scope" argument instead. """ +_INVALID_LOOP_FACTORIES_KWARG = """\ +mark.asyncio 'loop_factories' must be a non-empty sequence of strings. +""" + -def _get_marked_loop_scope( - asyncio_marker: Mark, default_loop_scope: _ScopeName -) -> _ScopeName: +def _parse_asyncio_marker( + asyncio_marker: Mark, +) -> tuple[_ScopeName | None, Sequence[str] | None]: assert asyncio_marker.name == "asyncio" - if asyncio_marker.args or ( - asyncio_marker.kwargs and set(asyncio_marker.kwargs) - {"loop_scope", "scope"} - ): - raise ValueError("mark.asyncio accepts only a keyword argument 'loop_scope'.") + _validate_asyncio_marker(asyncio_marker) if "scope" in asyncio_marker.kwargs: if "loop_scope" in asyncio_marker.kwargs: raise pytest.UsageError(_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR) @@ -839,10 +881,31 @@ def _get_marked_loop_scope( scope = asyncio_marker.kwargs.get("loop_scope") or asyncio_marker.kwargs.get( "scope" ) - if scope is None: - scope = default_loop_scope - assert scope in {"function", "class", "module", "package", "session"} - return scope + if scope is not None: + assert scope in {"function", "class", "module", "package", "session"} + marker_value = asyncio_marker.kwargs.get("loop_factories") + if marker_value is None: + return scope, None + if isinstance(marker_value, str) or not isinstance(marker_value, Sequence): + raise ValueError(_INVALID_LOOP_FACTORIES_KWARG) + if not marker_value or any( + not isinstance(factory_name, str) or not factory_name + for factory_name in marker_value + ): + raise ValueError(_INVALID_LOOP_FACTORIES_KWARG) + return scope, marker_value + + +def _validate_asyncio_marker(asyncio_marker: Mark) -> None: + if asyncio_marker.args or ( + asyncio_marker.kwargs + and set(asyncio_marker.kwargs) - {"loop_scope", "scope", "loop_factories"} + ): + msg = ( + "mark.asyncio accepts only keyword arguments 'loop_scope' and" + " 'loop_factories'." + ) + raise ValueError(msg) def _get_default_test_loop_scope(config: Config) -> Any: diff --git a/tests/markers/test_invalid_arguments.py b/tests/markers/test_invalid_arguments.py index 19aa3126..df19ecfb 100644 --- a/tests/markers/test_invalid_arguments.py +++ b/tests/markers/test_invalid_arguments.py @@ -35,7 +35,7 @@ async def test_anything(): result = pytester.runpytest("--assert=plain") result.assert_outcomes(errors=1) result.stdout.fnmatch_lines( - ["*ValueError: mark.asyncio accepts only a keyword argument*"] + ["*ValueError: mark.asyncio accepts only keyword arguments*"] ) @@ -53,7 +53,7 @@ async def test_anything(): result = pytester.runpytest("--assert=plain") result.assert_outcomes(errors=1) result.stdout.fnmatch_lines( - ["*ValueError: mark.asyncio accepts only a keyword argument 'loop_scope'*"] + ["*ValueError: mark.asyncio accepts only keyword arguments*"] ) @@ -71,5 +71,38 @@ async def test_anything(): result = pytester.runpytest("--assert=plain") result.assert_outcomes(errors=1) result.stdout.fnmatch_lines( - ["*ValueError: mark.asyncio accepts only a keyword argument*"] + ["*ValueError: mark.asyncio accepts only keyword arguments*"] + ) + + +@pytest.mark.parametrize( + "loop_factories_value", + ('"custom"', "[]", '[""]', "[1]"), +) +def test_error_when_loop_factories_marker_value_is_invalid( + pytester: pytest.Pytester, loop_factories_value: str +): + 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 {"custom": CustomEventLoop} + """)) + pytester.makepyfile(dedent(f"""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio(loop_factories={loop_factories_value}) + async def test_anything(): + pass + """)) + result = pytester.runpytest("--assert=plain") + result.assert_outcomes(errors=1) + result.stdout.fnmatch_lines( + ["*ValueError: mark.asyncio 'loop_factories' must be a non-empty sequence*"] ) diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index 3cbe02d3..11c5aadd 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -6,7 +6,7 @@ from pytest import Pytester -def test_hook_factories_apply_to_async_tests(pytester: Pytester) -> None: +def test_named_hook_factories_apply_to_async_tests(pytester: Pytester) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio @@ -15,7 +15,7 @@ class CustomEventLoop(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoop] + return {"custom": CustomEventLoop} """)) pytester.makepyfile(dedent("""\ import asyncio @@ -31,7 +31,7 @@ async def test_uses_custom_loop(): result.assert_outcomes(passed=1) -def test_hook_factories_parametrize_async_tests(pytester: Pytester) -> None: +def test_named_hook_factories_parametrize_async_tests(pytester: Pytester) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio @@ -43,7 +43,10 @@ class CustomEventLoopB(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoopA, CustomEventLoopB] + return { + "factory_a": CustomEventLoopA, + "factory_b": CustomEventLoopB, + } """)) pytester.makepyfile(dedent("""\ import asyncio @@ -60,7 +63,38 @@ async def test_runs_once_per_factory(): result.assert_outcomes(passed=2) -def test_hook_factories_apply_to_async_fixtures(pytester: Pytester) -> None: +def test_named_hook_factories_use_mapping_keys_as_test_ids( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + def pytest_asyncio_loop_factories(config, item): + return { + "factory_a": asyncio.new_event_loop, + "factory_b": asyncio.new_event_loop, + } + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_runs_once_per_factory(): + assert True + """)) + result = pytester.runpytest("--asyncio-mode=strict", "--collect-only", "-q") + result.stdout.fnmatch_lines( + [ + "*test_runs_once_per_factory[[]factory_a[]]", + "*test_runs_once_per_factory[[]factory_b[]]", + ] + ) + + +def test_named_hook_factories_apply_to_async_fixtures(pytester: Pytester) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio @@ -69,7 +103,7 @@ class CustomEventLoop(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoop] + return {"custom": CustomEventLoop} """)) pytester.makepyfile(dedent("""\ import asyncio @@ -91,7 +125,7 @@ async def test_fixture_uses_custom_loop(loop_fixture): result.assert_outcomes(passed=1) -def test_sync_tests_are_not_parametrized(pytester: Pytester) -> None: +def test_sync_tests_are_not_parametrized_by_hook_factories(pytester: Pytester) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio @@ -103,7 +137,10 @@ class CustomEventLoopB(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoopA, CustomEventLoopB] + return { + "factory_a": CustomEventLoopA, + "factory_b": CustomEventLoopB, + } """)) pytester.makepyfile(dedent("""\ import asyncio @@ -127,13 +164,14 @@ async def test_async(request): @pytest.mark.parametrize( "hook_body", ( - "return []", - "return (factory for factory in [CustomEventLoop])", - "return [CustomEventLoop, 1]", "return None", + "return {}", + "return [CustomEventLoop]", + "return {'': CustomEventLoop}", + "return {'default': 1}", ), ) -def test_hook_requires_non_empty_sequence_of_callables( +def test_hook_requires_non_empty_mapping_of_named_callables( pytester: Pytester, hook_body: str, ) -> None: @@ -159,33 +197,159 @@ async def test_async(): 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*"] + [ + "*pytest_asyncio_loop_factories must return a non-empty mapping of " + "factory*" + ] + ) + + +def test_hook_factories_use_first_non_none_result(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile( + plugin_none=dedent("""\ + import pytest + + @pytest.hookimpl(tryfirst=True) + def pytest_asyncio_loop_factories(config, item): + return None + """), + plugin_loop=dedent("""\ + import asyncio + import pytest + + class SecondaryCustomEventLoop(asyncio.SelectorEventLoop): + pass + + @pytest.hookimpl(trylast=True) + def pytest_asyncio_loop_factories(config, item): + return {"secondary": SecondaryCustomEventLoop} + """), + test_sample=dedent("""\ + import asyncio + import pytest + + pytest_plugins = ("pytest_asyncio", "plugin_none", "plugin_loop") + + @pytest.mark.asyncio + async def test_uses_secondary_loop(): + assert ( + type(asyncio.get_running_loop()).__name__ + == "SecondaryCustomEventLoop" + ) + """), + ) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=1) + + +def test_hook_factories_short_circuit_after_first_non_none_result( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile( + plugin_first=dedent("""\ + import asyncio + import pytest + + class PrimaryCustomEventLoop(asyncio.SelectorEventLoop): + pass + + @pytest.hookimpl(tryfirst=True) + def pytest_asyncio_loop_factories(config, item): + return {"primary": PrimaryCustomEventLoop} + """), + plugin_second=dedent("""\ + import pytest + + @pytest.hookimpl(trylast=True) + def pytest_asyncio_loop_factories(config, item): + raise RuntimeError("should not be called") + """), + test_sample=dedent("""\ + import asyncio + import pytest + + pytest_plugins = ("pytest_asyncio", "plugin_first", "plugin_second") + + @pytest.mark.asyncio + async def test_uses_primary_loop(): + assert ( + type(asyncio.get_running_loop()).__name__ + == "PrimaryCustomEventLoop" + ) + """), + ) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=1) + + +def test_hook_factories_error_when_all_implementations_return_none( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile( + plugin_none_a=dedent("""\ + import pytest + + @pytest.hookimpl(tryfirst=True) + def pytest_asyncio_loop_factories(config, item): + return None + """), + plugin_none_b=dedent("""\ + import pytest + + @pytest.hookimpl(trylast=True) + def pytest_asyncio_loop_factories(config, item): + return None + """), + test_sample=dedent("""\ + import pytest + + pytest_plugins = ("pytest_asyncio", "plugin_none_a", "plugin_none_b") + + @pytest.mark.asyncio + async def test_anything(): + 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 mapping of " + "factory*" + ] ) -def test_nested_conftest_multiple_hook_implementations_are_allowed( +def test_nested_conftest_hook_implementations_respect_hook_order( pytester: Pytester, ) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio + import pytest class RootCustomEventLoop(asyncio.SelectorEventLoop): pass + @pytest.hookimpl(trylast=True) def pytest_asyncio_loop_factories(config, item): - return [RootCustomEventLoop] + return {"root": RootCustomEventLoop} """)) subdir = pytester.mkdir("subtests") subdir.joinpath("conftest.py").write_text( dedent("""\ import asyncio + import pytest class SubCustomEventLoop(asyncio.SelectorEventLoop): pass + @pytest.hookimpl(tryfirst=True) def pytest_asyncio_loop_factories(config, item): - return [SubCustomEventLoop] + return {"sub": SubCustomEventLoop} """), ) pytester.makepyfile( @@ -196,9 +360,8 @@ def pytest_asyncio_loop_factories(config, item): pytest_plugins = "pytest_asyncio" @pytest.mark.asyncio - async def test_uses_root_loop(): - loop_name = type(asyncio.get_running_loop()).__name__ - assert loop_name in ("RootCustomEventLoop", "SubCustomEventLoop") + async def test_uses_sub_loop(): + assert type(asyncio.get_running_loop()).__name__ == "SubCustomEventLoop" """), ) subdir.joinpath("test_sub.py").write_text( @@ -210,24 +373,29 @@ async def test_uses_root_loop(): @pytest.mark.asyncio async def test_uses_sub_loop(): - loop_name = type(asyncio.get_running_loop()).__name__ - assert loop_name in ("RootCustomEventLoop", "SubCustomEventLoop") + assert type(asyncio.get_running_loop()).__name__ == "SubCustomEventLoop" """), ) result = pytester.runpytest("--asyncio-mode=strict") result.assert_outcomes(passed=2) -def test_hook_accepts_tuple_return(pytester: Pytester) -> None: +def test_asyncio_marker_loop_factories_select_subset(pytester: Pytester) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio - class CustomEventLoop(asyncio.SelectorEventLoop): + class MainCustomEventLoop(asyncio.SelectorEventLoop): + pass + + class AlternativeCustomEventLoop(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return (CustomEventLoop,) + return { + "main": MainCustomEventLoop, + "alternative": AlternativeCustomEventLoop, + } """)) pytester.makepyfile(dedent("""\ import asyncio @@ -235,14 +403,66 @@ def pytest_asyncio_loop_factories(config, item): pytest_plugins = "pytest_asyncio" - @pytest.mark.asyncio - async def test_uses_custom_loop(): - assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop" + @pytest.mark.asyncio(loop_factories=["alternative"]) + async def test_runs_only_with_uvloop(): + assert ( + type(asyncio.get_running_loop()).__name__ + == "AlternativeCustomEventLoop" + ) """)) result = pytester.runpytest("--asyncio-mode=strict") result.assert_outcomes(passed=1) +def test_asyncio_marker_loop_factories_unknown_name_errors(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + def pytest_asyncio_loop_factories(config, item): + return {"root": asyncio.new_event_loop} + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio(loop_factories=["missing"]) + async def test_errors(): + assert True + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(errors=1) + result.stdout.fnmatch_lines( + [ + "*Unknown factory name(s)*Available names:*", + ] + ) + + +def test_asyncio_marker_loop_factories_without_hook_errors( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio(loop_factories=["missing"]) + async def test_errors(): + assert True + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(errors=1) + result.stdout.fnmatch_lines( + [ + "*mark.asyncio 'loop_factories' requires at least one " + "pytest_asyncio_loop_factories hook implementation.*", + ] + ) + + @pytest.mark.parametrize("default_test_loop_scope", ("function", "module")) def test_hook_factories_can_vary_per_test_with_default_loop_scope( pytester: Pytester, @@ -263,9 +483,9 @@ class CustomEventLoopB(asyncio.SelectorEventLoop): def pytest_asyncio_loop_factories(config, item): if item.name.endswith("a"): - return [CustomEventLoopA] + return {"factory_a": CustomEventLoopA} else: - return [CustomEventLoopB] + return {"factory_b": CustomEventLoopB} """)) pytester.makepyfile(dedent("""\ import asyncio @@ -303,8 +523,9 @@ class CustomEventLoopB(asyncio.SelectorEventLoop): def pytest_asyncio_loop_factories(config, item): if "test_a.py::" in item.nodeid: - return [CustomEventLoopA] - return [CustomEventLoopB] + return {"factory_a": CustomEventLoopA} + else: + return {"factory_b": CustomEventLoopB} """)) pytester.makepyfile( test_a=dedent("""\ @@ -341,7 +562,7 @@ class CustomEventLoop(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return [CustomEventLoop] + return {"custom": CustomEventLoop} """)) pytester.makepyfile(dedent("""\ import asyncio @@ -373,9 +594,9 @@ class CustomEventLoopB(asyncio.SelectorEventLoop): def pytest_asyncio_loop_factories(config, item): if item.name.endswith("a"): - return [CustomEventLoopA] + return {"factory_a": CustomEventLoopA} else: - return [CustomEventLoopB] + return {"factory_b": CustomEventLoopB} """)) pytester.makepyfile(dedent("""\ import asyncio From e7e51aa6a53fd932e3b72756edfa4286415bd24a Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sun, 15 Mar 2026 20:21:39 +0000 Subject: [PATCH 06/11] Add note about other async tests --- docs/how-to-guides/custom_loop_factory.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst index c52363fc..ffff057f 100644 --- a/docs/how-to-guides/custom_loop_factory.rst +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -21,7 +21,7 @@ pytest-asyncio can run asynchronous tests with custom event loop factories by im "custom": CustomEventLoop, } -By default, each asynchronous test is run once per configured factory. Synchronous tests are not parametrized. The configured loop scope still determines how long each event loop instance is kept alive. +By default, each pytest-asyncio test is run once per configured factory. Tests managed by other async plugins are unaffected. 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 effective hook result must be a non-empty mapping of non-empty string names to callables. From 53930458c62e95a4e475595039679402cb94a483 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Sun, 15 Mar 2026 20:36:14 +0000 Subject: [PATCH 07/11] Split up docs --- docs/how-to-guides/custom_loop_factory.rst | 24 ++----------------- docs/how-to-guides/index.rst | 1 + .../run_test_with_specific_loop_factories.rst | 14 +++++++++++ docs/reference/hooks.rst | 16 +++++++++++++ docs/reference/index.rst | 1 + docs/reference/markers/index.rst | 2 ++ 6 files changed, 36 insertions(+), 22 deletions(-) create mode 100644 docs/how-to-guides/run_test_with_specific_loop_factories.rst create mode 100644 docs/reference/hooks.rst diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst index ffff057f..6e176d95 100644 --- a/docs/how-to-guides/custom_loop_factory.rst +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -21,26 +21,6 @@ pytest-asyncio can run asynchronous tests with custom event loop factories by im "custom": CustomEventLoop, } -By default, each pytest-asyncio test is run once per configured factory. Tests managed by other async plugins are unaffected. Synchronous tests are not parametrized. The configured loop scope still determines how long each event loop instance is kept alive. +See :doc:`run_test_with_specific_loop_factories` for running tests with only a subset of configured factories. -Factories should be callables without required parameters and should return an ``asyncio.AbstractEventLoop`` instance. The effective hook result must be a non-empty mapping of non-empty string names to callables. - -To run a test with only a subset of configured factories, use the ``loop_factories`` argument of ``pytest.mark.asyncio``: - -.. code-block:: python - - import pytest - - - @pytest.mark.asyncio(loop_factories=["custom"]) - async def test_only_with_custom_event_loop(): - pass - - -If ``loop_factories`` contains unknown names, pytest-asyncio raises a ``pytest.UsageError`` during collection. - -When multiple ``pytest_asyncio_loop_factories`` implementations are present, pytest-asyncio uses the first non-``None`` result in pytest's hook dispatch order. - -.. note:: - - When the hook is defined, async tests are parametrized via ``pytest.metafunc.parametrize``, and mapping keys are used as test IDs. For example, a test ``test_example`` with an event loop factory key ``foo`` will appear as ``test_example[foo]`` in test output. +See :doc:`../reference/hooks` and :doc:`../reference/markers/index` for the hook and marker reference. diff --git a/docs/how-to-guides/index.rst b/docs/how-to-guides/index.rst index 9f38b4f0..acf0d177 100644 --- a/docs/how-to-guides/index.rst +++ b/docs/how-to-guides/index.rst @@ -11,6 +11,7 @@ How-To Guides change_default_fixture_loop change_default_test_loop custom_loop_factory + run_test_with_specific_loop_factories 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/run_test_with_specific_loop_factories.rst b/docs/how-to-guides/run_test_with_specific_loop_factories.rst new file mode 100644 index 00000000..338d28ab --- /dev/null +++ b/docs/how-to-guides/run_test_with_specific_loop_factories.rst @@ -0,0 +1,14 @@ +========================================================= +How to run a test with specific event loop factories only +========================================================= + +To run a test with only a subset of configured factories, use the ``loop_factories`` argument of ``pytest.mark.asyncio``: + +.. code-block:: python + + import pytest + + + @pytest.mark.asyncio(loop_factories=["custom"]) + async def test_only_with_custom_event_loop(): + pass diff --git a/docs/reference/hooks.rst b/docs/reference/hooks.rst new file mode 100644 index 00000000..ed025b6f --- /dev/null +++ b/docs/reference/hooks.rst @@ -0,0 +1,16 @@ +===== +Hooks +===== + +``pytest_asyncio_loop_factories`` +================================= + +This hook returns a mapping from factory name strings to event loop factory callables for the current test item. + +By default, each pytest-asyncio test is run once per configured factory. Tests managed by other async plugins are unaffected. 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 effective hook result must be a non-empty mapping of non-empty string names to callables. + +When multiple ``pytest_asyncio_loop_factories`` implementations are present, pytest-asyncio uses the first non-``None`` result in pytest's hook dispatch order. + +When the hook is defined, async tests are parametrized via ``pytest.metafunc.parametrize``, and mapping keys are used as test IDs. For example, a test ``test_example`` with an event loop factory key ``foo`` will appear as ``test_example[foo]`` in test output. diff --git a/docs/reference/index.rst b/docs/reference/index.rst index b24c6e9c..5c3095f7 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -8,6 +8,7 @@ Reference configuration fixtures/index functions + hooks markers/index decorators/index changelog diff --git a/docs/reference/markers/index.rst b/docs/reference/markers/index.rst index 7715077b..761196a8 100644 --- a/docs/reference/markers/index.rst +++ b/docs/reference/markers/index.rst @@ -36,6 +36,8 @@ Subpackages do not share the loop with their parent package. Tests marked with *session* scope share the same event loop, even if the tests exist in different packages. +The ``pytest.mark.asyncio`` marker also accepts a ``loop_factories`` keyword argument to select a subset of configured event loop factories for a test. If ``loop_factories`` contains unknown names, pytest-asyncio raises a ``pytest.UsageError`` during collection. + .. |auto mode| replace:: *auto mode* .. _auto mode: ../../concepts.html#auto-mode .. |pytestmark| replace:: ``pytestmark`` From 72646b445131b8df818223de22eae8d93d564cef Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Mon, 16 Mar 2026 00:12:14 +0000 Subject: [PATCH 08/11] ref: move line to reduce diff --- pytest_asyncio/plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index e12dd89a..88f51618 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -523,9 +523,9 @@ def _loop_scope(self) -> _ScopeName: marker is present, the the loop scope is determined by the configuration value of `asyncio_default_test_loop_scope`, instead. """ - default_loop_scope = _get_default_test_loop_scope(self.config) marker = self.get_closest_marker("asyncio") assert marker is not None + default_loop_scope = _get_default_test_loop_scope(self.config) loop_scope = marker.kwargs.get("loop_scope") or marker.kwargs.get("scope") if loop_scope is None: return default_loop_scope From 85ce9f66d60033a1db17e4f80d7252565969694f Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Tue, 17 Mar 2026 22:18:35 +0000 Subject: [PATCH 09/11] Use Item.ihook instead of Config.hook --- pytest_asyncio/plugin.py | 2 +- tests/test_loop_factory_parametrization.py | 16 +++++++--------- 2 files changed, 8 insertions(+), 10 deletions(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 88f51618..7061407d 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -246,7 +246,7 @@ def _collect_hook_loop_factories( config: Config, item: Item, ) -> dict[str, LoopFactory] | None: - hook_caller = config.hook.pytest_asyncio_loop_factories + hook_caller = item.ihook.pytest_asyncio_loop_factories if not hook_caller.get_hookimpls(): return None diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index 11c5aadd..0d26651e 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -323,31 +323,27 @@ async def test_anything(): ) -def test_nested_conftest_hook_implementations_respect_hook_order( +def test_nested_conftest_hook_respects_conftest_locality( pytester: Pytester, ) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio - import pytest class RootCustomEventLoop(asyncio.SelectorEventLoop): pass - @pytest.hookimpl(trylast=True) def pytest_asyncio_loop_factories(config, item): return {"root": RootCustomEventLoop} """)) - subdir = pytester.mkdir("subtests") + subdir = pytester.mkdir("subdir") subdir.joinpath("conftest.py").write_text( dedent("""\ import asyncio - import pytest class SubCustomEventLoop(asyncio.SelectorEventLoop): pass - @pytest.hookimpl(tryfirst=True) def pytest_asyncio_loop_factories(config, item): return {"sub": SubCustomEventLoop} """), @@ -360,8 +356,10 @@ def pytest_asyncio_loop_factories(config, item): pytest_plugins = "pytest_asyncio" @pytest.mark.asyncio - async def test_uses_sub_loop(): - assert type(asyncio.get_running_loop()).__name__ == "SubCustomEventLoop" + async def test_root_uses_root_loop(): + assert ( + type(asyncio.get_running_loop()).__name__ == "RootCustomEventLoop" + ) """), ) subdir.joinpath("test_sub.py").write_text( @@ -372,7 +370,7 @@ async def test_uses_sub_loop(): pytest_plugins = "pytest_asyncio" @pytest.mark.asyncio - async def test_uses_sub_loop(): + async def test_sub_uses_sub_loop(): assert type(asyncio.get_running_loop()).__name__ == "SubCustomEventLoop" """), ) From 128989f5f2b271555f2b7a7742bec4bf82bedc90 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Tue, 17 Mar 2026 22:25:17 +0000 Subject: [PATCH 10/11] Clean up tests --- tests/test_loop_factory_parametrization.py | 43 +--------------------- 1 file changed, 1 insertion(+), 42 deletions(-) diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index 0d26651e..f6bac235 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -149,7 +149,7 @@ def pytest_asyncio_loop_factories(config, item): pytest_plugins = "pytest_asyncio" def test_sync(request): - assert "_asyncio_loop_factory" not in request.fixturenames + assert True @pytest.mark.asyncio async def test_async(request): @@ -243,47 +243,6 @@ async def test_uses_secondary_loop(): result.assert_outcomes(passed=1) -def test_hook_factories_short_circuit_after_first_non_none_result( - pytester: Pytester, -) -> None: - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - plugin_first=dedent("""\ - import asyncio - import pytest - - class PrimaryCustomEventLoop(asyncio.SelectorEventLoop): - pass - - @pytest.hookimpl(tryfirst=True) - def pytest_asyncio_loop_factories(config, item): - return {"primary": PrimaryCustomEventLoop} - """), - plugin_second=dedent("""\ - import pytest - - @pytest.hookimpl(trylast=True) - def pytest_asyncio_loop_factories(config, item): - raise RuntimeError("should not be called") - """), - test_sample=dedent("""\ - import asyncio - import pytest - - pytest_plugins = ("pytest_asyncio", "plugin_first", "plugin_second") - - @pytest.mark.asyncio - async def test_uses_primary_loop(): - assert ( - type(asyncio.get_running_loop()).__name__ - == "PrimaryCustomEventLoop" - ) - """), - ) - result = pytester.runpytest("--asyncio-mode=strict") - result.assert_outcomes(passed=1) - - def test_hook_factories_error_when_all_implementations_return_none( pytester: Pytester, ) -> None: From 74fa976e1edd0e0bd9b564dff730f4fdb93d8e46 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Tue, 17 Mar 2026 22:46:27 +0000 Subject: [PATCH 11/11] Use triple-quoted string --- pytest_asyncio/plugin.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 7061407d..a69350bd 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -236,10 +236,10 @@ def _get_asyncio_debug(config: Config) -> bool: return val == "true" -_INVALID_LOOP_FACTORIES = ( - "pytest_asyncio_loop_factories must return a non-empty mapping of factory names " - "to callables." -) +_INVALID_LOOP_FACTORIES = """\ +pytest_asyncio_loop_factories must return a non-empty mapping of \ +factory names to callables. +""" def _collect_hook_loop_factories(