Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions changelog.d/1501.fixed.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
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
through ``request.getfixturevalue()`` do not trigger this parametrization.
21 changes: 21 additions & 0 deletions docs/how-to-guides/custom_loop_factory.rst
Original file line number Diff line number Diff line change
Expand Up @@ -23,4 +23,25 @@ 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.

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`.
2 changes: 1 addition & 1 deletion docs/reference/hooks.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
80 changes: 73 additions & 7 deletions pytest_asyncio/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -685,6 +685,38 @@ def _resolve_asyncio_marker(item: Function) -> Mark | None:
return None


def _get_managed_fixture_loop_scope(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This reimplements pytest's fixture traversal system in a way that's error-prone is difficult to maintain.

@TLNing260310 TLNing260310 Sep 1, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in d86d02c. The recursive fixture traversal, FixtureManager fallback, and override-chain bookkeeping have been removed. Sync detection now only scans the pytest-computed metafunc.fixturenames closure and reads the active fixturedefs[-1]. The docs explicitly limit the behavior to dependencies represented in the static closure rather than reimplementing traversal for dynamic or same-name override cases.

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")
loop_scopes: list[Scope] = []
# 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:
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)
)
if is_managed:
loop_scope = (
getattr(func, "_loop_scope", None)
or default_loop_scope
or fixturedef.scope
)
loop_scopes.append(Scope(loop_scope))
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)
Expand Down Expand Up @@ -720,6 +752,14 @@ 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)

Expand All @@ -731,17 +771,43 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> 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
)
_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."
Expand All @@ -750,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]
Expand All @@ -772,19 +838,19 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
),
)
)
for name in marker_selected_factory_names
for name in 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
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(
_asyncio_loop_factory.__name__,
factory_params,
ids=(pytest.HIDDEN_PARAM,) if hide_id else factory_ids,
indirect=True,
scope=loop_scope,
scope=effective_loop_scope,
)


Expand Down
Loading
Loading