Method test_upload_batching

test/test_gridfs_bucket.py:328–335

(self)

    @patch("gridfs.synchronous.grid_file._UPLOAD_BUFFER_CHUNKS", 10)
    def test_upload_batching(self):
        with self.fs.open_upload_stream("test_file", chunk_size_bytes=1) as gin:
            gin.write(b"s" * (10 - 1))
            # No chunks were uploaded yet.
            self.assertEqual(0, self.db.fs.chunks.count_documents({"files_id": gin._id}))
            gin.write(b"s")
            # All chunks were uploaded since we hit the _UPLOAD_BUFFER_CHUNKS limit.
            self.assertEqual(10, self.db.fs.chunks.count_documents({"files_id": gin._id}))

    def test_open_upload_stream(self):
        gin = self.fs.open_upload_stream("from_stream")