From 7271d15487de80ec93798d413da7e07e699c579b Mon Sep 17 00:00:00 2001 From: TLNing260310 <266924230+TLNing260310@users.noreply.github.com> Date: Mon, 31 Aug 2026 19:31:21 +0800 Subject: [PATCH 1/3] Fix loop factory parametrization for sync fixtures --- changelog.d/1501.fixed.rst | 5 + docs/how-to-guides/custom_loop_factory.rst | 15 + pytest_asyncio/plugin.py | 117 +++++- tests/test_loop_factory_parametrization.py | 449 +++++++++++++++++++++ 4 files changed, 576 insertions(+), 10 deletions(-) create mode 100644 changelog.d/1501.fixed.rst diff --git a/changelog.d/1501.fixed.rst b/changelog.d/1501.fixed.rst new file mode 100644 index 00000000..a99e4028 --- /dev/null +++ b/changelog.d/1501.fixed.rst @@ -0,0 +1,5 @@ +Synchronous tests whose statically resolved fixture closure contains a +pytest-asyncio-managed fixture are now parametrized over configured event loop +factories, preventing shared fixtures from being torn down at +synchronous/asynchronous test boundaries. Fixtures requested only dynamically +through ``request.getfixturevalue()`` do not trigger this parametrization. diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst index fb4d9f9c..efd55480 100644 --- a/docs/how-to-guides/custom_loop_factory.rst +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -23,4 +23,19 @@ pytest-asyncio can run asynchronous tests with custom event loop factories by im The hook receives the current pytest ``item``, so it can return different factory mappings for different tests. See :doc:`configure_loop_factories_per_test` for item-based factory configuration. +Synchronous tests are not parametrized by the hook unless their statically +resolved fixture closure contains a pytest-asyncio-managed fixture. This +includes fixtures requested through test arguments, ``usefixtures``, autouse +fixtures, and transitive fixture dependencies. Such a test runs once for each +configured loop factory, ensuring the managed fixture is created and torn down +on the corresponding event loop. + +Direct test parameters shadow fixtures with the same name and therefore do not +trigger parametrization through those fixtures. Indirect parameters continue to +use their fixture definitions normally. + +A managed fixture requested only dynamically through +``request.getfixturevalue()`` is not known during collection and does not +trigger loop factory parametrization for a synchronous test. + To run a test with only some configured factories, see :doc:`run_test_with_specific_loop_factories`. diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 38b75e41..650d3cd5 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -685,6 +685,83 @@ def _resolve_asyncio_marker(item: Function) -> Mark | None: return None +def _get_managed_fixture_loop_scope( + metafunc: pytest.Metafunc, +) -> _ScopeName | None: + """Return the widest loop scope of managed fixtures used by a sync test.""" + asyncio_mode = _get_asyncio_mode(metafunc.config) + default_loop_scope = metafunc.config.getini( + "asyncio_default_fixture_loop_scope" + ) + fixtureinfo = metafunc.definition._fixtureinfo + unresolved_initialnames = { + fixture_name + for fixture_name in fixtureinfo.initialnames + if fixture_name not in metafunc._arg2fixturedefs + } + discovered_fixturedefs: dict[str, Sequence[FixtureDef] | None] = {} + + def get_fixturedefs(fixture_name: str) -> Sequence[FixtureDef] | None: + fixturedefs = metafunc._arg2fixturedefs.get(fixture_name) + if fixturedefs is not None: + return fixturedefs + # Directly parametrized test arguments have no static FixtureDef and must + # not fall back to a fixture with the same name. + if fixture_name in unresolved_initialnames: + return None + if fixture_name not in discovered_fixturedefs: + # pytest 8.4 does not add dependencies of overridden super fixtures + # to names_closure, so resolve those dependencies on demand. + discovered_fixturedefs[fixture_name] = ( + metafunc.definition.session._fixturemanager.getfixturedefs( + fixture_name, metafunc.definition + ) + ) + return discovered_fixturedefs[fixture_name] + + loop_scopes: list[Scope] = [] + current_fixturedef_indices: dict[str, int] = {} + + def collect_loop_scopes(fixture_name: str) -> None: + fixturedef_index = current_fixturedef_indices.get(fixture_name) + if fixturedef_index == -1: + return + fixturedefs = get_fixturedefs(fixture_name) + if not fixturedefs: + return + if fixturedef_index is None: + fixturedef_index = -1 + if -fixturedef_index > len(fixturedefs): + return + fixturedef = fixturedefs[fixturedef_index] + func = fixturedef.func + is_managed = _is_asyncio_fixture_function(func) or ( + asyncio_mode == Mode.AUTO and _is_coroutine_or_asyncgen(func) + ) + if is_managed: + loop_scope = ( + getattr(func, "_loop_scope", None) + or default_loop_scope + or fixturedef.scope + ) + loop_scopes.append(Scope(loop_scope)) + + # A fixture can request an overridden fixture with the same name. Track + # the active definition index so that such a request visits the next + # definition in the override chain, matching pytest's runtime lookup. + current_fixturedef_indices[fixture_name] = fixturedef_index - 1 + for dependency_name in fixturedef.argnames: + collect_loop_scopes(dependency_name) + current_fixturedef_indices[fixture_name] = fixturedef_index + + for fixture_name in fixtureinfo.initialnames: + collect_loop_scopes(fixture_name) + if not loop_scopes: + return None + # Scope is ordered from function (narrowest) to session (widest). + return max(loop_scopes).value + + # 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) @@ -724,20 +801,26 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass( hook_result.force_result(updated_node_collection) -@pytest.hookimpl(tryfirst=True) +# Direct parametrization replaces same-named fixtures during this hook. Run after +# pytest and user hooks so the resolved fixture graph reflects those replacements. +@pytest.hookimpl(trylast=True) def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: specialized_item_class = PytestAsyncioFunction.item_subclass_for( metafunc.definition ) + sync_test_uses_managed_fixture = specialized_item_class is None 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 - ) + marker_loop_scope = _get_managed_fixture_loop_scope(metafunc) + if marker_loop_scope is None: + return + marker_selected_factory_names = None + else: + 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: @@ -774,7 +857,21 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: ) for name in marker_selected_factory_names ] - metafunc.fixturenames.append(_asyncio_loop_factory.__name__) + if sync_test_uses_managed_fixture: + # Resolve the parameter before any managed fixture checks its cache. This + # ensures a factory change tears down and rebuilds wider-scoped fixtures + # before their values are supplied to a synchronous test. + fixtureinfo = metafunc.definition._fixtureinfo + if _asyncio_loop_factory.__name__ not in fixtureinfo.initialnames: + object.__setattr__( + fixtureinfo, + "initialnames", + (_asyncio_loop_factory.__name__, *fixtureinfo.initialnames), + ) + if _asyncio_loop_factory.__name__ not in metafunc.fixturenames: + metafunc.fixturenames.insert(0, _asyncio_loop_factory.__name__) + else: + 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 # pytest.HIDDEN_PARAM was added in pytest 8.4 diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index 224c9141..afa7e6db 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -190,6 +190,455 @@ async def test_async(request): result.assert_outcomes(passed=3) +@pytest.mark.parametrize( + "async_test_first", (True, False), ids=("async-first", "sync-first") +) +def test_sync_test_using_shared_async_fixture_uses_loop_factory_parameter( + pytester: Pytester, + async_test_first: bool, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = session") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + events = [] + + def pytest_asyncio_loop_factories(config, item): + return {"default": asyncio.new_event_loop} + + @pytest_asyncio.fixture(scope="session") + async def parent(): + events.append("parent setup") + yield "parent" + events.append("parent teardown") + + @pytest_asyncio.fixture(scope="session") + async def child(parent): + events.append("child setup") + yield "child" + events.append("child teardown") + + def pytest_sessionfinish(session): + assert events == [ + "parent setup", + "child setup", + "child teardown", + "parent teardown", + ] + """)) + async_test = dedent("""\ + @pytest.mark.asyncio(loop_scope="session") + async def test_async(parent): + assert parent == "parent" + """) + sync_test = dedent("""\ + def test_sync(child): + assert child == "child" + """) + test_bodies = async_test + sync_test if async_test_first else sync_test + async_test + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + """) + test_bodies) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +@pytest.mark.parametrize( + ("fixture_scope", "loop_scope"), + ( + ("function", "function"), + ("function", "module"), + ("module", "session"), + ("session", "session"), + ), +) +def test_sync_test_async_fixture_runs_and_tears_down_for_each_loop_factory( + pytester: Pytester, + fixture_scope: str, + loop_scope: str, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent(f"""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {{"loop_a": CustomLoopA, "loop_b": CustomLoopB}} + + @pytest_asyncio.fixture(scope="{fixture_scope}", loop_scope="{loop_scope}") + async def loop_name(): + name = type(asyncio.get_running_loop()).__name__ + yield name + print(f"TEARDOWN:{{name}}") + """)) + pytester.makepyfile(dedent("""\ + pytest_plugins = "pytest_asyncio" + + def test_sync(loop_name): + assert loop_name in ("CustomLoopA", "CustomLoopB") + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-s") + result.assert_outcomes(passed=2) + output = result.stdout.str() + assert output.count("TEARDOWN:CustomLoopA") == 1 + assert output.count("TEARDOWN:CustomLoopB") == 1 + + +def test_sync_fixture_override_using_async_super_fixture_runs_for_each_factory( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture(scope="session", loop_scope="session") + async def loop_name(): + return type(asyncio.get_running_loop()).__name__ + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.fixture + def loop_name(loop_name): + return f"wrapped:{loop_name}" + + def test_sync(loop_name): + assert loop_name in ( + "wrapped:CustomLoopA", + "wrapped:CustomLoopB", + ) + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_sync_fixture_override_finds_async_dependency_of_super_fixture( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture(scope="session", loop_scope="session") + async def managed_loop_name(): + return type(asyncio.get_running_loop()).__name__ + + @pytest.fixture(scope="session") + def resource(managed_loop_name): + return managed_loop_name + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.fixture + def resource(resource): + return f"wrapped:{resource}" + + def test_sync(resource): + assert resource in ( + "wrapped:CustomLoopA", + "wrapped:CustomLoopB", + ) + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_direct_parameter_shadows_managed_fixture_for_sync_test( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture + async def value(): + raise AssertionError("the directly parametrized value must win") + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.parametrize("value", (1, 2)) + def test_sync(value): + assert value in (1, 2) + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_direct_parameter_shadows_transitive_managed_fixture_for_sync_test( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture + async def value(): + raise AssertionError("the directly parametrized value must win") + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.fixture + def resource(value): + return value + + @pytest.mark.parametrize("value", (1, 2)) + def test_sync(resource): + assert resource in (1, 2) + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_mixed_indirect_direct_parameter_shadows_transitive_managed_fixture( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture + async def value(): + raise AssertionError("the directly parametrized value must win") + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.fixture + def resource(value): + return value + + @pytest.fixture + def other(request): + return request.param + + @pytest.mark.parametrize( + ("value", "other"), + ((1, "a"), (2, "b")), + indirect=["other"], + ) + def test_sync(resource, other): + assert resource in (1, 2) + assert other in ("a", "b") + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_generate_tests_direct_parameter_shadows_transitive_managed_fixture( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture + async def value(): + raise AssertionError("the directly parametrized value must win") + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + def pytest_generate_tests(metafunc): + if "value" in metafunc.fixturenames: + metafunc.parametrize("value", (1, 2)) + + @pytest.fixture + def resource(value): + return value + + def test_sync(resource): + assert resource in (1, 2) + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_indirect_parameter_keeps_managed_fixture_for_sync_test( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture + async def value(request): + return request.param, type(asyncio.get_running_loop()).__name__ + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.parametrize("value", (1, 2), indirect=True) + def test_sync(value): + parameter, loop_name = value + assert parameter in (1, 2) + assert loop_name in ("CustomLoopA", "CustomLoopB") + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=4) + + +def test_sync_test_with_autouse_managed_sync_fixture_runs_for_each_loop_factory( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + + @pytest_asyncio.fixture(autouse=True) + def custom_loop_is_active(): + loop_name = type(asyncio.get_event_loop()).__name__ + assert loop_name in ("CustomLoopA", "CustomLoopB") + """)) + pytester.makepyfile(dedent("""\ + pytest_plugins = "pytest_asyncio" + + def test_sync(): + pass + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=2) + + +def test_sync_test_with_auto_mode_async_fixture_runs_for_each_loop_factory( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + + class CustomLoopA(asyncio.SelectorEventLoop): + pass + + class CustomLoopB(asyncio.SelectorEventLoop): + pass + + def pytest_asyncio_loop_factories(config, item): + return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + """)) + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.fixture + async def loop_name(): + return type(asyncio.get_running_loop()).__name__ + + def test_sync(loop_name): + assert loop_name in ("CustomLoopA", "CustomLoopB") + """)) + result = pytester.runpytest("--asyncio-mode=auto") + result.assert_outcomes(passed=2) + + @pytest.mark.parametrize( "hook_body", ( From 920ab14965aa5ba3e0c27555ef064c6b29e690eb Mon Sep 17 00:00:00 2001 From: TLNing260310 <266924230+TLNing260310@users.noreply.github.com> Date: Mon, 31 Aug 2026 20:13:31 +0800 Subject: [PATCH 2/3] Apply shed formatting --- pytest_asyncio/plugin.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 650d3cd5..7d638fb8 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -690,9 +690,7 @@ def _get_managed_fixture_loop_scope( ) -> _ScopeName | None: """Return the widest loop scope of managed fixtures used by a sync test.""" asyncio_mode = _get_asyncio_mode(metafunc.config) - default_loop_scope = metafunc.config.getini( - "asyncio_default_fixture_loop_scope" - ) + default_loop_scope = metafunc.config.getini("asyncio_default_fixture_loop_scope") fixtureinfo = metafunc.definition._fixtureinfo unresolved_initialnames = { fixture_name From d86d02c7154e1109052059c7114e7a8b6439a85c Mon Sep 17 00:00:00 2001 From: TLNing260310 <266924230+TLNing260310@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:36:55 +0800 Subject: [PATCH 3/3] Refine sync loop factory parametrization --- changelog.d/1501.fixed.rst | 2 +- docs/how-to-guides/custom_loop_factory.rst | 6 + docs/reference/hooks.rst | 2 +- pytest_asyncio/plugin.py | 143 +++++++----------- tests/test_loop_factory_parametrization.py | 168 ++++++++++----------- 5 files changed, 142 insertions(+), 179 deletions(-) diff --git a/changelog.d/1501.fixed.rst b/changelog.d/1501.fixed.rst index a99e4028..347ec571 100644 --- a/changelog.d/1501.fixed.rst +++ b/changelog.d/1501.fixed.rst @@ -1,4 +1,4 @@ -Synchronous tests whose statically resolved fixture closure contains a +Synchronous tests whose pytest-computed static fixture closure contains a pytest-asyncio-managed fixture are now parametrized over configured event loop factories, preventing shared fixtures from being torn down at synchronous/asynchronous test boundaries. Fixtures requested only dynamically diff --git a/docs/how-to-guides/custom_loop_factory.rst b/docs/how-to-guides/custom_loop_factory.rst index efd55480..f782ebcb 100644 --- a/docs/how-to-guides/custom_loop_factory.rst +++ b/docs/how-to-guides/custom_loop_factory.rst @@ -38,4 +38,10 @@ A managed fixture requested only dynamically through ``request.getfixturevalue()`` is not known during collection and does not trigger loop factory parametrization for a synchronous test. +pytest-asyncio relies on the static fixture closure supplied by pytest and does +not perform additional fixture traversal. Dependencies introduced only by a +dynamic ``pytest_generate_tests`` rewrite, or hidden behind an overridden +same-name fixture definition that pytest does not include in that closure, do +not trigger parametrization. + To run a test with only some configured factories, see :doc:`run_test_with_specific_loop_factories`. diff --git a/docs/reference/hooks.rst b/docs/reference/hooks.rst index ed025b6f..bd942cfb 100644 --- a/docs/reference/hooks.rst +++ b/docs/reference/hooks.rst @@ -7,7 +7,7 @@ Hooks 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. +By default, each pytest-asyncio test is run once per configured factory. Tests managed by other async plugins are unaffected. A synchronous test is also parametrized when pytest's statically resolved fixture closure contains a pytest-asyncio-managed fixture; other 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. diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 7d638fb8..33cc42a6 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -691,47 +691,15 @@ def _get_managed_fixture_loop_scope( """Return the widest loop scope of managed fixtures used by a sync test.""" asyncio_mode = _get_asyncio_mode(metafunc.config) default_loop_scope = metafunc.config.getini("asyncio_default_fixture_loop_scope") - fixtureinfo = metafunc.definition._fixtureinfo - unresolved_initialnames = { - fixture_name - for fixture_name in fixtureinfo.initialnames - if fixture_name not in metafunc._arg2fixturedefs - } - discovered_fixturedefs: dict[str, Sequence[FixtureDef] | None] = {} - - def get_fixturedefs(fixture_name: str) -> Sequence[FixtureDef] | None: - fixturedefs = metafunc._arg2fixturedefs.get(fixture_name) - if fixturedefs is not None: - return fixturedefs - # Directly parametrized test arguments have no static FixtureDef and must - # not fall back to a fixture with the same name. - if fixture_name in unresolved_initialnames: - return None - if fixture_name not in discovered_fixturedefs: - # pytest 8.4 does not add dependencies of overridden super fixtures - # to names_closure, so resolve those dependencies on demand. - discovered_fixturedefs[fixture_name] = ( - metafunc.definition.session._fixturemanager.getfixturedefs( - fixture_name, metafunc.definition - ) - ) - return discovered_fixturedefs[fixture_name] - loop_scopes: list[Scope] = [] - current_fixturedef_indices: dict[str, int] = {} - - def collect_loop_scopes(fixture_name: str) -> None: - fixturedef_index = current_fixturedef_indices.get(fixture_name) - if fixturedef_index == -1: - return - fixturedefs = get_fixturedefs(fixture_name) + # Let pytest own fixture traversal and shadowing. There is no public API for + # retrieving the active FixtureDef from Metafunc, so this is intentionally a + # shallow, read-only use of pytest's already-computed fixture closure. + for fixture_name in metafunc.fixturenames: + fixturedefs = metafunc._arg2fixturedefs.get(fixture_name) if not fixturedefs: - return - if fixturedef_index is None: - fixturedef_index = -1 - if -fixturedef_index > len(fixturedefs): - return - fixturedef = fixturedefs[fixturedef_index] + continue + fixturedef = fixturedefs[-1] func = fixturedef.func is_managed = _is_asyncio_fixture_function(func) or ( asyncio_mode == Mode.AUTO and _is_coroutine_or_asyncgen(func) @@ -743,17 +711,6 @@ def collect_loop_scopes(fixture_name: str) -> None: or fixturedef.scope ) loop_scopes.append(Scope(loop_scope)) - - # A fixture can request an overridden fixture with the same name. Track - # the active definition index so that such a request visits the next - # definition in the override chain, matching pytest's runtime lookup. - current_fixturedef_indices[fixture_name] = fixturedef_index - 1 - for dependency_name in fixturedef.argnames: - collect_loop_scopes(dependency_name) - current_fixturedef_indices[fixture_name] = fixturedef_index - - for fixture_name in fixtureinfo.initialnames: - collect_loop_scopes(fixture_name) if not loop_scopes: return None # Scope is ordered from function (narrowest) to session (widest). @@ -795,34 +752,62 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass( and _resolve_asyncio_marker(node) is not None ): updated_item = specialized_item_class._from_function(node) + elif ( + hasattr(node, "callspec") + and _asyncio_loop_factory.__name__ in node.callspec.params + ): + # pytest prunes the dynamically parametrized fixture name from + # the static closure before creating the Function item. Put it + # first so its cache key changes before managed fixtures are read. + node.fixturenames.insert(0, _asyncio_loop_factory.__name__) updated_node_collection.append(updated_item) hook_result.force_result(updated_node_collection) -# Direct parametrization replaces same-named fixtures during this hook. Run after -# pytest and user hooks so the resolved fixture graph reflects those replacements. -@pytest.hookimpl(trylast=True) +@pytest.hookimpl(tryfirst=True) def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: specialized_item_class = PytestAsyncioFunction.item_subclass_for( metafunc.definition ) - sync_test_uses_managed_fixture = specialized_item_class is None if specialized_item_class is None: - marker_loop_scope = _get_managed_fixture_loop_scope(metafunc) - if marker_loop_scope is None: - return - marker_selected_factory_names = None - else: - 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 - ) + 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 + ) + _parametrize_loop_factories( + metafunc, marker_loop_scope, marker_selected_factory_names + ) + + +@pytest.hookimpl(specname="pytest_generate_tests", wrapper=True, tryfirst=True) +def pytest_generate_tests_for_sync_functions( + metafunc: pytest.Metafunc, +) -> Generator[None, object, object]: + hook_result = yield + specialized_item_class = PytestAsyncioFunction.item_subclass_for( + metafunc.definition + ) + if specialized_item_class is not None: + return hook_result + managed_fixture_loop_scope = _get_managed_fixture_loop_scope(metafunc) + if managed_fixture_loop_scope is None: + return hook_result + _parametrize_loop_factories(metafunc, managed_fixture_loop_scope, None) + return hook_result + +def _parametrize_loop_factories( + metafunc: pytest.Metafunc, + loop_scope: _ScopeName | None, + selected_factory_names: Sequence[str] | None, +) -> None: + """Parametrize a test over the loop factories selected for its item.""" hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition) if hook_factories is None: - if marker_selected_factory_names is not None: + if selected_factory_names is not None: raise pytest.UsageError( "mark.asyncio 'loop_factories' requires at least one " "pytest_asyncio_loop_factories hook implementation." @@ -831,13 +816,13 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: factory_params: Collection[object] factory_ids: Collection[str] - if marker_selected_factory_names is None: + if selected_factory_names is None: factory_params = hook_factories.values() factory_ids = hook_factories.keys() else: # Iterate in marker order to preserve explicit user selection # order. - factory_ids = marker_selected_factory_names + factory_ids = selected_factory_names factory_params = [ ( hook_factories[name] @@ -853,25 +838,11 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: ), ) ) - for name in marker_selected_factory_names + for name in selected_factory_names ] - if sync_test_uses_managed_fixture: - # Resolve the parameter before any managed fixture checks its cache. This - # ensures a factory change tears down and rebuilds wider-scoped fixtures - # before their values are supplied to a synchronous test. - fixtureinfo = metafunc.definition._fixtureinfo - if _asyncio_loop_factory.__name__ not in fixtureinfo.initialnames: - object.__setattr__( - fixtureinfo, - "initialnames", - (_asyncio_loop_factory.__name__, *fixtureinfo.initialnames), - ) - if _asyncio_loop_factory.__name__ not in metafunc.fixturenames: - metafunc.fixturenames.insert(0, _asyncio_loop_factory.__name__) - else: - metafunc.fixturenames.append(_asyncio_loop_factory.__name__) + 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 + effective_loop_scope = loop_scope or default_loop_scope # pytest.HIDDEN_PARAM was added in pytest 8.4 hide_id = len(factory_ids) == 1 and hasattr(pytest, "HIDDEN_PARAM") metafunc.parametrize( @@ -879,7 +850,7 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: factory_params, ids=(pytest.HIDDEN_PARAM,) if hide_id else factory_ids, indirect=True, - scope=loop_scope, + scope=effective_loop_scope, ) diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index afa7e6db..d47e5fc7 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -22,16 +22,27 @@ def pytest_asyncio_loop_factories(config, item): """)) pytester.makepyfile(dedent("""\ import pytest + import pytest_asyncio pytest_plugins = "pytest_asyncio" + @pytest_asyncio.fixture + async def managed_fixture(): + return "managed" + @pytest.mark.asyncio async def test_example(): assert True + + def test_sync(managed_fixture): + assert managed_fixture == "managed" """)) result = pytester.runpytest("--asyncio-mode=strict", "--collect-only", "-q") result.stdout.fnmatch_lines( - ["test_single_factory_does_not_add_suffix_to_test_name.py::test_example"] + [ + "test_single_factory_does_not_add_suffix_to_test_name.py::test_example", + "test_single_factory_does_not_add_suffix_to_test_name.py::test_sync", + ] ) @@ -123,6 +134,50 @@ async def test_runs_once_per_factory(): ) +def test_factory_ids_preserve_async_order_and_follow_sync_parameters( + pytester: Pytester, +) -> None: + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest_asyncio + + def pytest_asyncio_loop_factories(config, item): + return { + "factory_a": asyncio.new_event_loop, + "factory_b": asyncio.new_event_loop, + } + + @pytest_asyncio.fixture + async def managed_fixture(): + return "managed" + """)) + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.mark.parametrize("value", [pytest.param(1, id="value")]) + @pytest.mark.asyncio + async def test_async(value): + assert value == 1 + + @pytest.mark.parametrize("value", [pytest.param(1, id="value")]) + def test_sync(value, managed_fixture): + assert value == 1 + assert managed_fixture == "managed" + """)) + result = pytester.runpytest("--asyncio-mode=strict", "--collect-only", "-q") + result.stdout.fnmatch_lines( + [ + "*test_async[[]factory_a-value[]]", + "*test_async[[]factory_b-value[]]", + "*test_sync[[]value-factory_a[]]", + "*test_sync[[]value-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("""\ @@ -293,91 +348,6 @@ def test_sync(loop_name): assert output.count("TEARDOWN:CustomLoopB") == 1 -def test_sync_fixture_override_using_async_super_fixture_runs_for_each_factory( - pytester: Pytester, -) -> None: - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makeconftest(dedent("""\ - import asyncio - import pytest_asyncio - - class CustomLoopA(asyncio.SelectorEventLoop): - pass - - class CustomLoopB(asyncio.SelectorEventLoop): - pass - - def pytest_asyncio_loop_factories(config, item): - return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} - - @pytest_asyncio.fixture(scope="session", loop_scope="session") - async def loop_name(): - return type(asyncio.get_running_loop()).__name__ - """)) - pytester.makepyfile(dedent("""\ - import pytest - - pytest_plugins = "pytest_asyncio" - - @pytest.fixture - def loop_name(loop_name): - return f"wrapped:{loop_name}" - - def test_sync(loop_name): - assert loop_name in ( - "wrapped:CustomLoopA", - "wrapped:CustomLoopB", - ) - """)) - result = pytester.runpytest("--asyncio-mode=strict") - result.assert_outcomes(passed=2) - - -def test_sync_fixture_override_finds_async_dependency_of_super_fixture( - pytester: Pytester, -) -> None: - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makeconftest(dedent("""\ - import asyncio - import pytest - import pytest_asyncio - - class CustomLoopA(asyncio.SelectorEventLoop): - pass - - class CustomLoopB(asyncio.SelectorEventLoop): - pass - - def pytest_asyncio_loop_factories(config, item): - return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} - - @pytest_asyncio.fixture(scope="session", loop_scope="session") - async def managed_loop_name(): - return type(asyncio.get_running_loop()).__name__ - - @pytest.fixture(scope="session") - def resource(managed_loop_name): - return managed_loop_name - """)) - pytester.makepyfile(dedent("""\ - import pytest - - pytest_plugins = "pytest_asyncio" - - @pytest.fixture - def resource(resource): - return f"wrapped:{resource}" - - def test_sync(resource): - assert resource in ( - "wrapped:CustomLoopA", - "wrapped:CustomLoopB", - ) - """)) - result = pytester.runpytest("--asyncio-mode=strict") - result.assert_outcomes(passed=2) - - def test_direct_parameter_shadows_managed_fixture_for_sync_test( pytester: Pytester, ) -> None: @@ -574,11 +544,24 @@ def test_sync(value): result.assert_outcomes(passed=4) -def test_sync_test_with_autouse_managed_sync_fixture_runs_for_each_loop_factory( +@pytest.mark.parametrize( + ("fixture_decorator", "test_decorator"), + ( + ("pytest_asyncio.fixture(autouse=True)", ""), + ( + "pytest_asyncio.fixture", + '@pytest.mark.usefixtures("custom_loop_is_active")', + ), + ), + ids=("autouse", "usefixtures"), +) +def test_sync_test_with_implicit_managed_sync_fixture_runs_for_each_loop_factory( pytester: Pytester, + fixture_decorator: str, + test_decorator: str, ) -> None: pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makeconftest(dedent("""\ + pytester.makeconftest(dedent(f"""\ import asyncio import pytest_asyncio @@ -589,16 +572,19 @@ class CustomLoopB(asyncio.SelectorEventLoop): pass def pytest_asyncio_loop_factories(config, item): - return {"loop_a": CustomLoopA, "loop_b": CustomLoopB} + return {{"loop_a": CustomLoopA, "loop_b": CustomLoopB}} - @pytest_asyncio.fixture(autouse=True) + @{fixture_decorator} def custom_loop_is_active(): loop_name = type(asyncio.get_event_loop()).__name__ assert loop_name in ("CustomLoopA", "CustomLoopB") """)) - pytester.makepyfile(dedent("""\ + pytester.makepyfile(dedent(f"""\ + import pytest + pytest_plugins = "pytest_asyncio" + {test_decorator} def test_sync(): pass """))