Merge resolution

Author: NoahStapp

test/asynchronous/test_discovery_and_monitoring.py

@@ -26,25 +26,32 @@
 
 sys.path[0:0] = [""]
 
-from test.asynchronous import AsyncIntegrationTest, AsyncPyMongoTestCase, AsyncUnitTest, unittest
+from test.asynchronous import (
+    AsyncIntegrationTest,
+    AsyncPyMongoTestCase,
+    AsyncUnitTest,
+    async_client_context,
+    unittest,
+)
 from test.asynchronous.pymongo_mocks import DummyMonitor
 from test.asynchronous.unified_format import generate_test_classes
-from test.utils import (
+from test.asynchronous.utils import (
+    async_get_pool,
+)
+from test.utils_shared import (
     CMAPListener,
     HeartbeatEventListener,
     HeartbeatEventsListListener,
     assertion_context,
     async_barrier_wait,
-    async_client_context,
     async_create_barrier,
-    async_get_pool,
     async_wait_until,
     server_name_to_type,
 )
 from unittest.mock import patch
 
 from bson import Timestamp, json_util
-from pymongo import AsyncMongoClient, common, monitoring
+from pymongo import common, monitoring
 from pymongo.asynchronous.settings import TopologySettings
 from pymongo.asynchronous.topology import Topology, _ErrorContext
 from pymongo.errors import (
@@ -291,7 +298,7 @@ async def test_ignore_stale_connection_errors(self):
         if not _IS_SYNC and sys.version_info < (3, 11):
             self.skipTest("Test requires asyncio.Barrier (added in Python 3.11)")
         N_TASKS = 5
-        barrier = async_create_barrier(N_TASKS, timeout=30)
+        barrier = async_create_barrier(N_TASKS)
         client = await self.async_rs_or_single_client(minPoolSize=N_TASKS)
 
         # Wait for initial discovery.

test/asynchronous/test_monitor.py

@@ -25,10 +25,10 @@
 sys.path[0:0] = [""]
 
 from test.asynchronous import AsyncIntegrationTest, async_client_context, connected, unittest
-from test.utils import (
-    ServerAndTopologyEventListener,
+from test.asynchronous.utils import (
     async_wait_until,
 )
+from test.utils_shared import ServerAndTopologyEventListener
 
 from pymongo.periodic_executor import _EXECUTORS
 

test/asynchronous/test_session.py

@@ -30,7 +30,6 @@
 
 from test.asynchronous import (
     AsyncIntegrationTest,
-    AsyncPyMongoTestCase,
     AsyncUnitTest,
     SkipTest,
     async_client_context,

test/asynchronous/utils.py

@@ -18,6 +18,7 @@
 import asyncio
 import contextlib
 import random
+import threading  # Used in the synchronized version of this file
 import time
 from asyncio import iscoroutinefunction
 

test/test_discovery_and_monitoring.py

@@ -20,31 +20,38 @@
 import socketserver
 import sys
 import threading
-from test.utils import get_pool
 from asyncio import StreamReader, StreamWriter
 from pathlib import Path
 from test.helpers import ConcurrentRunner
 
 sys.path[0:0] = [""]
 
-from test import IntegrationTest, PyMongoTestCase, UnitTest, unittest
+from test import (
+    IntegrationTest,
+    PyMongoTestCase,
+    UnitTest,
+    client_context,
+    unittest,
+)
 from test.pymongo_mocks import DummyMonitor
 from test.unified_format import generate_test_classes
+from test.utils import (
+    get_pool,
+)
 from test.utils_shared import (
     CMAPListener,
     HeartbeatEventListener,
     HeartbeatEventsListListener,
     assertion_context,
     barrier_wait,
-    client_context,
     create_barrier,
     server_name_to_type,
     wait_until,
 )
 from unittest.mock import patch
 
 from bson import Timestamp, json_util
-from pymongo import MongoClient, common, monitoring
+from pymongo import common, monitoring
 from pymongo.errors import (
     AutoReconnect,
     ConfigurationError,
@@ -291,7 +298,7 @@ def test_ignore_stale_connection_errors(self):
         if not _IS_SYNC and sys.version_info < (3, 11):
             self.skipTest("Test requires asyncio.Barrier (added in Python 3.11)")
         N_TASKS = 5
-        barrier = create_barrier(N_TASKS, timeout=30)
+        barrier = create_barrier(N_TASKS)
         client = self.rs_or_single_client(minPoolSize=N_TASKS)
 
         # Wait for initial discovery.

test/test_monitor.py

@@ -24,13 +24,11 @@
 
 sys.path[0:0] = [""]
 
-from test import IntegrationTest, connected, unittest
-from test.utils_shared import (
 from test import IntegrationTest, client_context, connected, unittest
 from test.utils import (
-    ServerAndTopologyEventListener,
     wait_until,
 )
+from test.utils_shared import ServerAndTopologyEventListener
 
 from pymongo.periodic_executor import _EXECUTORS
 

test/test_session.py

@@ -30,15 +30,13 @@
 
 from test import (
     IntegrationTest,
-    PyMongoTestCase,
     SkipTest,
     UnitTest,
     client_context,
     unittest,
 )
-from test.utils_shared import (
 from test.helpers import client_knobs
-from test.utils import (
+from test.utils_shared import (
     EventListener,
     HeartbeatEventListener,
     OvertCommandListener,

test/utils.py

@@ -18,6 +18,7 @@
 import asyncio
 import contextlib
 import random
+import threading  # Used in the synchronized version of this file
 import time
 from asyncio import iscoroutinefunction
 
@@ -206,21 +207,3 @@ def update_is_writable(self, is_writable):
 
     def remove_stale_sockets(self, *args, **kwargs):
         pass
-def create_event():
-    return threading.Event()
-
-
-def async_create_barrier(N_TASKS, timeout: float | None = None):
-    return asyncio.Barrier(N_TASKS)
-
-
-def create_barrier(N_TASKS, timeout: float | None = None):
-    return threading.Barrier(N_TASKS, timeout=timeout)
-
-
-async def async_barrier_wait(barrier, timeout: float | None = None):
-    await asyncio.wait_for(barrier.wait(), timeout=timeout)
-
-
-def barrier_wait(barrier, timeout: float | None = None):
-    barrier.wait()

test/utils_shared.py

@@ -687,3 +687,19 @@ def create_async_event():
 
 def create_event():
     return threading.Event()
+
+
+def async_create_barrier(n_tasks: int):
+    return asyncio.Barrier(n_tasks)
+
+
+def create_barrier(n_tasks: int, timeout: float | None = None):
+    return threading.Barrier(n_tasks, timeout=timeout)
+
+
+async def async_barrier_wait(barrier, timeout: float | None = None):
+    await asyncio.wait_for(barrier.wait(), timeout=timeout)
+
+
+def barrier_wait(barrier, timeout: float | None = None):
+    barrier.wait(timeout=timeout)