[feat] Detect multiple higher-scoped event loops requested in the sam…

…e test. Previously, the detection was limited to a function-scoped loop and a higher-scoped loop. Signed-off-by: Michael Seifert <[email protected]>
pytest-dev · Aug 12, 2024 · bae1d1e · bae1d1e
1 parent 8095cde
commit bae1d1e
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 3 deletions.
diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py
@@ -731,7 +731,19 @@ 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:
-        if "event_loop" in metafunc.fixturenames:
+        collectors = _iter_collectors(metafunc.definition)
+        collector_event_loop_fixture_ids = map(
+            lambda c: c.stash.get(_event_loop_fixture_id, None),  # type: ignore
+            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,

diff --git a/tests/async_fixtures/test_autouse_fixtures.py b/tests/async_fixtures/test_autouse_fixtures.py
@@ -1,21 +1,24 @@
 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)
+            @pytest_asyncio.fixture(autouse=True, scope="{autouse_fixture_scope}")
             async def autouse_fixture():
                 pass