[refactor] Extracted a context manager that sets a temporary event loop policy for the scope of the context.

Signed-off-by: Michael Seifert <[email protected]>

Author: seifertm

pytest_asyncio/plugin.py

@@ -590,21 +590,11 @@ def scoped_event_loop(
         event_loop_policy,
     ) -> Iterator[asyncio.AbstractEventLoop]:
         new_loop_policy = event_loop_policy
-        old_loop_policy = asyncio.get_event_loop_policy()
-        old_loop = asyncio.get_event_loop()
-        asyncio.set_event_loop_policy(new_loop_policy)
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        yield loop
-        loop.close()
-        asyncio.set_event_loop_policy(old_loop_policy)
-        # When a test uses both a scoped event loop and the event_loop fixture,
-        # the "_provide_clean_event_loop" finalizer of the event_loop fixture
-        # will already have installed a fresh event loop, in order to shield
-        # subsequent tests from side-effects. We close this loop before restoring
-        # the old loop to avoid ResourceWarnings.
-        asyncio.get_event_loop().close()
-        asyncio.set_event_loop(old_loop)
+        with _temporary_event_loop_policy(new_loop_policy):
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            yield loop
+            loop.close()
 
     # @pytest.fixture does not register the fixture anywhere, so pytest doesn't
     # know it exists. We work around this by attaching the fixture function to the
@@ -630,6 +620,24 @@ def _removesuffix(s: str, suffix: str) -> str:
     return s.removesuffix(suffix)
 
 
+@contextlib.contextmanager
+def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
+    old_loop_policy = asyncio.get_event_loop_policy()
+    old_loop = asyncio.get_event_loop()
+    asyncio.set_event_loop_policy(policy)
+    try:
+        yield
+    finally:
+        asyncio.set_event_loop_policy(old_loop_policy)
+        # When a test uses both a scoped event loop and the event_loop fixture,
+        # the "_provide_clean_event_loop" finalizer of the event_loop fixture
+        # will already have installed a fresh event loop, in order to shield
+        # subsequent tests from side-effects. We close this loop before restoring
+        # the old loop to avoid ResourceWarnings.
+        asyncio.get_event_loop().close()
+        asyncio.set_event_loop(old_loop)
+
+
 def pytest_collection_modifyitems(
     session: Session, config: Config, items: List[Item]
 ) -> None:
@@ -958,21 +966,11 @@ def _session_event_loop(
     request: FixtureRequest, event_loop_policy: AbstractEventLoopPolicy
 ) -> Iterator[asyncio.AbstractEventLoop]:
     new_loop_policy = event_loop_policy
-    old_loop_policy = asyncio.get_event_loop_policy()
-    old_loop = asyncio.get_event_loop()
-    asyncio.set_event_loop_policy(new_loop_policy)
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    yield loop
-    loop.close()
-    asyncio.set_event_loop_policy(old_loop_policy)
-    # When a test uses both a scoped event loop and the event_loop fixture,
-    # the "_provide_clean_event_loop" finalizer of the event_loop fixture
-    # will already have installed a fresh event loop, in order to shield
-    # subsequent tests from side-effects. We close this loop before restoring
-    # the old loop to avoid ResourceWarnings.
-    asyncio.get_event_loop().close()
-    asyncio.set_event_loop(old_loop)
+    with _temporary_event_loop_policy(new_loop_policy):
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        yield loop
+        loop.close()
 
 
 @pytest.fixture(scope="session", autouse=True)