diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index f663ff68..21809b6d 100644 --- a/pytest_asyncio/plugin.py +++ b/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) diff --git a/tests/test_event_loop_fixture_finalizer.py b/tests/test_event_loop_fixture_finalizer.py new file mode 100644 index 00000000..d3622a08 --- /dev/null +++ b/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) diff --git a/tests/test_event_loop_scope.py b/tests/test_event_loop_scope.py deleted file mode 100644 index 21fd6415..00000000 --- a/tests/test_event_loop_scope.py +++ /dev/null @@ -1,37 +0,0 @@ -"""Test the event loop fixture provides a separate loop for each test. - -These tests need to be run together. -""" -import asyncio - -import pytest - -loop: asyncio.AbstractEventLoop - - -def test_1(): - global loop - # The main thread should have a default event loop. - loop = asyncio.get_event_loop_policy().get_event_loop() - - -@pytest.mark.asyncio -async def test_2(): - 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 - loop = running_loop # Store the loop reference for later - - -def test_3(): - global loop - current_loop = asyncio.get_event_loop_policy().get_event_loop() - # Now the event loop from test_2 should have been cleaned up - assert loop is not current_loop - - -def test_4(event_loop): - # If a test sets the loop to None -- pytest_fixture_post_finalizer() - # still should work - asyncio.get_event_loop_policy().set_event_loop(None)