(self)
| 417 | self._test_ops(client, *ops) |
| 418 | |
| 419 | def test_cursor_clone(self): |
| 420 | coll = self.client.pymongo_test.collection |
| 421 | # Ensure some batches. |
| 422 | coll.insert_many({} for _ in range(10)) |
| 423 | self.addCleanup(coll.drop) |
| 424 | |
| 425 | with self.client.start_session() as s: |
| 426 | cursor = coll.find(session=s) |
| 427 | self.assertIs(cursor.session, s) |
| 428 | clone = cursor.clone() |
| 429 | self.assertIs(clone.session, s) |
| 430 | |
| 431 | # No explicit session. |
| 432 | cursor = coll.find(batch_size=2) |
| 433 | next(cursor) |
| 434 | # Session is "owned" by cursor. |
| 435 | self.assertIsNone(cursor.session) |
| 436 | self.assertIsNotNone(cursor._session) |
| 437 | clone = cursor.clone() |
| 438 | next(clone) |
| 439 | self.assertIsNone(clone.session) |
| 440 | self.assertIsNotNone(clone._session) |
| 441 | self.assertIsNot(cursor._session, clone._session) |
| 442 | cursor.close() |
| 443 | clone.close() |
| 444 | |
| 445 | def test_cursor(self): |
| 446 | listener = self.listener |
nothing calls this directly
no test coverage detected