(self)
| 434 | @no_type_check |
| 435 | @client_context.require_version_min(4, 2, 0) |
| 436 | def test_start_after(self): |
| 437 | resume_token = self.get_resume_token(invalidate=True) |
| 438 | |
| 439 | # resume_after cannot resume after invalidate. |
| 440 | with self.assertRaises(OperationFailure): |
| 441 | self.change_stream(resume_after=resume_token) |
| 442 | |
| 443 | # start_after can resume after invalidate. |
| 444 | with self.change_stream(start_after=resume_token) as change_stream: |
| 445 | self.watched_collection().insert_one({"_id": 2}) |
| 446 | change = change_stream.next() |
| 447 | self.assertEqual(change["operationType"], "insert") |
| 448 | self.assertEqual(change["fullDocument"], {"_id": 2}) |
| 449 | |
| 450 | @no_type_check |
| 451 | @client_context.require_version_min(4, 2, 0) |
nothing calls this directly
no test coverage detected