Method test_batch_size

test/test_cursor.py:590–645

(self)

            coll.find().max([("j", 3)]).to_list()

    def test_batch_size(self):
        db = self.db
        db.test.drop()
        db.test.insert_many([{"x": x} for x in range(200)])

        with self.assertRaises(TypeError):
            db.test.find().batch_size(None)  # type: ignore[arg-type]
        with self.assertRaises(TypeError):
            db.test.find().batch_size("hello")  # type: ignore[arg-type]
        with self.assertRaises(TypeError):
            db.test.find().batch_size(5.5)  # type: ignore[arg-type]
        with self.assertRaises(ValueError):
            db.test.find().batch_size(-1)
        self.assertTrue((db.test.find()).batch_size(5))
        a = db.test.find()
        for _ in a:
            break
        self.assertRaises(InvalidOperation, a.batch_size, 5)

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

        cursor_count((db.test.find()).batch_size(0), 200)
        cursor_count((db.test.find()).batch_size(1), 200)
        cursor_count((db.test.find()).batch_size(2), 200)
        cursor_count((db.test.find()).batch_size(5), 200)
        cursor_count((db.test.find()).batch_size(100), 200)
        cursor_count((db.test.find()).batch_size(500), 200)

        cursor_count((db.test.find()).batch_size(0).limit(1), 1)
        cursor_count((db.test.find()).batch_size(1).limit(1), 1)
        cursor_count((db.test.find()).batch_size(2).limit(1), 1)
        cursor_count((db.test.find()).batch_size(5).limit(1), 1)
        cursor_count((db.test.find()).batch_size(100).limit(1), 1)
        cursor_count((db.test.find()).batch_size(500).limit(1), 1)

        cursor_count((db.test.find()).batch_size(0).limit(10), 10)
        cursor_count((db.test.find()).batch_size(1).limit(10), 10)
        cursor_count((db.test.find()).batch_size(2).limit(10), 10)
        cursor_count((db.test.find()).batch_size(5).limit(10), 10)
        cursor_count((db.test.find()).batch_size(100).limit(10), 10)
        cursor_count((db.test.find()).batch_size(500).limit(10), 10)

        cur = db.test.find().batch_size(1)
        next(cur)
        # find command batchSize should be 1
        self.assertEqual(0, len(cur._data))
        next(cur)
        self.assertEqual(0, len(cur._data))
        next(cur)
        self.assertEqual(0, len(cur._data))
        next(cur)
        self.assertEqual(0, len(cur._data))

    def test_limit_and_batch_size(self):