(self)
| 134 | class APITestsMixin: |
| 135 | @no_type_check |
| 136 | def test_watch(self): |
| 137 | with self.change_stream( |
| 138 | [{"$project": {"foo": 0}}], |
| 139 | full_document="updateLookup", |
| 140 | max_await_time_ms=1000, |
| 141 | batch_size=100, |
| 142 | ) as change_stream: |
| 143 | self.assertEqual([{"$project": {"foo": 0}}], change_stream._pipeline) |
| 144 | self.assertEqual("updateLookup", change_stream._full_document) |
| 145 | self.assertEqual(1000, change_stream._max_await_time_ms) |
| 146 | self.assertEqual(100, change_stream._batch_size) |
| 147 | self.assertIsInstance(change_stream._cursor, CommandCursor) |
| 148 | self.assertEqual(1000, change_stream._cursor._max_await_time_ms) |
| 149 | self.watched_collection(write_concern=WriteConcern("majority")).insert_one({}) |
| 150 | _ = change_stream.next() |
| 151 | resume_token = change_stream.resume_token |
| 152 | with self.assertRaises(TypeError): |
| 153 | self.change_stream(pipeline={}) |
| 154 | with self.assertRaises(TypeError): |
| 155 | self.change_stream(full_document={}) |
| 156 | # No Error. |
| 157 | with self.change_stream(resume_after=resume_token): |
| 158 | pass |
| 159 | |
| 160 | @no_type_check |
| 161 | def test_try_next(self): |
nothing calls this directly
no test coverage detected