Add cleanup code to event_loop that mimics the behavior of asyncio.run

pytest_asyncio/plugin.py

@@ -3,6 +3,7 @@
 import contextlib
 import enum
 import functools
+import gc
 import inspect
 import socket
 import sys
@@ -360,8 +361,17 @@ def pytest_fixture_post_finalizer(fixturedef: FixtureDef, request: SubRequest) -
         except RuntimeError:
             loop = None
         if loop is not None:
-            # Clean up existing loop to avoid ResourceWarnings
-            loop.close()
+            # Cleanup code based on the implementation of asyncio.run()
+            try:
+                if not loop.is_closed():
+                    asyncio.runners._cancel_all_tasks(  # type: ignore[attr-defined]
+                        loop
+                    )
+                    loop.run_until_complete(loop.shutdown_asyncgens())
+                    if sys.version_info >= (3, 9):
+                        loop.run_until_complete(loop.shutdown_default_executor())
+            finally:
+                loop.close()
         new_loop = policy.new_event_loop()  # Replace existing event loop
         # Ensure subsequent calls to get_event_loop() succeed
         policy.set_event_loop(new_loop)
@@ -486,9 +496,13 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
 @pytest.fixture
 def event_loop(request: "pytest.FixtureRequest") -> Iterator[asyncio.AbstractEventLoop]:
     """Create an instance of the default event loop for each test case."""
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
+    yield asyncio.get_event_loop_policy().new_event_loop()
+    # Call the garbage collector to trigger ResourceWarning's as soon
+    # as possible (these are triggered in various __del__ methods).
+    # Without this, resources opened in one test can fail other tests
+    # when the warning is generated.
+    gc.collect()
+    # Event loop cleanup handled by pytest_fixture_post_finalizer
 
 
 def _unused_port(socket_type: int) -> int:

tests/test_event_loop_cleanup.py

@@ -0,0 +1,38 @@
+from textwrap import dedent
+
+
+def test_task_canceled_on_test_end(testdir):
+    testdir.makepyfile(
+        dedent(
+            """\
+        import asyncio
+        import pytest
+
+        pytest_plugins = 'pytest_asyncio'
+
+        @pytest.mark.asyncio
+        async def test_a():
+            loop = asyncio.get_event_loop()
+
+            async def run_forever():
+                while True:
+                    await asyncio.sleep(0.1)
+
+            loop.create_task(run_forever())
+        """
+        )
+    )
+    testdir.makefile(
+        ".ini",
+        pytest=dedent(
+            """\
+        [pytest]
+        asyncio_mode = strict
+        filterwarnings =
+            error
+    """
+        ),
+    )
+    result = testdir.runpytest_subprocess()
+    result.assert_outcomes(passed=1)
+    result.stderr.no_fnmatch_line("Task was destroyed but it is pending!")

tests/test_event_loop_scope.py

@@ -34,4 +34,7 @@ def test_3():
 def test_4(event_loop):
     # If a test sets the loop to None -- pytest_fixture_post_finalizer()
     # still should work
+
+    # Close to avoid ResourceWarning about unclosed socket as a side effect
+    event_loop.close()
     asyncio.get_event_loop_policy().set_event_loop(None)