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..7225a545 --- /dev/null +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -0,0 +1,46 @@ +================================================== +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. + +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 + + 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..a92fc23a 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] | None: + 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,38 @@ 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 + 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) + 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 +659,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.__name__) + loop_scope = _get_item_loop_scope(metafunc.definition, metafunc.config) + metafunc.parametrize( + _asyncio_loop_factory.__name__, + 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 +867,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 +903,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..3cbe02d3 --- /dev/null +++ b/tests/test_loop_factory_parametrization.py @@ -0,0 +1,395 @@ +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_nested_conftest_multiple_hook_implementations_are_allowed( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class RootCustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return [RootCustomEventLoop] + """)) + subdir = pytester.mkdir("subtests") + subdir.joinpath("conftest.py").write_text( + dedent("""\ + import asyncio + + class SubCustomEventLoop(asyncio.SelectorEventLoop): + pass + + 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_uses_root_loop(): + loop_name = type(asyncio.get_running_loop()).__name__ + assert loop_name in ("RootCustomEventLoop", "SubCustomEventLoop") + """), + ) + 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: + 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)