PYTHON-5071 Cleanup

Author: ShaneHarvey

test/__init__.py

@@ -867,6 +867,28 @@ def max_message_size_bytes(self):
 # Reusable client context
 client_context = ClientContext()
 
+# Global event loop for async tests.
+LOOP = None
+
+
+def get_loop() -> asyncio.AbstractEventLoop:
+    """Get the test suite's global event loop."""
+    global LOOP
+    if LOOP is None:
+        try:
+            LOOP = asyncio.get_running_loop()
+        except RuntimeError:
+            # no running event loop, fallback to get_event_loop.
+            try:
+                # Ignore DeprecationWarning: There is no current event loop
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore", DeprecationWarning)
+                    LOOP = asyncio.get_event_loop()
+            except RuntimeError:
+                LOOP = asyncio.new_event_loop()
+                asyncio.set_event_loop(LOOP)
+    return LOOP
+
 
 class PyMongoTestCase(unittest.TestCase):
     if not _IS_SYNC:
@@ -878,7 +900,6 @@ def setUp(self):
         def tearDown(self):
             pass
 
-        # See TestCase.addCleanup.
         def addCleanup(self, func, /, *args, **kwargs):
             self.addCleanup(*(func, *args), **kwargs)
 
@@ -1214,31 +1235,10 @@ def tearDown(self) -> None:
         super().tearDown()
 
 
-LOOP = None
-
-
-def get_loop() -> asyncio.AbstractEventLoop:
-    global LOOP
-    if LOOP is None:
-        try:
-            LOOP = asyncio.get_running_loop()
-        except RuntimeError:
-            # no running event loop, fallback to get_event_loop.
-            try:
-                # Ignore DeprecationWarning: There is no current event loop
-                with warnings.catch_warnings():
-                    warnings.simplefilter("ignore", DeprecationWarning)
-                    LOOP = asyncio.get_event_loop()
-            except RuntimeError:
-                LOOP = asyncio.new_event_loop()
-                asyncio.set_event_loop(LOOP)
-    return LOOP
-
-
 def setup():
     if not _IS_SYNC:
-        global LOOP
-        LOOP = asyncio.get_running_loop()
+        # Set up the event loop.
+        get_loop()
     client_context.init()
     warnings.resetwarnings()
     warnings.simplefilter("always")

test/asynchronous/__init__.py

@@ -869,6 +869,28 @@ async def max_message_size_bytes(self):
 # Reusable client context
 async_client_context = AsyncClientContext()
 
+# Global event loop for async tests.
+LOOP = None
+
+
+def get_loop() -> asyncio.AbstractEventLoop:
+    """Get the test suite's global event loop."""
+    global LOOP
+    if LOOP is None:
+        try:
+            LOOP = asyncio.get_running_loop()
+        except RuntimeError:
+            # no running event loop, fallback to get_event_loop.
+            try:
+                # Ignore DeprecationWarning: There is no current event loop
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore", DeprecationWarning)
+                    LOOP = asyncio.get_event_loop()
+            except RuntimeError:
+                LOOP = asyncio.new_event_loop()
+                asyncio.set_event_loop(LOOP)
+    return LOOP
+
 
 class AsyncPyMongoTestCase(unittest.TestCase):
     if not _IS_SYNC:
@@ -880,7 +902,6 @@ async def asyncSetUp(self):
         async def asyncTearDown(self):
             pass
 
-        # See IsolatedAsyncioTestCase.addAsyncCleanup.
         def addAsyncCleanup(self, func, /, *args, **kwargs):
             self.addCleanup(*(func, *args), **kwargs)
 
@@ -1232,31 +1253,10 @@ async def asyncTearDown(self) -> None:
         await super().asyncTearDown()
 
 
-LOOP = None
-
-
-def get_loop() -> asyncio.AbstractEventLoop:
-    global LOOP
-    if LOOP is None:
-        try:
-            LOOP = asyncio.get_running_loop()
-        except RuntimeError:
-            # no running event loop, fallback to get_event_loop.
-            try:
-                # Ignore DeprecationWarning: There is no current event loop
-                with warnings.catch_warnings():
-                    warnings.simplefilter("ignore", DeprecationWarning)
-                    LOOP = asyncio.get_event_loop()
-            except RuntimeError:
-                LOOP = asyncio.new_event_loop()
-                asyncio.set_event_loop(LOOP)
-    return LOOP
-
-
 async def async_setup():
     if not _IS_SYNC:
-        global LOOP
-        LOOP = asyncio.get_running_loop()
+        # Set up the event loop.
+        get_loop()
     await async_client_context.init()
     warnings.resetwarnings()
     warnings.simplefilter("always")