Method test_clone

test/test_cursor.py:898–997

(self)

    # oplog_reply, and snapshot are all deprecated.
    @ignore_deprecations
    def test_clone(self):
        self.db.test.insert_many([{"x": i} for i in range(1, 4)])

        cursor = self.db.test.find().limit(2)

        count = 0
        for _ in cursor:
            count += 1
        self.assertEqual(2, count)

        count = 0
        for _ in cursor:
            count += 1
        self.assertEqual(0, count)

        cursor = cursor.clone()
        cursor2 = cursor.clone()
        count = 0
        for _ in cursor:
            count += 1
        self.assertEqual(2, count)
        for _ in cursor2:
            count += 1
        self.assertEqual(4, count)

        cursor.rewind()
        count = 0
        for _ in cursor:
            break
        cursor = cursor.clone()
        for _ in cursor:
            count += 1
        self.assertEqual(2, count)

        self.assertNotEqual(cursor, cursor.clone())

        # Just test attributes
        cursor = (
            self.db.test.find(
                {"x": re.compile("^hello.*")},
                projection={"_id": False},
                skip=1,
                no_cursor_timeout=True,
                cursor_type=CursorType.TAILABLE_AWAIT,
                sort=[("x", 1)],
                allow_partial_results=True,
                oplog_replay=True,
                batch_size=123,
                collation={"locale": "en_US"},
                hint=[("_id", 1)],
                max_scan=100,
                max_time_ms=1000,
                return_key=True,
                show_record_id=True,
                snapshot=True,
                allow_disk_use=True,
            )
        ).limit(2)