Improve detection of multiple event loops being requested in a test

docs/source/reference/changelog.rst

@@ -9,6 +9,7 @@ Changelog
 - Deprecates the optional `scope` keyword argument to `pytest.mark.asyncio` for API consistency with ``pytest_asyncio.fixture``. Users are encouraged to use the `loop_scope` keyword argument, which does exactly the same.
 - Raises an error when passing `scope` or `loop_scope` as a positional argument to ``@pytest.mark.asyncio``. `#812 <https://github.com/pytest-dev/pytest-asyncio/issues/812>`_
 - Fixes a bug that caused module-scoped async fixtures to fail when reused in other modules `#862 <https://github.com/pytest-dev/pytest-asyncio/issues/862>`_ `#668 <https://github.com/pytest-dev/pytest-asyncio/issues/668>`_
+- Improves detection of multiple event loops being requested by the same test in strict mode `#868 <https://github.com/pytest-dev/pytest-asyncio/issues/868>`_
 
 
 0.23.8 (2024-07-17)

pytest_asyncio/plugin.py

@@ -8,6 +8,7 @@
 import socket
 import warnings
 from asyncio import AbstractEventLoopPolicy
+from itertools import dropwhile
 from textwrap import dedent
 from typing import (
     Any,
@@ -730,6 +731,26 @@ def pytest_generate_tests(metafunc: Metafunc) -> None:
     event_loop_fixture_id = event_loop_node.stash.get(_event_loop_fixture_id, None)
 
     if event_loop_fixture_id:
+        collectors = _iter_collectors(metafunc.definition)
+        collector_event_loop_fixture_ids = (
+            c.stash.get(_event_loop_fixture_id, None)  # type: ignore[arg-type]
+            for c in collectors
+        )
+        possible_event_loop_fixture_ids = {"event_loop"} | set(
+            collector_event_loop_fixture_ids
+        )
+        used_fixture_ids = {event_loop_fixture_id, *metafunc.fixturenames}
+        used_event_loop_fixture_ids = possible_event_loop_fixture_ids.intersection(
+            used_fixture_ids
+        )
+        if len(used_event_loop_fixture_ids) > 1:
+            raise MultipleEventLoopsRequestedError(
+                _MULTIPLE_LOOPS_REQUESTED_ERROR.format(
+                    test_name=metafunc.definition.nodeid,
+                    scope=scope,
+                    scoped_loop_node=event_loop_node.nodeid,
+                ),
+            )
         # This specific fixture name may already be in metafunc.argnames, if this
         # test indirectly depends on the fixture. For example, this is the case
         # when the test depends on an async fixture, both of which share the same
@@ -738,14 +759,6 @@ def pytest_generate_tests(metafunc: Metafunc) -> None:
             return
         fixturemanager = metafunc.config.pluginmanager.get_plugin("funcmanage")
         assert fixturemanager is not None
-        if "event_loop" in metafunc.fixturenames:
-            raise MultipleEventLoopsRequestedError(
-                _MULTIPLE_LOOPS_REQUESTED_ERROR.format(
-                    test_name=metafunc.definition.nodeid,
-                    scope=scope,
-                    scoped_loop_node=event_loop_node.nodeid,
-                ),
-            )
         # Add the scoped event loop fixture to Metafunc's list of fixture names and
         # fixturedefs and leave the actual parametrization to pytest
         # The fixture needs to be appended to avoid messing up the fixture evaluation
@@ -1009,18 +1022,26 @@ def _retrieve_scope_root(item: Union[Collector, Item], scope: str) -> Collector:
         "package": Package,
         "session": Session,
     }
+    collectors = _iter_collectors(item)
     scope_root_type = node_type_by_scope[scope]
-    for node in reversed(item.listchain()):
-        if isinstance(node, scope_root_type):
-            assert isinstance(node, pytest.Collector)
-            return node
+    collector_with_specified_scope = next(
+        dropwhile(lambda c: not isinstance(c, scope_root_type), collectors), None
+    )
+    if collector_with_specified_scope:
+        return collector_with_specified_scope
     error_message = (
         f"{item.name} is marked to be run in an event loop with scope {scope}, "
         f"but is not part of any {scope}."
     )
     raise pytest.UsageError(error_message)
 
 
+def _iter_collectors(item: Union[Collector, Item]) -> Iterable[Collector]:
+    for node in reversed(item.listchain()):
+        if isinstance(node, pytest.Collector):
+            yield node
+
+
 @pytest.fixture
 def event_loop(request: FixtureRequest) -> Iterator[asyncio.AbstractEventLoop]:
     """Create an instance of the default event loop for each test case."""

tests/async_fixtures/test_autouse_fixtures.py

@@ -0,0 +1,39 @@
+from textwrap import dedent
+
+import pytest
+from pytest import Pytester
+
+
+@pytest.mark.parametrize("autouse_fixture_scope", ("function", "module"))
+def test_autouse_fixture_in_different_scope_triggers_multiple_event_loop_error(
+    pytester: Pytester,
+    autouse_fixture_scope: str,
+):
+    pytester.makepyfile(
+        dedent(
+            f"""\
+            import asyncio
+            import pytest
+            import pytest_asyncio
+
+            loop: asyncio.AbstractEventLoop
+
+            @pytest_asyncio.fixture(autouse=True, scope="{autouse_fixture_scope}")
+            async def autouse_fixture():
+                pass
+
+            @pytest_asyncio.fixture(scope="session")
+            async def any_fixture():
+                global loop
+                loop = asyncio.get_running_loop()
+
+            @pytest.mark.asyncio(scope="session")
+            async def test_runs_in_session_scoped_loop(any_fixture):
+                global loop
+                assert asyncio.get_running_loop() is loop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(errors=1)
+    result.stdout.fnmatch_lines("*MultipleEventLoopsRequestedError: *")