Test with RawBSONDocument.
(self)
| 966 | self.assertEqual(change["fullDocument"], doc) |
| 967 | |
| 968 | def test_raw(self): |
| 969 | """Test with RawBSONDocument.""" |
| 970 | raw_coll = self.watched_collection(codec_options=DEFAULT_RAW_BSON_OPTIONS) |
| 971 | with raw_coll.watch() as change_stream: |
| 972 | raw_doc = RawBSONDocument(encode({"_id": 1})) |
| 973 | self.watched_collection().insert_one(raw_doc) |
| 974 | change = next(change_stream) |
| 975 | self.assertIsInstance(change, RawBSONDocument) |
| 976 | self.assertEqual(change["operationType"], "insert") |
| 977 | self.assertEqual(change["ns"]["db"], self.watched_collection().database.name) |
| 978 | self.assertEqual(change["ns"]["coll"], self.watched_collection().name) |
| 979 | self.assertEqual(change["fullDocument"], raw_doc) |
| 980 | |
| 981 | @client_context.require_version_min(4, 0) # Needed for start_at_operation_time. |
| 982 | def test_uuid_representations(self): |
nothing calls this directly
no test coverage detected