refactor: Extracted function to determine the event loop fixture used to run an async fixture.

Author: seifertm

pytest_asyncio/plugin.py

@@ -322,20 +322,9 @@ def _wrap_asyncgen_fixture(fixturedef: FixtureDef) -> None:
     def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
         unittest = fixturedef.unittest if hasattr(fixturedef, "unittest") else False
         func = _perhaps_rebind_fixture_func(fixture, request.instance, unittest)
-        default_loop_scope = request.config.getini("asyncio_default_fixture_loop_scope")
-        loop_scope = (
-            getattr(func, "_loop_scope", None) or default_loop_scope or request.scope
+        event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
+            request, func
         )
-        if loop_scope == "function":
-            event_loop_fixture_id = "event_loop"
-        else:
-            event_loop_node = _retrieve_scope_root(request._pyfuncitem, loop_scope)
-            event_loop_fixture_id = event_loop_node.stash.get(
-                # Type ignored because of non-optimal mypy inference.
-                _event_loop_fixture_id,  # type: ignore[arg-type]
-                "",
-            )
-        assert event_loop_fixture_id
         event_loop = request.getfixturevalue(event_loop_fixture_id)
         kwargs.pop(event_loop_fixture_id, None)
         gen_obj = func(**_add_kwargs(func, kwargs, event_loop, request))
@@ -373,20 +362,9 @@ def _wrap_async_fixture(fixturedef: FixtureDef) -> None:
     def _async_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
         unittest = False if pytest.version_tuple >= (8, 2) else fixturedef.unittest
         func = _perhaps_rebind_fixture_func(fixture, request.instance, unittest)
-        default_loop_scope = request.config.getini("asyncio_default_fixture_loop_scope")
-        loop_scope = (
-            getattr(func, "_loop_scope", None) or default_loop_scope or request.scope
+        event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
+            request, func
         )
-        if loop_scope == "function":
-            event_loop_fixture_id = "event_loop"
-        else:
-            event_loop_node = _retrieve_scope_root(request._pyfuncitem, loop_scope)
-            event_loop_fixture_id = event_loop_node.stash.get(
-                # Type ignored because of non-optimal mypy inference.
-                _event_loop_fixture_id,  # type: ignore[arg-type]
-                "",
-            )
-        assert event_loop_fixture_id
         event_loop = request.getfixturevalue(event_loop_fixture_id)
         kwargs.pop(event_loop_fixture_id, None)
 
@@ -399,6 +377,26 @@ async def setup():
     fixturedef.func = _async_fixture_wrapper
 
 
+def _get_event_loop_fixture_id_for_async_fixture(
+    request: FixtureRequest, func: Any
+) -> str:
+    default_loop_scope = request.config.getini("asyncio_default_fixture_loop_scope")
+    loop_scope = (
+        getattr(func, "_loop_scope", None) or default_loop_scope or request.scope
+    )
+    if loop_scope == "function":
+        event_loop_fixture_id = "event_loop"
+    else:
+        event_loop_node = _retrieve_scope_root(request._pyfuncitem, loop_scope)
+        event_loop_fixture_id = event_loop_node.stash.get(
+            # Type ignored because of non-optimal mypy inference.
+            _event_loop_fixture_id,  # type: ignore[arg-type]
+            "",
+        )
+    assert event_loop_fixture_id
+    return event_loop_fixture_id
+
+
 class PytestAsyncioFunction(Function):
     """Base class for all test functions managed by pytest-asyncio."""