(self, change_stream)
| 312 | @no_type_check |
| 313 | @client_context.require_sync |
| 314 | def _test_next_blocks(self, change_stream): |
| 315 | inserted_doc = {"_id": ObjectId()} |
| 316 | changes = [] |
| 317 | t = threading.Thread(target=lambda: changes.append(change_stream.next())) |
| 318 | t.start() |
| 319 | # Sleep for a bit to prove that the call to next() blocks. |
| 320 | time.sleep(1) |
| 321 | self.assertTrue(t.is_alive()) |
| 322 | self.assertFalse(changes) |
| 323 | self.watched_collection().insert_one(inserted_doc) |
| 324 | # Join with large timeout to give the server time to return the change, |
| 325 | # in particular for shard clusters. |
| 326 | t.join(30) |
| 327 | self.assertFalse(t.is_alive()) |
| 328 | self.assertEqual(1, len(changes)) |
| 329 | self.assertEqual(changes[0]["operationType"], "insert") |
| 330 | self.assertEqual(changes[0]["fullDocument"], inserted_doc) |
| 331 | |
| 332 | @no_type_check |
| 333 | @client_context.require_sync |
no test coverage detected