diff --git a/changelog.d/1164.added.rst b/changelog.d/1164.added.rst new file mode 100644 index 00000000..ccf2988a --- /dev/null +++ b/changelog.d/1164.added.rst @@ -0,0 +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 new file mode 100644 index 00000000..6e176d95 --- /dev/null +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -0,0 +1,26 @@ +================================================ +How to use custom event loop factories for tests +================================================ + +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 + + import asyncio + + import pytest + + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + + def pytest_asyncio_loop_factories(config, item): + return { + "stdlib": asyncio.new_event_loop, + "custom": CustomEventLoop, + } + +See :doc:`run_test_with_specific_loop_factories` for running tests with only a subset of configured factories. + +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 2dadc881..acf0d177 100644 --- a/docs/how-to-guides/index.rst +++ b/docs/how-to-guides/index.rst @@ -10,6 +10,8 @@ How-To Guides change_fixture_loop 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/how-to-guides/uvloop.rst b/docs/how-to-guides/uvloop.rst index a796bea7..99afd238 100644 --- a/docs/how-to-guides/uvloop.rst +++ b/docs/how-to-guides/uvloop.rst @@ -2,8 +2,31 @@ 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 maps factory names to loop factories: + +.. code-block:: python + + import uvloop + + + def pytest_asyncio_loop_factories(config, item): + return { + "uvloop": 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/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`` diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 28c97cc9..a69350bd 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 @@ -27,6 +28,7 @@ Any, Literal, ParamSpec, + TypeAlias, TypeVar, overload, ) @@ -63,6 +65,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 +77,19 @@ class Mode(str, enum.Enum): STRICT = "strict" +hookspec = pluggy.HookspecMarker("pytest") + + +class PytestAsyncioSpecs: + @hookspec(firstresult=True) + def pytest_asyncio_loop_factories( + self, + config: Config, + item: Item, + ) -> Mapping[str, 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 +99,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 +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, +) -> dict[str, LoopFactory] | None: + hook_caller = item.ihook.pytest_asyncio_loop_factories + if not hook_caller.get_hookimpls(): + return None + + 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 + + _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 \ @@ -481,7 +526,11 @@ def _loop_scope(self) -> _ScopeName: 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]: @@ -570,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) @@ -600,17 +659,68 @@ 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: + specialized_item_class = PytestAsyncioFunction.item_subclass_for( + metafunc.definition + ) + if specialized_item_class is None: + return + + 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__) + 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__, + effective_factories.values(), + ids=effective_factories.keys(), + indirect=True, + scope=loop_scope, + ) + + @contextlib.contextmanager def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]: old_loop_policy = _get_event_loop_policy() @@ -754,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) @@ -770,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: @@ -798,12 +930,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 +966,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/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 new file mode 100644 index 00000000..f6bac235 --- /dev/null +++ b/tests/test_loop_factory_parametrization.py @@ -0,0 +1,573 @@ +from __future__ import annotations + +from textwrap import dedent + +import pytest +from pytest import Pytester + + +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 + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"custom": 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_named_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 { + "factory_a": CustomEventLoopA, + "factory_b": 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_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 + + class CustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"custom": 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_by_hook_factories(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 { + "factory_a": CustomEventLoopA, + "factory_b": CustomEventLoopB, + } + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + def test_sync(request): + assert True + + @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 None", + "return {}", + "return [CustomEventLoop]", + "return {'': CustomEventLoop}", + "return {'default': 1}", + ), +) +def test_hook_requires_non_empty_mapping_of_named_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 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_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_hook_respects_conftest_locality( + 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 {"root": RootCustomEventLoop} + """)) + subdir = pytester.mkdir("subdir") + subdir.joinpath("conftest.py").write_text( + dedent("""\ + import asyncio + + class SubCustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"sub": SubCustomEventLoop} + """), + ) + pytester.makepyfile( + test_root=dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_root_uses_root_loop(): + assert ( + type(asyncio.get_running_loop()).__name__ == "RootCustomEventLoop" + ) + """), + ) + subdir.joinpath("test_sub.py").write_text( + dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio + async def test_sub_uses_sub_loop(): + assert type(asyncio.get_running_loop()).__name__ == "SubCustomEventLoop" + """), + ) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +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 MainCustomEventLoop(asyncio.SelectorEventLoop): + pass + + class AlternativeCustomEventLoop(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return { + "main": MainCustomEventLoop, + "alternative": AlternativeCustomEventLoop, + } + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @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, + 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 {"factory_a": CustomEventLoopA} + else: + return {"factory_b": 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 {"factory_a": CustomEventLoopA} + else: + return {"factory_b": 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 {"custom": 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 {"factory_a": CustomEventLoopA} + else: + return {"factory_b": 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)