PYTHON-5078 Convert test.test_discovery_and_monitoring to async

test/asynchronous/helpers.py

@@ -408,7 +408,7 @@ async def run(self):
         finally:
             self.stopped = True
 
-
+            
 class ExceptionCatchingTask(ConcurrentRunner):
     """A Task that stores any exception encountered while running."""
 
@@ -422,3 +422,20 @@ async def run(self):
         except BaseException as exc:
             self.exc = exc
             raise
+
+
+def create_barrier(N_TASKS, timeout: float | None = None):
+    return threading.Barrier(N_TASKS, timeout)
+
+
+def async_create_barrier(N_TASKS, timeout: float | None = None):
+    return asyncio.Barrier(N_TASKS)
+
+
+def barrier_wait(barrier, timeout: float | None = None):
+    barrier.wait()
+
+
+async def async_barrier_wait(barrier, timeout: float | None = None):
+    await asyncio.wait_for(barrier.wait(), timeout)
+

test/asynchronous/test_discovery_and_monitoring.py

@@ -22,6 +22,7 @@
 import threading
 from asyncio import StreamReader, StreamWriter
 from pathlib import Path
+from test.asynchronous.helpers import ConcurrentRunner, async_barrier_wait, async_create_barrier
 
 sys.path[0:0] = [""]
 
@@ -275,84 +276,44 @@ async def send_cluster_time(time, inc, should_update):
 
 
 class TestIgnoreStaleErrors(AsyncIntegrationTest):
-    if _IS_SYNC:
-
-        async def test_ignore_stale_connection_errors(self):
-            N_THREADS = 5
-            barrier = threading.Barrier(N_THREADS, timeout=30)
-            client = await self.async_rs_or_single_client(minPoolSize=N_THREADS)
-
-            # Wait for initial discovery.
-            await client.admin.command("ping")
-            pool = await async_get_pool(client)
-            starting_generation = pool.gen.get_overall()
-            await async_wait_until(lambda: len(pool.conns) == N_THREADS, "created conns")
-
-            def mock_command(*args, **kwargs):
-                # Synchronize all threads to ensure they use the same generation.
-                barrier.wait()
-                raise AutoReconnect("mock AsyncConnection.command error")
-
-            for conn in pool.conns:
-                conn.command = mock_command
-
-            async def insert_command(i):
-                try:
-                    await client.test.command("insert", "test", documents=[{"i": i}])
-                except AutoReconnect:
-                    pass
-
-            threads = []
-            for i in range(N_THREADS):
-                threads.append(threading.Thread(target=insert_command, args=(i,)))
-            for t in threads:
-                t.start()
-            for t in threads:
-                t.join()
-
-            # Expect a single pool reset for the network error
-            self.assertEqual(starting_generation + 1, pool.gen.get_overall())
-
-            # Server should be selectable.
-            await client.admin.command("ping")
-    else:
-
-        async def test_ignore_stale_connection_errors(self):
-            N_TASKS = 5
-            barrier = asyncio.Barrier(N_TASKS)
-            client = await self.async_rs_or_single_client(minPoolSize=N_TASKS)
-
-            # Wait for initial discovery.
-            await client.admin.command("ping")
-            pool = await async_get_pool(client)
-            starting_generation = pool.gen.get_overall()
-            await async_wait_until(lambda: len(pool.conns) == N_TASKS, "created conns")
-
-            async def mock_command(*args, **kwargs):
-                # Synchronize all threads to ensure they use the same generation.
-                await asyncio.wait_for(barrier.wait(), timeout=30)
-                raise AutoReconnect("mock AsyncConnection.command error")
-
-            for conn in pool.conns:
-                conn.command = mock_command
+    async def test_ignore_stale_connection_errors(self):
+        N_TASKS = 5
+        barrier = async_create_barrier(N_TASKS, timeout=30)
+        client = await self.async_rs_or_single_client(minPoolSize=N_TASKS)
 
