From 8dece7fa90f21f9b2fd6f9ae104ac1d8bf017009 Mon Sep 17 00:00:00 2001 From: Tom Kuson Date: Fri, 10 Apr 2026 11:51:28 +0100 Subject: [PATCH] Skip unavailable requested loop factories --- .../run_test_with_specific_loop_factories.rst | 2 + docs/reference/markers/index.rst | 2 +- pytest_asyncio/plugin.py | 41 ++++++---- tests/test_loop_factory_parametrization.py | 79 ++++++++++++++++--- 4 files changed, 98 insertions(+), 26 deletions(-) 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 index 338d28ab..5be0333a 100644 --- a/docs/how-to-guides/run_test_with_specific_loop_factories.rst +++ b/docs/how-to-guides/run_test_with_specific_loop_factories.rst @@ -12,3 +12,5 @@ To run a test with only a subset of configured factories, use the ``loop_factori @pytest.mark.asyncio(loop_factories=["custom"]) async def test_only_with_custom_event_loop(): pass + +If a requested factory name is not available from the hook, the test variant for that factory is skipped. diff --git a/docs/reference/markers/index.rst b/docs/reference/markers/index.rst index bc0a584a..80275f53 100644 --- a/docs/reference/markers/index.rst +++ b/docs/reference/markers/index.rst @@ -43,7 +43,7 @@ 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. +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 names not available from the hook, those test variants are skipped. .. |auto mode| replace:: *auto mode* .. _auto mode: ../../concepts.html#auto-mode diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 5358d476..3e3d3703 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -698,30 +698,39 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: ) return + factory_params: Iterable[object] + factory_ids: Iterable[str] if marker_selected_factory_names is None: - effective_factories = hook_factories + factory_params = hook_factories.values() + factory_ids = hook_factories.keys() 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)}." + # Iterate in marker order to preserve explicit user selection + # order. + factory_ids = marker_selected_factory_names + factory_params = [ + ( + hook_factories[name] + if name in hook_factories + else pytest.param( + None, + marks=pytest.mark.skip( + reason=( + f"Loop factory {name!r} is not available." + f" Available factories:" + f" {', '.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 - } + 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(), + factory_params, + ids=factory_ids, indirect=True, scope=loop_scope, ) diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index f6bac235..41224ea8 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -371,7 +371,7 @@ async def test_runs_only_with_uvloop(): result.assert_outcomes(passed=1) -def test_asyncio_marker_loop_factories_unknown_name_errors(pytester: Pytester) -> None: +def test_unavailable_factory_skips_with_reason(pytester: Pytester) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import asyncio @@ -385,16 +385,77 @@ def pytest_asyncio_loop_factories(config, item): pytest_plugins = "pytest_asyncio" @pytest.mark.asyncio(loop_factories=["missing"]) - async def test_errors(): + async def test_skipped(): assert True """)) - result = pytester.runpytest("--asyncio-mode=strict") - result.assert_outcomes(errors=1) - result.stdout.fnmatch_lines( - [ - "*Unknown factory name(s)*Available names:*", - ] - ) + result = pytester.runpytest("--asyncio-mode=strict", "-rs") + result.assert_outcomes(skipped=1) + result.stdout.fnmatch_lines(["*SKIPPED*Loop factory 'missing' is not available*"]) + + +def test_partial_intersection_runs_available_and_skips_missing( + 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 { + "available": CustomEventLoop, + "other": asyncio.new_event_loop, + } + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio(loop_factories=["available", "missing"]) + async def test_runs_with_available(): + assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop" + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-rs") + result.assert_outcomes(passed=1, skipped=1) + result.stdout.fnmatch_lines(["*SKIPPED*Loop factory 'missing' is not available*"]) + + +def test_platform_conditional_factories(pytester: Pytester) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import sys + + def pytest_asyncio_loop_factories(config, item): + factories = {"default": asyncio.new_event_loop} + if sys.platform == "a_platform_that_does_not_exist": + factories["exotic"] = asyncio.new_event_loop + return factories + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.asyncio(loop_factories=["exotic"]) + async def test_exotic_only(): + assert True + + @pytest.mark.asyncio(loop_factories=["default"]) + async def test_default_only(): + assert True + + @pytest.mark.asyncio(loop_factories=["default", "exotic"]) + async def test_both(): + assert True + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-rs") + result.assert_outcomes(passed=2, skipped=2) + result.stdout.fnmatch_lines(["*SKIPPED*Loop factory 'exotic' is not available*"]) def test_asyncio_marker_loop_factories_without_hook_errors(