Run a CMAP spec test.
(self, scenario_def, test)
| 215 | self.configure_fail_point(self.client, command_args) |
| 216 | |
| 217 | def run_scenario(self, scenario_def, test): |
| 218 | """Run a CMAP spec test.""" |
| 219 | self.logs: list = [] |
| 220 | self.assertEqual(scenario_def["version"], 1) |
| 221 | self.assertIn(scenario_def["style"], ["unit", "integration"]) |
| 222 | self.listener = CMAPListener() |
| 223 | self._ops: list = [] |
| 224 | |
| 225 | # Configure the fail point before creating the client. |
| 226 | if "failPoint" in test: |
| 227 | fp = test["failPoint"] |
| 228 | self.set_fail_point(fp) |
| 229 | self.addCleanup( |
| 230 | self.set_fail_point, {"configureFailPoint": fp["configureFailPoint"], "mode": "off"} |
| 231 | ) |
| 232 | |
| 233 | opts = test["poolOptions"].copy() |
| 234 | opts["event_listeners"] = [self.listener] |
| 235 | opts["_monitor_class"] = DummyMonitor |
| 236 | opts["connect"] = False |
| 237 | # Support backgroundThreadIntervalMS, default to 50ms. |
| 238 | interval = opts.pop("backgroundThreadIntervalMS", 50) |
| 239 | if interval < 0: |
| 240 | kill_cursor_frequency = 99999999 |
| 241 | else: |
| 242 | kill_cursor_frequency = interval / 1000.0 |
| 243 | with client_knobs(kill_cursor_frequency=kill_cursor_frequency, min_heartbeat_interval=0.05): |
| 244 | client = self.single_client(**opts) |
| 245 | # Update the SD to a known type because the DummyMonitor will not. |
| 246 | # Note we cannot simply call topology.on_change because that would |
| 247 | # internally call pool.ready() which introduces unexpected |
| 248 | # PoolReadyEvents. Instead, update the initial state before |
| 249 | # opening the Topology. |
| 250 | td = client_context.client._topology.description |
| 251 | sd = td.server_descriptions()[(client_context.host, client_context.port)] |
| 252 | client._topology._description = updated_topology_description( |
| 253 | client._topology._description, sd |
| 254 | ) |
| 255 | # When backgroundThreadIntervalMS is negative we do not start the |
| 256 | # background thread to ensure it never runs. |
| 257 | if interval < 0: |
| 258 | client._topology.open() |
| 259 | else: |
| 260 | client._get_topology() |
| 261 | self.pool = list(client._topology._servers.values())[0].pool |
| 262 | |
| 263 | # Map of target names to Thread objects. |
| 264 | self.targets: dict = {} |
| 265 | # Map of label names to Connection objects |
| 266 | self.labels: dict = {} |
| 267 | |
| 268 | def cleanup(): |
| 269 | for t in self.targets.values(): |
| 270 | t.stop() |
| 271 | for t in self.targets.values(): |
| 272 | t.join(5) |
| 273 | for conn in self.labels.values(): |
| 274 | conn.close_conn(None) |
no test coverage detected