fix: Fix broken event loop when a function-scoped test is in between two wider-scoped tests.

The event_loop fixture finalizers only close event loops that were not created by pytest-asyncio. This prevents the finalizers from accidentally closing a module-scoped loop, for example.

Author: seifertm

docs/reference/changelog.rst

@@ -2,6 +2,11 @@
 Changelog
 =========
 
+0.25.1 (UNRELEASED)
+===================
+- Fixes an issue that caused a broken event loop when a function-scoped test was executed in between two tests with wider loop scope `#950 <https://github.com/pytest-dev/pytest-asyncio/issues/950>`_
+
+
 0.25.0 (2024-12-13)
 ===================
 - Deprecated: Added warning when asyncio test requests async ``@pytest.fixture`` in strict mode. This will become an error in a future version of flake8-asyncio. `#979 <https://github.com/pytest-dev/pytest-asyncio/pull/979>`_

pytest_asyncio/plugin.py

@@ -10,7 +10,7 @@
 import inspect
 import socket
 import warnings
-from asyncio import AbstractEventLoopPolicy
+from asyncio import AbstractEventLoop, AbstractEventLoopPolicy
 from collections.abc import (
     AsyncIterator,
     Awaitable,
@@ -762,6 +762,19 @@ def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[No
     try:
         yield
     finally:
+        # Try detecting user-created event loops that were left unclosed
+        # at the end of a test.
+        try:
+            current_loop: AbstractEventLoop | None = _get_event_loop_no_warn()
+        except RuntimeError:
+            current_loop = None
+        if current_loop is not None and not current_loop.is_closed():
+            warnings.warn(
+                _UNCLOSED_EVENT_LOOP_WARNING % current_loop,
+                DeprecationWarning,
+            )
+            current_loop.close()
+
         asyncio.set_event_loop_policy(old_loop_policy)
         # When a test uses both a scoped event loop and the event_loop fixture,
         # the "_provide_clean_event_loop" finalizer of the event_loop fixture
@@ -906,7 +919,7 @@ def _close_event_loop() -> None:
         loop = policy.get_event_loop()
     except RuntimeError:
         loop = None
-    if loop is not None:
+    if loop is not None and not getattr(loop, "__pytest_asyncio", False):
         if not loop.is_closed():
             warnings.warn(
                 _UNCLOSED_EVENT_LOOP_WARNING % loop,
@@ -923,7 +936,7 @@ def _restore_policy():
             loop = _get_event_loop_no_warn(previous_policy)
         except RuntimeError:
             loop = None
-        if loop:
+        if loop and not getattr(loop, "__pytest_asyncio", False):
             loop.close()
         asyncio.set_event_loop_policy(previous_policy)
 
@@ -938,8 +951,13 @@ def _provide_clean_event_loop() -> None:
     # Note that we cannot set the loop to None, because get_event_loop only creates
     # a new loop, when set_event_loop has not been called.
     policy = asyncio.get_event_loop_policy()
-    new_loop = policy.new_event_loop()
-    policy.set_event_loop(new_loop)
+    try:
+        old_loop = _get_event_loop_no_warn(policy)
+    except RuntimeError:
+        old_loop = None
+    if old_loop is not None and not getattr(old_loop, "__pytest_asyncio", False):
+        new_loop = policy.new_event_loop()
+        policy.set_event_loop(new_loop)
 
 
 def _get_event_loop_no_warn(
@@ -1122,16 +1140,16 @@ def _retrieve_scope_root(item: Collector | Item, scope: str) -> Collector:
 def event_loop(request: FixtureRequest) -> Iterator[asyncio.AbstractEventLoop]:
     """Create an instance of the default event loop for each test case."""
     new_loop_policy = request.getfixturevalue(event_loop_policy.__name__)
-    asyncio.set_event_loop_policy(new_loop_policy)
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    # Add a magic value to the event loop, so pytest-asyncio can determine if the
-    # event_loop fixture was overridden. Other implementations of event_loop don't
-    # set this value.
-    # The magic value must be set as part of the function definition, because pytest
-    # seems to have multiple instances of the same FixtureDef or fixture function
-    loop.__original_fixture_loop = True  # type: ignore[attr-defined]
-    yield loop
-    loop.close()
+    with _temporary_event_loop_policy(new_loop_policy):
+        loop = asyncio.get_event_loop_policy().new_event_loop()
+        # Add a magic value to the event loop, so pytest-asyncio can determine if the
+        # event_loop fixture was overridden. Other implementations of event_loop don't
+        # set this value.
+        # The magic value must be set as part of the function definition, because pytest
+        # seems to have multiple instances of the same FixtureDef or fixture function
+        loop.__original_fixture_loop = True  # type: ignore[attr-defined]
+        yield loop
+        loop.close()
 
 
 @pytest.fixture(scope="session")

tests/markers/test_mixed_scope.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from textwrap import dedent
+
+from pytest import Pytester
+
+
+def test_function_scoped_loop_restores_previous_loop_scope(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+
+            module_loop: asyncio.AbstractEventLoop
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_remember_loop():
+                global module_loop
+                module_loop = asyncio.get_running_loop()
+
+            @pytest.mark.asyncio(loop_scope="function")
+            async def test_with_function_scoped_loop():
+                pass
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_runs_in_same_loop():
+                global module_loop
+                assert asyncio.get_running_loop() is module_loop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=3)

tests/test_event_loop_fixture_finalizer.py

@@ -14,7 +14,8 @@ def test_event_loop_fixture_finalizer_returns_fresh_loop_after_test(pytester: Py
 
             import pytest
 
-            loop = asyncio.get_event_loop_policy().get_event_loop()
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
 
             @pytest.mark.asyncio
             async def test_1():