(self, op)
| 1042 | self._test_reads(lambda coll, session: coll.estimated_document_count(session=session)) |
| 1043 | |
| 1044 | def _test_writes(self, op): |
| 1045 | coll = self.client.pymongo_test.test |
| 1046 | with self.client.start_session() as sess: |
| 1047 | op(coll, sess) |
| 1048 | operation_time = sess.operation_time |
| 1049 | self.assertIsNotNone(operation_time) |
| 1050 | self.listener.reset() |
| 1051 | coll.find_one({}, session=sess) |
| 1052 | act = ( |
| 1053 | self.listener.started_events[0] |
| 1054 | .command.get("readConcern", {}) |
| 1055 | .get("afterClusterTime") |
| 1056 | ) |
| 1057 | self.assertEqual(operation_time, act) |
| 1058 | |
| 1059 | @client_context.require_no_standalone |
| 1060 | def test_writes(self): |
no test coverage detected