-            async def insert_command(i):
-                try:
-                    await client.test.command("insert", "test", documents=[{"i": i}])
-                except AutoReconnect:
-                    pass
-
-            tasks = []
-            for i in range(N_TASKS):
-                tasks.append(asyncio.create_task(insert_command(i)))
-            for t in tasks:
-                await t
-
-            # Expect a single pool reset for the network error
-            self.assertEqual(starting_generation + 1, pool.gen.get_overall())
-
-            # Server should be selectable.
-            await client.admin.command("ping")
+        # Wait for initial discovery.
+        await client.admin.command("ping")
+        pool = await async_get_pool(client)
+        starting_generation = pool.gen.get_overall()
+        await async_wait_until(lambda: len(pool.conns) == N_TASKS, "created conns")
+
+        async def mock_command(*args, **kwargs):
+            # Synchronize all threads to ensure they use the same generation.
+            await async_barrier_wait(barrier, timeout=30)
+            raise AutoReconnect("mock AsyncConnection.command error")
+
+        for conn in pool.conns:
+            conn.command = mock_command
+
+        async def insert_command(i):
+            try:
+                await client.test.command("insert", "test", documents=[{"i": i}])
+            except AutoReconnect:
+                pass
+
+        tasks = []
+        for i in range(N_TASKS):
+            tasks.append(ConcurrentRunner(target=insert_command, args=(i,)))
+        for t in tasks:
+            await t.start()
+        for t in tasks:
+            await t.join()
+
+        # Expect a single pool reset for the network error
+        self.assertEqual(starting_generation + 1, pool.gen.get_overall())
+
+        # Server should be selectable.
+        await client.admin.command("ping")
 
 
 class CMAPHeartbeatListener(HeartbeatEventListener, CMAPListener):
@@ -499,14 +460,12 @@ async def handle_client(reader: StreamReader, writer: StreamWriter):
             server = await asyncio.start_server(handle_client, "localhost", 9999)
             server.events = events
             await server.start_serving()
-            print(server.is_serving())
             _c = self.simple_client(
                 "mongodb://localhost:9999",
                 serverSelectionTimeoutMS=500,
                 event_listeners=(listener,),
             )
-            if _c._options.connect:
-                await _c.aconnect()
+            await _c.aconnect()
 
             await listener.async_wait_for_event(ServerHeartbeatStartedEvent, 1)
             await listener.async_wait_for_event(ServerHeartbeatFailedEvent, 1)

test/helpers.py

@@ -422,3 +422,20 @@ def run(self):
         except BaseException as exc:
             self.exc = exc
             raise
+
+
+def create_barrier(N_TASKS, timeout: float | None = None):
+    return threading.Barrier(N_TASKS, timeout)
+
+
+def create_barrier(N_TASKS, timeout: float | None = None):
+    return asyncio.Barrier(N_TASKS)
+
+
+def barrier_wait(barrier, timeout: float | None = None):
+    barrier.wait()
+
+
+def barrier_wait(barrier, timeout: float | None = None):
+    asyncio.wait_for(barrier.wait(), timeout)
+

test/test_discovery_and_monitoring.py

@@ -22,6 +22,7 @@
 import threading
 from asyncio import StreamReader, StreamWriter
 from pathlib import Path
+from test.helpers import ConcurrentRunner, barrier_wait, create_barrier
 
 sys.path[0:0] = [""]
 
@@ -275,84 +276,44 @@ def send_cluster_time(time, inc, should_update):
 
 
 class TestIgnoreStaleErrors(IntegrationTest):
