(self)
| 151 | knobs: client_knobs |
| 152 | |
| 153 | def setUp(self) -> None: |
| 154 | super().setUp() |
| 155 | # Speed up the tests by decreasing the heartbeat frequency. |
| 156 | self.knobs = client_knobs(heartbeat_frequency=0.1, min_heartbeat_interval=0.1) |
| 157 | self.knobs.enable() |
| 158 | self.listener = OvertCommandListener() |
| 159 | self.client = self.rs_or_single_client(retryWrites=True, event_listeners=[self.listener]) |
| 160 | self.db = self.client.pymongo_test |
| 161 | |
| 162 | if client_context.is_rs and client_context.test_commands_enabled: |
| 163 | self.client.admin.command( |
| 164 | SON([("configureFailPoint", "onPrimaryTransactionalWrite"), ("mode", "alwaysOn")]) |
| 165 | ) |
| 166 | |
| 167 | def tearDown(self): |
| 168 | if client_context.is_rs and client_context.test_commands_enabled: |
no test coverage detected