Fix unclosed event loop

Author: tjkuson

changelog.d/724.fixed.rst

@@ -0,0 +1 @@
+Fixed a ``ResourceWarning: unclosed event loop`` warning that could occur when a synchronous test called ``asyncio.run()`` or otherwise unset the current event loop after pytest-asyncio had run an async test or fixture.

pytest_asyncio/plugin.py

@@ -806,23 +806,13 @@ def _temporary_event_loop(loop: AbstractEventLoop) -> Iterator[None]:
 @contextlib.contextmanager
 def _temporary_event_loop_policy(
     policy: AbstractEventLoopPolicy,
-    *,
-    has_custom_factory: bool,
 ) -> Iterator[None]:
     old_loop_policy = _get_event_loop_policy()
-    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
     finally:
         _set_event_loop_policy(old_loop_policy)
-        _set_event_loop(old_loop)
 
 
 def _get_event_loop_policy() -> AbstractEventLoopPolicy:
@@ -1042,10 +1032,7 @@ 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,
-            has_custom_factory=_asyncio_loop_factory is not None,
-        ):
+        with _temporary_event_loop_policy(new_loop_policy):
             runner = Runner(
                 debug=debug_mode,
                 loop_factory=_asyncio_loop_factory,
@@ -1068,6 +1055,9 @@ def _scoped_runner(
                             _RUNNER_TEARDOWN_WARNING % traceback.format_exc(),
                             RuntimeWarning,
                         )
+            finally:
+                if _asyncio_loop_factory is not None:
+                    _set_event_loop(None)
 
     return _scoped_runner
 

tests/test_set_event_loop.py

@@ -99,6 +99,40 @@ async def test_after(self):
     result.assert_outcomes(passed=3)
 
 
+def test_asyncio_run_after_async_fixture_does_not_leak_loop(
+    pytester: Pytester,
+):
+    pytester.makeini(dedent("""\
+            [pytest]
+            asyncio_default_fixture_loop_scope = function
+            """))
+    pytester.makepyfile(dedent("""\
+            import asyncio
+            import gc
+            import pytest
+            import pytest_asyncio
+
+            pytest_plugins = "pytest_asyncio"
+
+            @pytest_asyncio.fixture
+            async def async_fixture():
+                yield
+
+            @pytest.mark.asyncio
+            async def test_async_function_uses_async_fixture(async_fixture):
+                pass
+
+            def test_collect_unclosed_loops():
+                async def amain():
+                    pass
+
+                asyncio.run(amain())
+                gc.collect()
+            """))
+    result = pytester.runpytest_subprocess("-W", "error")
+    result.assert_outcomes(passed=2)
+
+
 @pytest.mark.parametrize("test_loop_scope", ("module", "package", "session"))
 @pytest.mark.parametrize(
     "loop_breaking_action",