-    if _IS_SYNC:
-
-        def test_ignore_stale_connection_errors(self):
-            N_THREADS = 5
-            barrier = threading.Barrier(N_THREADS, timeout=30)
-            client = self.rs_or_single_client(minPoolSize=N_THREADS)
-
-            # Wait for initial discovery.
-            client.admin.command("ping")
-            pool = get_pool(client)
-            starting_generation = pool.gen.get_overall()
-            wait_until(lambda: len(pool.conns) == N_THREADS, "created conns")
-
-            def mock_command(*args, **kwargs):
-                # Synchronize all threads to ensure they use the same generation.
-                barrier.wait()
-                raise AutoReconnect("mock Connection.command error")
-
-            for conn in pool.conns:
-                conn.command = mock_command
-
-            def insert_command(i):
-                try:
-                    client.test.command("insert", "test", documents=[{"i": i}])
-                except AutoReconnect:
-                    pass
-
-            threads = []
-            for i in range(N_THREADS):
-                threads.append(threading.Thread(target=insert_command, args=(i,)))
-            for t in threads:
-                t.start()
-            for t in threads:
-                t.join()
-
-            # Expect a single pool reset for the network error
-            self.assertEqual(starting_generation + 1, pool.gen.get_overall())
-
-            # Server should be selectable.
-            client.admin.command("ping")
-    else:
-
-        def test_ignore_stale_connection_errors(self):
-            N_TASKS = 5
-            barrier = asyncio.Barrier(N_TASKS)
-            client = self.rs_or_single_client(minPoolSize=N_TASKS)
-
-            # Wait for initial discovery.
-            client.admin.command("ping")
-            pool = get_pool(client)
-            starting_generation = pool.gen.get_overall()
-            wait_until(lambda: len(pool.conns) == N_TASKS, "created conns")
-
-            def mock_command(*args, **kwargs):
-                # Synchronize all threads to ensure they use the same generation.
-                asyncio.wait_for(barrier.wait(), timeout=30)
-                raise AutoReconnect("mock Connection.command error")
-
-            for conn in pool.conns:
-                conn.command = mock_command
+    def test_ignore_stale_connection_errors(self):
+        N_TASKS = 5
+        barrier = create_barrier(N_TASKS, timeout=30)
+        client = self.rs_or_single_client(minPoolSize=N_TASKS)
 
-            def insert_command(i):
-                try:
-                    client.test.command("insert", "test", documents=[{"i": i}])
-                except AutoReconnect:
-                    pass
-
-            tasks = []
-            for i in range(N_TASKS):
-                tasks.append(asyncio.create_task(insert_command(i)))
-            for t in tasks:
-                t
-
-            # Expect a single pool reset for the network error
-            self.assertEqual(starting_generation + 1, pool.gen.get_overall())
-
-            # Server should be selectable.
-            client.admin.command("ping")
+        # Wait for initial discovery.
+        client.admin.command("ping")
+        pool = get_pool(client)
+        starting_generation = pool.gen.get_overall()
+        wait_until(lambda: len(pool.conns) == N_TASKS, "created conns")
+
+        def mock_command(*args, **kwargs):
+            # Synchronize all threads to ensure they use the same generation.
+            barrier_wait(barrier, timeout=30)
+            raise AutoReconnect("mock Connection.command error")
+
+        for conn in pool.conns:
+            conn.command = mock_command
+
+        def insert_command(i):
+            try:
+                client.test.command("insert", "test", documents=[{"i": i}])
+            except AutoReconnect:
+                pass
+
+        tasks = []
+        for i in range(N_TASKS):
+            tasks.append(ConcurrentRunner(target=insert_command, args=(i,)))
+        for t in tasks:
+            t.start()
+        for t in tasks:
+            t.join()
+
+        # Expect a single pool reset for the network error
+        self.assertEqual(starting_generation + 1, pool.gen.get_overall())
+
+        # Server should be selectable.
+        client.admin.command("ping")
 
 
 class CMAPHeartbeatListener(HeartbeatEventListener, CMAPListener):
@@ -499,14 +460,12 @@ def handle_client(reader: StreamReader, writer: StreamWriter):
             server = asyncio.start_server(handle_client, "localhost", 9999)
             server.events = events
             server.start_serving()
-            print(server.is_serving())
             _c = self.simple_client(
                 "mongodb://localhost:9999",
                 serverSelectionTimeoutMS=500,
                 event_listeners=(listener,),
             )
-            if _c._options.connect:
-                _c._connect()
+            _c._connect()
 
             listener.wait_for_event(ServerHeartbeatStartedEvent, 1)
             listener.wait_for_event(ServerHeartbeatFailedEvent, 1)

tools/synchro.py

@@ -123,6 +123,8 @@
     "AsyncMockConnection": "MockConnection",
     "AsyncMockPool": "MockPool",
     "create_async_event": "create_event",
+    "async_create_barrier": "create_barrier",
+    "async_barrier_wait": "barrier_wait",
 }
 
 docstring_replacements: dict[tuple[str, str], str] = {