Added tests

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java

@@ -428,12 +428,12 @@ private Mono<Response<BlockBlobItem>> determineUploadFullOrChunked(final Flux<By
                         return false;
                     }
                 }
-                /*
-                 * Use cutBefore = true as we want to window all data under 4MB into one window.
-                 * Set the prefetch to CHUNKED_UPLOAD_REQUIREMENT in the case that there are numerous tiny buffers,
-                 * windowUntil uses a default limit of 256 and once that is hit it will trigger onComplete which causes
-                 * downstream issues.
-                 */
+            /*
+             * Use cutBefore = true as we want to window all data under 4MB into one window.
+             * Set the prefetch to the maxSingleUploadSize in the case that there are numerous tiny buffers,
+             * windowUntil uses a default limit of 256 and once that is hit it will trigger onComplete which causes
+             * downstream issues.
+             */
             }, true, parallelTransferOptions.getMaxSingleUploadSize())
             .buffer(2)
             .next()

sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/specialized/BlockBlobAPITest.groovy

@@ -809,10 +809,9 @@ class BlockBlobAPITest extends APISpec {
         where:
         dataSize                                       | singleUploadSize | blockSize || expectedBlockCount
         BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES - 1 | null             | null      || 0 // Test that the default for singleUploadSize is the maximum
-        BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES + 1 | null             | null      || Math.ceil((double) BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES + 1 / (double) BlobClient.BLOB_DEFAULT_UPLOAD_BLOCK_SIZE) // "". This also validates the default for blockSize
+        BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES + 1 | null             | null      || Math.ceil(((double) BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES + 1) / (double) BlobClient.BLOB_DEFAULT_UPLOAD_BLOCK_SIZE) // "". This also validates the default for blockSize
         100                                            | 50               | null      || 1 // Test that singleUploadSize is respected
         100                                            | 50               | 20        || 5 // Test that blockSize is respected
-
     }
 
     def "Upload min"() {
@@ -1319,6 +1318,28 @@ class BlockBlobAPITest extends APISpec {
         "foo" | "bar"  | "fizz" | "buzz"
     }
 
+    @Unroll
+    @Requires({ liveMode() })
+    def "Buffered upload options"() {
+        setup:
+        def data = getRandomData(dataSize)
+
+        when:
+        blobAsyncClient.uploadWithResponse(Flux.just(data),
+            new ParallelTransferOptions(blockSize, null, null, singleUploadSize), null, null, null, null).block()
+
+        then:
+        blobAsyncClient.getBlockBlobAsyncClient()
+            .listBlocks(BlockListType.COMMITTED).block().getCommittedBlocks().size() == expectedBlockCount
+
+        where:
+        dataSize                                       | singleUploadSize | blockSize || expectedBlockCount
+        BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES - 1 | null             | null      || 0 // Test that the default for singleUploadSize is the maximum
+        BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES + 1 | null             | null      || Math.ceil(((double) BlockBlobAsyncClient.MAX_UPLOAD_BLOB_BYTES + 1) / (double) BlobClient.BLOB_DEFAULT_UPLOAD_BLOCK_SIZE) // "". This also validates the default for blockSize
+        100                                            | 50               | null      || 1 // Test that singleUploadSize is respected
+        100                                            | 50               | 20        || 5 // Test that blockSize is respected
+    }
+
     // Only run these tests in live mode as they use variables that can't be captured.
     @Unroll
     @Requires({ liveMode() })