Explain new event loop in fixture finalizer

pytest_asyncio/plugin.py

@@ -372,7 +372,12 @@ def _hypothesis_test_wraps_coroutine(function: Any) -> bool:
 
 @pytest.hookimpl(trylast=True)
 def pytest_fixture_post_finalizer(fixturedef: FixtureDef, request: SubRequest) -> None:
-    """Called after fixture teardown"""
+    """
+    Called after fixture teardown.
+
+    Note that this function may be called multiple times for any specific fixture.
+    see https://github.com/pytest-dev/pytest/issues/5848
+    """
     if fixturedef.argname == "event_loop":
         policy = asyncio.get_event_loop_policy()
         try:
@@ -382,8 +387,13 @@ def pytest_fixture_post_finalizer(fixturedef: FixtureDef, request: SubRequest) -
         if loop is not None:
             # Clean up existing loop to avoid ResourceWarnings
             loop.close()
-        new_loop = policy.new_event_loop()  # Replace existing event loop
-        # Ensure subsequent calls to get_event_loop() succeed
+        # At this point, the event loop for the current thread is closed.
+        # When a user calls asyncio.get_event_loop(), they will get a closed loop.
+        # In order to avoid this side effect from pytest-asyncio, we need to replace
+        # the current loop with a fresh one.
+        # 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.
+        new_loop = policy.new_event_loop()
         policy.set_event_loop(new_loop)
 
 

tests/test_event_loop_fixture_finalizer.py

@@ -0,0 +1,88 @@
+from textwrap import dedent
+
+from pytest import Pytester
+
+
+def test_event_loop_fixture_finalizer_returns_fresh_loop_after_test(pytester: Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+
+            import pytest
+
+            loop = asyncio.get_event_loop_policy().get_event_loop()
+
+            @pytest.mark.asyncio
+            async def test_1():
+                # This async test runs in its own event loop
+                global loop
+                running_loop = asyncio.get_event_loop_policy().get_event_loop()
+                # Make sure this test case received a different loop
+                assert running_loop is not loop
+
+            def test_2():
+                # Code outside of pytest-asyncio should not receive a "used" event loop
+                current_loop = asyncio.get_event_loop_policy().get_event_loop()
+                assert not current_loop.is_running()
+                assert not current_loop.is_closed()
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)
+
+
+def test_event_loop_fixture_finalizer_handles_loop_set_to_none_sync(
+    pytester: Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+
+            def test_sync(event_loop):
+                asyncio.get_event_loop_policy().set_event_loop(None)
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_event_loop_fixture_finalizer_handles_loop_set_to_none_async_without_fixture(
+    pytester: Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio
+            async def test_async_without_explicit_fixture_request():
+                asyncio.get_event_loop_policy().set_event_loop(None)
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_event_loop_fixture_finalizer_handles_loop_set_to_none_async_with_fixture(
+    pytester: Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            @pytest.mark.asyncio
+            async def test_async_with_explicit_fixture_request(event_loop):
+                asyncio.get_event_loop_policy().set_event_loop(None)
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)