(self)
| 105 | self.initial_lsids = {s["id"] for s in session_ids(self.client)} |
| 106 | |
| 107 | async def asyncTearDown(self): |
| 108 | monitoring._SENSITIVE_COMMANDS.update(self.sensitive_commands) |
| 109 | await self.client.drop_database("pymongo_test") |
| 110 | used_lsids = self.initial_lsids.copy() |
| 111 | for event in self.session_checker_listener.started_events: |
| 112 | if "lsid" in event.command: |
| 113 | used_lsids.add(event.command["lsid"]["id"]) |
| 114 | |
| 115 | current_lsids = {s["id"] for s in session_ids(self.client)} |
| 116 | self.assertLessEqual(used_lsids, current_lsids) |
| 117 | |
| 118 | await super().asyncTearDown() |
| 119 | |
| 120 | async def _test_ops(self, client, *ops): |
| 121 | listener = client.options.event_listeners[0] |
nothing calls this directly
no test coverage detected