Method test_aggregation_cursor

test/test_collection.py:1574–1603

(self)

        self.assertIsInstance(cursor, CommandCursor)

    def test_aggregation_cursor(self):
        db = self.db
        if client_context.has_secondaries:
            # Test that getMore messages are sent to the right server.
            db = self.client.get_database(
                db.name,
                read_preference=ReadPreference.SECONDARY,
                write_concern=WriteConcern(w=self.w),
            )

        for collection_size in (10, 1000):
            db.drop_collection("test")
            db.test.insert_many([{"_id": i} for i in range(collection_size)])
            expected_sum = sum(range(collection_size))
            # Use batchSize to ensure multiple getMore messages
            cursor = db.test.aggregate([{"$project": {"_id": "$_id"}}], batchSize=5)

            self.assertEqual(expected_sum, sum(doc["_id"] for doc in cursor.to_list()))

        # Test that batchSize is handled properly.
        cursor = db.test.aggregate([], batchSize=5)
        self.assertEqual(5, len(cursor._data))
        # Force a getMore
        cursor._data.clear()
        next(cursor)
        # batchSize - 1
        self.assertEqual(4, len(cursor._data))
        # Exhaust the cursor. There shouldn't be any errors.
        for _doc in cursor:
            pass

    def test_aggregation_cursor_alive(self):
        self.db.test.delete_many({})