Fix event loop leak on Python <3.14

Author: tjkuson

changelog.d/1373.fixed.rst

@@ -1 +1 @@
-Fixed ``pytest_asyncio_loop_factories`` not installing the custom event loop as the current loop, and async fixture teardown/cache invalidation not being tied to the runner lifecycle, and sync ``@pytest_asyncio.fixture`` seeing the wrong event loop when multiple loop scopes are active.
+Fixed ``pytest_asyncio_loop_factories`` not installing the custom event loop as the current loop, async fixture teardown/cache invalidation not being tied to the runner lifecycle, sync ``@pytest_asyncio.fixture`` seeing the wrong event loop when multiple loop scopes are active, and an event loop leak on Python 3.10-3.13.

pytest_asyncio/plugin.py

@@ -788,12 +788,19 @@ def _temporary_event_loop(loop: AbstractEventLoop) -> Iterator[None]:
 
 
 @contextlib.contextmanager
-def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
+def _temporary_event_loop_policy(
+    policy: AbstractEventLoopPolicy,
+    *,
+    has_custom_factory: bool,
+) -> Iterator[None]:
     old_loop_policy = _get_event_loop_policy()
-    try:
-        old_loop = _get_event_loop_no_warn()
-    except RuntimeError:
+    if has_custom_factory:
         old_loop = None
+    else:
+        try:
+            old_loop = _get_event_loop_no_warn()
+        except RuntimeError:
+            old_loop = None
     _set_event_loop_policy(policy)
     try:
         yield
@@ -1006,7 +1013,10 @@ def _scoped_runner(
     ) -> Iterator[Runner]:
         new_loop_policy = event_loop_policy
         debug_mode = _get_asyncio_debug(request.config)
-        with _temporary_event_loop_policy(new_loop_policy):
+        with _temporary_event_loop_policy(
+            new_loop_policy,
+            has_custom_factory=_asyncio_loop_factory is not None,
+        ):
             runner = Runner(
                 debug=debug_mode,
                 loop_factory=_asyncio_loop_factory,

tests/test_loop_factory_parametrization.py

@@ -533,6 +533,42 @@ async def test_uses_custom_loop():
     result.assert_outcomes(passed=1)
 
 
+def test_no_event_loop_leak_with_custom_factory(pytester: Pytester) -> None:
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makeconftest(dedent("""\
+        import asyncio
+        import pytest_asyncio
+
+        class CustomEventLoop(asyncio.SelectorEventLoop):
+            pass
+
+        def pytest_asyncio_loop_factories(config, item):
+            return {"custom": CustomEventLoop}
+
+        @pytest_asyncio.fixture(autouse=True, scope="session", loop_scope="session")
+        async def session_fixture():
+            yield
+
+        @pytest_asyncio.fixture(autouse=True)
+        def sync_fixture():
+            asyncio.get_event_loop()
+        """))
+    pytester.makepyfile(dedent("""\
+        import pytest
+
+        pytest_plugins = "pytest_asyncio"
+
+        @pytest.mark.asyncio
+        async def test_passes():
+            assert True
+        """))
+    result = pytester.runpytest_subprocess(
+        "--asyncio-mode=auto", "-W", "error::ResourceWarning"
+    )
+    result.assert_outcomes(passed=1)
+    result.stderr.no_fnmatch_line("*unclosed event loop*")
+
+
 def test_function_loop_scope_allows_per_test_factories_with_session_default(
     pytester: Pytester,
 ) -> None: