Emit ResourceWarning on event_loop teardown when loop is unclosed

docs/source/reference/changelog.rst

@@ -5,6 +5,7 @@ Changelog
 UNRELEASED
 =================
 - Drop compatibility with pytest 6.1. Pytest-asyncio now depends on pytest 7.0 or newer.
+- event_loop fixture teardown emits a ResourceWarning when the current event loop has not been closed.
 
 0.20.3 (22-12-08)
 =================

pytest_asyncio/plugin.py

@@ -7,6 +7,7 @@
 import socket
 import sys
 import warnings
+from textwrap import dedent
 from typing import (
     Any,
     AsyncIterator,
@@ -370,39 +371,22 @@ def _hypothesis_test_wraps_coroutine(function: Any) -> bool:
     return _is_coroutine(function.hypothesis.inner_test)
 
 
-@pytest.hookimpl(trylast=True)
-def pytest_fixture_post_finalizer(fixturedef: FixtureDef, request: SubRequest) -> None:
-    """
-    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:
-            loop = policy.get_event_loop()
-        except RuntimeError:
-            loop = None
-        if loop is not None:
-            # Clean up existing loop to avoid ResourceWarnings
-            loop.close()
-        # 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)
-
-
 @pytest.hookimpl(hookwrapper=True)
 def pytest_fixture_setup(
     fixturedef: FixtureDef, request: SubRequest
 ) -> Optional[object]:
     """Adjust the event loop policy when an event loop is produced."""
     if fixturedef.argname == "event_loop":
+        # The use of a fixture finalizer is preferred over the
+        # pytest_fixture_post_finalizer hook. The fixture finalizer is invoked once
+        # for each fixture, whereas the hook may be invoked multiple times for
+        # any specific fixture.
+        # see https://github.com/pytest-dev/pytest/issues/5848
+        _add_finalizers(
+            fixturedef,
+            _close_event_loop,
+            _provide_clean_event_loop,
+        )
         outcome = yield
         loop = outcome.get_result()
         policy = asyncio.get_event_loop_policy()
@@ -421,6 +405,61 @@ def pytest_fixture_setup(
     yield
 
 
+def _add_finalizers(fixturedef: FixtureDef, *finalizers: Callable[[], object]) -> None:
+    """
+    Regsiters the specified fixture finalizers in the fixture.
+
+    Finalizers need to specified in the exact order in which they should be invoked.
+
+    :param fixturedef: Fixture definition which finalizers should be added to
+    :param finalizers: Finalizers to be added
+    """
+    for finalizer in reversed(finalizers):
+        fixturedef.addfinalizer(finalizer)
+
+
+_UNCLOSED_EVENT_LOOP_WARNING = dedent(
+    """\
+    unclosed event loop %r.
+    Possible causes are:
+        1. A custom "event_loop" fixture is used which doesn't close the loop
+        2. Your code or one of your dependencies created a new event loop during
+           the test run
+    """
+)
+
+
+def _close_event_loop() -> None:
+    policy = asyncio.get_event_loop_policy()
+    try:
+        loop = policy.get_event_loop()
+    except RuntimeError:
+        loop = None
+    if loop is not None:
+        # Emit ResourceWarnings in the context of the fixture/test case
+        # rather than waiting for the interpreter to trigger the warning when
+        # garbage collecting the event loop.
+        if not loop.is_closed():
+            warnings.warn(
+                _UNCLOSED_EVENT_LOOP_WARNING % loop,
+                ResourceWarning,
+                source=loop,
+            )
+        loop.close()
+
+
+def _provide_clean_event_loop() -> None:
+    # 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.
+    policy = asyncio.get_event_loop_policy()
+    new_loop = policy.new_event_loop()
+    policy.set_event_loop(new_loop)
+
+
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_pyfunc_call(pyfuncitem: pytest.Function) -> Optional[object]:
     """

tests/async_fixtures/test_async_fixtures_scope.py

@@ -10,7 +10,9 @@
 @pytest.fixture(scope="module")
 def event_loop():
     """A module-scoped event loop."""
-    return asyncio.new_event_loop()
+    loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()
 
 
 @pytest.fixture(scope="module")

tests/hypothesis/test_base.py

@@ -1,18 +1,11 @@
 """Tests for the Hypothesis integration, which wraps async functions in a
 sync shim for Hypothesis.
 """
-import asyncio
 from textwrap import dedent
 
 import pytest
 from hypothesis import given, strategies as st
-
-
-@pytest.fixture(scope="module")
-def event_loop():
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
+from pytest import Pytester
 
 
 @given(st.integers())
@@ -35,16 +28,38 @@ async def test_mark_and_parametrize(x, y):
     assert y in (1, 2)
 
 
-@given(st.integers())
-@pytest.mark.asyncio
-async def test_can_use_fixture_provided_event_loop(event_loop, n):
-    semaphore = asyncio.Semaphore(value=0)
-    event_loop.call_soon(semaphore.release)
-    await semaphore.acquire()
+def test_can_use_explicit_event_loop_fixture(pytester: Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+            from hypothesis import given
+            import hypothesis.strategies as st
+
+            pytest_plugins = 'pytest_asyncio'
+
+            @pytest.fixture(scope="module")
+            def event_loop():
+                loop = asyncio.get_event_loop_policy().new_event_loop()
+                yield loop
+                loop.close()
+
+            @given(st.integers())
+            @pytest.mark.asyncio
+            async def test_explicit_fixture_request(event_loop, n):
+                semaphore = asyncio.Semaphore(value=0)
+                event_loop.call_soon(semaphore.release)
+                await semaphore.acquire()
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
 
 
-def test_async_auto_marked(testdir):
-    testdir.makepyfile(
+def test_async_auto_marked(pytester: Pytester):
+    pytester.makepyfile(
         dedent(
             """\
         import asyncio
@@ -60,13 +75,13 @@ async def test_hypothesis(n: int):
         """
         )
     )
-    result = testdir.runpytest("--asyncio-mode=auto")
+    result = pytester.runpytest("--asyncio-mode=auto")
     result.assert_outcomes(passed=1)
 
 
-def test_sync_not_auto_marked(testdir):
+def test_sync_not_auto_marked(pytester: Pytester):
     """Assert that synchronous Hypothesis functions are not marked with asyncio"""
-    testdir.makepyfile(
+    pytester.makepyfile(
         dedent(
             """\
         import asyncio
@@ -84,5 +99,5 @@ def test_hypothesis(request, n: int):
         """
         )
     )
-    result = testdir.runpytest("--asyncio-mode=auto")
+    result = pytester.runpytest("--asyncio-mode=auto")
     result.assert_outcomes(passed=1)

tests/test_event_loop_fixture.py

@@ -0,0 +1,53 @@
+from textwrap import dedent
+
+from pytest import Pytester
+
+
+def test_event_loop_fixture_respects_event_loop_policy(pytester: Pytester):
+    pytester.makeconftest(
+        dedent(
+            """\
+            '''Defines and sets a custom event loop policy'''
+            import asyncio
+            from asyncio import DefaultEventLoopPolicy, SelectorEventLoop
+
+            class TestEventLoop(SelectorEventLoop):
+                pass
+
+            class TestEventLoopPolicy(DefaultEventLoopPolicy):
+                def new_event_loop(self):
+                    return TestEventLoop()
+
+            # This statement represents a code which sets a custom event loop policy
+            asyncio.set_event_loop_policy(TestEventLoopPolicy())
+            """
+        )
+    )
+    pytester.makepyfile(
+        dedent(
+            """\
+            '''Tests that any externally provided event loop policy remains unaltered'''
+            import asyncio
+
+            import pytest
+
+
+            @pytest.mark.asyncio
+            async def test_uses_loop_provided_by_custom_policy():
+                '''Asserts that test cases use the event loop
+                provided by the custom event loop policy'''
+                assert type(asyncio.get_event_loop()).__name__ == "TestEventLoop"
+
+
+            @pytest.mark.asyncio
+            async def test_custom_policy_is_not_overwritten():
+                '''
+                Asserts that any custom event loop policy stays the same
+                across test cases.
+                '''
+                assert type(asyncio.get_event_loop()).__name__ == "TestEventLoop"
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)

tests/test_event_loop_fixture_finalizer.py

@@ -86,3 +86,31 @@ async def test_async_with_explicit_fixture_request(event_loop):
     )
     result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(passed=1)
+
+
+def test_event_loop_fixture_finalizer_raises_warning_when_loop_is_unclosed(
+    pytester: Pytester,
+):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+            import pytest_asyncio
+
+            pytest_plugins = 'pytest_asyncio'
+
+            @pytest.fixture
+            def event_loop():
+                loop = asyncio.get_event_loop_policy().new_event_loop()
+                yield loop
+
+            @pytest.mark.asyncio
+            async def test_ends_with_unclosed_loop():
+                pass
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict", "-W", "default")
+    result.assert_outcomes(passed=1, warnings=1)
+    result.stdout.fnmatch_lines("*unclosed event loop*")