(self, op, exception=None)
| 998 | self.assertEqual(sess.operation_time, sess2.operation_time) |
| 999 | |
| 1000 | def _test_reads(self, op, exception=None): |
| 1001 | coll = self.client.pymongo_test.test |
| 1002 | with self.client.start_session() as sess: |
| 1003 | coll.find_one({}, session=sess) |
| 1004 | operation_time = sess.operation_time |
| 1005 | self.assertIsNotNone(operation_time) |
| 1006 | self.listener.reset() |
| 1007 | if exception: |
| 1008 | with self.assertRaises(exception): |
| 1009 | op(coll, sess) |
| 1010 | else: |
| 1011 | op(coll, sess) |
| 1012 | act = ( |
| 1013 | self.listener.started_events[0] |
| 1014 | .command.get("readConcern", {}) |
| 1015 | .get("afterClusterTime") |
| 1016 | ) |
| 1017 | self.assertEqual(operation_time, act) |
| 1018 | |
| 1019 | @client_context.require_no_standalone |
| 1020 | def test_reads(self): |
no test coverage detected