PYTHON-5202 Advance clusterTime without using ping command

Author: ShaneHarvey

test/asynchronous/unified_format.py

@@ -222,7 +222,6 @@ def __init__(self, test_class):
         self._listeners: Dict[str, EventListenerUtil] = {}
         self._session_lsids: Dict[str, Mapping[str, Any]] = {}
         self.test: UnifiedSpecTestMixinV1 = test_class
-        self._cluster_time: Mapping[str, Any] = {}
 
     def __contains__(self, item):
         return item in self._entities
@@ -421,13 +420,11 @@ def get_lsid_for_session(self, session_name):
             # session has been closed.
             return self._session_lsids[session_name]
 
-    async def advance_cluster_times(self) -> None:
+    async def advance_cluster_times(self, cluster_time) -> None:
         """Manually synchronize entities when desired"""
-        if not self._cluster_time:
-            self._cluster_time = (await self.test.client.admin.command("ping")).get("$clusterTime")
         for entity in self._entities.values():
-            if isinstance(entity, AsyncClientSession) and self._cluster_time:
-                entity.advance_cluster_time(self._cluster_time)
+            if isinstance(entity, AsyncClientSession) and cluster_time:
+                entity.advance_cluster_time(cluster_time)
 
 
 class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
@@ -1044,7 +1041,7 @@ async def _testOperation_targetedFailPoint(self, spec):
 
     async def _testOperation_createEntities(self, spec):
         await self.entity_map.create_entities_from_spec(spec["entities"], uri=self._uri)
-        await self.entity_map.advance_cluster_times()
+        await self.entity_map.advance_cluster_times(self._cluster_time)
 
     def _testOperation_assertSessionTransactionState(self, spec):
         session = self.entity_map[spec["session"]]
@@ -1446,8 +1443,8 @@ async def _run_scenario(self, spec, uri=None):
         # process initialData
         if "initialData" in self.TEST_SPEC:
             await self.insert_initial_data(self.TEST_SPEC["initialData"])
-            self._cluster_time = (await self.client.admin.command("ping")).get("$clusterTime")
-            await self.entity_map.advance_cluster_times()
+            self._cluster_time = self.client._topology.max_cluster_time()
+            await self.entity_map.advance_cluster_times(self._cluster_time)
 
         if "expectLogMessages" in spec:
             expect_log_messages = spec["expectLogMessages"]

test/unified_format.py

@@ -221,7 +221,6 @@ def __init__(self, test_class):
         self._listeners: Dict[str, EventListenerUtil] = {}
         self._session_lsids: Dict[str, Mapping[str, Any]] = {}
         self.test: UnifiedSpecTestMixinV1 = test_class
-        self._cluster_time: Mapping[str, Any] = {}
 
     def __contains__(self, item):
         return item in self._entities
@@ -420,13 +419,11 @@ def get_lsid_for_session(self, session_name):
             # session has been closed.
             return self._session_lsids[session_name]
 
-    def advance_cluster_times(self) -> None:
+    def advance_cluster_times(self, cluster_time) -> None:
         """Manually synchronize entities when desired"""
-        if not self._cluster_time:
-            self._cluster_time = (self.test.client.admin.command("ping")).get("$clusterTime")
         for entity in self._entities.values():
-            if isinstance(entity, ClientSession) and self._cluster_time:
-                entity.advance_cluster_time(self._cluster_time)
+            if isinstance(entity, ClientSession) and cluster_time:
+                entity.advance_cluster_time(cluster_time)
 
 
 class UnifiedSpecTestMixinV1(IntegrationTest):
@@ -1035,7 +1032,7 @@ def _testOperation_targetedFailPoint(self, spec):
 
     def _testOperation_createEntities(self, spec):
         self.entity_map.create_entities_from_spec(spec["entities"], uri=self._uri)
-        self.entity_map.advance_cluster_times()
+        self.entity_map.advance_cluster_times(self._cluster_time)
 
     def _testOperation_assertSessionTransactionState(self, spec):
         session = self.entity_map[spec["session"]]
@@ -1431,8 +1428,8 @@ def _run_scenario(self, spec, uri=None):
         # process initialData
         if "initialData" in self.TEST_SPEC:
             self.insert_initial_data(self.TEST_SPEC["initialData"])
-            self._cluster_time = (self.client.admin.command("ping")).get("$clusterTime")
-            self.entity_map.advance_cluster_times()
+            self._cluster_time = self.client._topology.max_cluster_time()
+            self.entity_map.advance_cluster_times(self._cluster_time)
 
         if "expectLogMessages" in spec:
             expect_log_messages = spec["expectLogMessages"]