update discovery and monitoring based on #1925

Author: sleepyStick

test/asynchronous/test_discovery_and_monitoring.py

@@ -258,7 +258,7 @@ class TestClusterTimeComparison(AsyncPyMongoTestCase):
     async def test_cluster_time_comparison(self):
         t = await create_mock_topology("mongodb://host")
 
-        async def send_cluster_time(time, inc, should_update):
+        async def send_cluster_time(time, inc):
             old = t.max_cluster_time()
             new = {"clusterTime": Timestamp(time, inc)}
             await got_hello(
@@ -273,16 +273,14 @@ async def send_cluster_time(time, inc, should_update):
             )
 
             actual = t.max_cluster_time()
-            if should_update:
-                self.assertEqual(actual, new)
-            else:
-                self.assertEqual(actual, old)
-
-        await send_cluster_time(0, 1, True)
-        await send_cluster_time(2, 2, True)
-        await send_cluster_time(2, 1, False)
-        await send_cluster_time(1, 3, False)
-        await send_cluster_time(2, 3, True)
+            # We never update $clusterTime from monitoring connections.
+            self.assertEqual(actual, old)
+
+        await send_cluster_time(0, 1)
+        await send_cluster_time(2, 2)
+        await send_cluster_time(2, 1)
+        await send_cluster_time(1, 3)
+        await send_cluster_time(2, 3)
 
 
 class TestIgnoreStaleErrors(AsyncIntegrationTest):

test/asynchronous/unified_format.py

@@ -1385,20 +1385,18 @@ async def run_scenario(self, spec, uri=None):
         # operations during test set up and tear down.
         await self.kill_all_sessions()
 
-        if "csot" in self.id().lower() or "discovery_and_monitoring" in self.id().lower():
+        if "csot" in self.id().lower():
             # Retry CSOT tests up to 2 times to deal with flakey tests.
             # discovery_and_monitoring tests on windows are also flakey
             attempts = 3
             for i in range(attempts):
                 try:
                     return await self._run_scenario(spec, uri)
-                except (AssertionError, OperationFailure, _OperationCancelled) as exc:
+                except (AssertionError, OperationFailure) as exc:
                     if isinstance(exc, OperationFailure) and (
                         _IS_SYNC or "failpoint" not in exc._message
                     ):
                         raise
-                    if isinstance(exc, _OperationCancelled) and _IS_SYNC:
-                        raise
                     if i < attempts - 1:
                         print(
                             f"Retrying after attempt {i+1} of {self.id()} failed with:\n"

test/unified_format.py

@@ -1372,20 +1372,18 @@ def run_scenario(self, spec, uri=None):
         # operations during test set up and tear down.
         self.kill_all_sessions()
 
-        if "csot" in self.id().lower() or "discovery_and_monitoring" in self.id().lower():
+        if "csot" in self.id().lower():
             # Retry CSOT tests up to 2 times to deal with flakey tests.
             # discovery_and_monitoring tests on windows are also flakey
             attempts = 3
             for i in range(attempts):
                 try:
                     return self._run_scenario(spec, uri)
-                except (AssertionError, OperationFailure, _OperationCancelled) as exc:
+                except (AssertionError, OperationFailure) as exc:
                     if isinstance(exc, OperationFailure) and (
                         _IS_SYNC or "failpoint" not in exc._message
                     ):
                         raise
-                    if isinstance(exc, _OperationCancelled) and _IS_SYNC:
-                        raise
                     if i < attempts - 1:
                         print(
                             f"Retrying after attempt {i+1} of {self.id()} failed with:\n"