Fix error when a shared event loop is unset

changelog.d/1172.added.rst

@@ -0,0 +1 @@
+Pyright type checking support in tox configuration to improve type safety and compatibility.

changelog.d/1177.fixed.rst

@@ -0,0 +1 @@
+``RuntimeError: There is no current event loop in thread 'MainThread'`` when using shared event loops after any test unsets the event loop (such as when using ``asyncio.run`` and ``asyncio.Runner``).

pytest_asyncio/plugin.py

@@ -10,6 +10,7 @@
 import inspect
 import socket
 import sys
+import threading
 import traceback
 import warnings
 from asyncio import AbstractEventLoop, AbstractEventLoopPolicy
@@ -602,6 +603,12 @@ def _set_event_loop(loop: AbstractEventLoop | None) -> None:
         asyncio.set_event_loop(loop)
 
 
+def _reinstate_event_loop_on_main_thread() -> None:
+    if threading.current_thread() is threading.main_thread():
+        policy = _get_event_loop_policy()
+        policy.set_event_loop(policy.new_event_loop())
+
+
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_pyfunc_call(pyfuncitem: Function) -> object | None:
     """
@@ -663,7 +670,12 @@ def wrap_in_sync(
     @functools.wraps(func)
     def inner(*args, **kwargs):
         coro = func(*args, **kwargs)
-        _loop = _get_event_loop_no_warn()
+        try:
+            _loop = _get_event_loop_no_warn()
+        except RuntimeError:
+            # Handle situation where asyncio.set_event_loop(None) removes shared loops.
+            _reinstate_event_loop_on_main_thread()
+            _loop = _get_event_loop_no_warn()
         task = asyncio.ensure_future(coro, loop=_loop)
         try:
             _loop.run_until_complete(task)

tests/test_set_event_loop.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from textwrap import dedent
+
+import pytest
+from pytest import Pytester
+
+
+@pytest.mark.parametrize("test_loop_scope", ("function", "module", "session"))
+def test_set_event_loop_none(pytester: Pytester, test_loop_scope: str):
+    pytester.makeini(
+        dedent(
+            f"""\
+            [pytest]
+            asyncio_default_test_loop_scope = {test_loop_scope}
+            asyncio_default_fixture_loop_scope = function
+            """
+        )
+    )
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            pytest_plugins = "pytest_asyncio"
+
+            @pytest.mark.asyncio
+            async def test_before():
+                pass
+
+
+            def test_set_event_loop_none():
+                asyncio.set_event_loop(None)
+
+
+            @pytest.mark.asyncio
+            async def test_after():
+                pass
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess()
+    result.assert_outcomes(passed=3)
+
+
+def test_set_event_loop_none_class(pytester: Pytester):
+    pytester.makeini(
+        dedent(
+            """\
+            [pytest]
+            asyncio_default_test_loop_scope = class
+            asyncio_default_fixture_loop_scope = function
+            """
+        )
+    )
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            pytest_plugins = "pytest_asyncio"
+
+
+            class TestClass:
+                @pytest.mark.asyncio
+                async def test_before(self):
+                    pass
+
+
+                def test_set_event_loop_none(self):
+                    asyncio.set_event_loop(None)
+
+
+                @pytest.mark.asyncio
+                async def test_after(self):
+                    pass
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess()
+    result.assert_outcomes(passed=3)