Azure Storage Turn on Fixed Tests

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

@@ -39,9 +39,8 @@ public final class DownloadAsyncResponse {
 
 
     // The constructor is package-private because customers should not be creating their own responses.
-    // TODO (unknown): resolve comment vs code mismatch
     DownloadAsyncResponse(ResponseBase<BlobDownloadHeaders, Flux<ByteBuf>> response,
-            HTTPGetterInfo info, Function<HTTPGetterInfo, Mono<DownloadAsyncResponse>> getter) {
+                     HTTPGetterInfo info, Function<HTTPGetterInfo, Mono<DownloadAsyncResponse>> getter) {
         Utility.assertNotNull("getter", getter);
         Utility.assertNotNull("info", info);
         Utility.assertNotNull("info.eTag", info.eTag());
@@ -55,10 +54,8 @@ public final class DownloadAsyncResponse {
      * {@code options.maxRetryRequests > 0}. If retries are enabled, if a connection fails while reading, the stream
      * will make additional requests to reestablish a connection and continue reading.
      *
-     * @param options
-     *         {@link ReliableDownloadOptions}
-     *
-     * @return A {@code Flux} which emits the data as {@code ByteBuffer}s.
+     * @param options {@link ReliableDownloadOptions}
+     * @return A {@link Flux} which emits the data as {@link ByteBuf ByteBufs}
      */
     public Flux<ByteBuf> body(ReliableDownloadOptions options) {
         ReliableDownloadOptions optionsReal = options == null ? new ReliableDownloadOptions() : options;
@@ -86,38 +83,32 @@ possible the method call that returns a Single is what throws (like how our apis
             call time rather than at subscription time.
              */
             try {
-                // Get a new response and try reading from it.
-                return getter.apply(this.info)
-                        .flatMapMany(response ->
-                            /*
-                            Do not compound the number of retries by passing in another set of downloadOptions; just get
-                            the raw body.
-                             */
-                            this.applyReliableDownload(this.rawResponse.value(), retryCount, options));
+                /*Get a new response and try reading from it.
+                Do not compound the number of retries by passing in another set of downloadOptions; just get
+                the raw body.
+                */
+                return getter.apply(this.info).flatMapMany(response -> this.applyReliableDownload(this.rawResponse.value(), retryCount, options));
             } catch (Exception e) {
                 // If the getter fails, return the getter failure to the user.
                 return Flux.error(e);
             }
         }
     }
 
-    private Flux<ByteBuf> applyReliableDownload(Flux<ByteBuf> data,
-            int currentRetryCount, ReliableDownloadOptions options) {
-        return data
-                .doOnNext(buffer -> {
-                    /*
-                    Update how much data we have received in case we need to retry and propagate to the user the data we
-                    have received.
-                     */
-                    this.info.offset(this.info.offset() + buffer.readableBytes()); // was `remaining()` in Rx world
-                    if (this.info.count() != null) {
-                        this.info.count(this.info.count() - buffer.readableBytes()); // was `remaining()` in Rx world
-                    }
-                })
-                .onErrorResume(t2 -> {
-                    // Increment the retry count and try again with the new exception.
-                    return tryContinueFlux(t2, currentRetryCount + 1, options);
-                });
+    private Flux<ByteBuf> applyReliableDownload(Flux<ByteBuf> data, int currentRetryCount, ReliableDownloadOptions options) {
+        return data.doOnNext(buffer -> {
+            /*
+            Update how much data we have received in case we need to retry and propagate to the user the data we
+            have received.
+             */
+            this.info.offset(this.info.offset() + buffer.readableBytes()); // was `remaining()` in Rx world
+            if (this.info.count() != null) {
+                this.info.count(this.info.count() - buffer.readableBytes()); // was `remaining()` in Rx world
+            }
+        }).onErrorResume(t2 -> {
+            // Increment the retry count and try again with the new exception.
+            return tryContinueFlux(t2, currentRetryCount + 1, options);
+        });
     }
 
     /**

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

@@ -16,6 +16,7 @@
 import com.azure.storage.blob.models.ModifiedAccessConditions;
 import com.azure.storage.blob.models.PageBlobAccessConditions;
 import com.azure.storage.blob.models.PageBlobItem;
+import com.azure.storage.blob.models.PageList;
 import com.azure.storage.blob.models.PageRange;
 import com.azure.storage.blob.models.SequenceNumberActionType;
 import com.azure.storage.blob.models.SourceModifiedAccessConditions;
@@ -336,7 +337,7 @@ public Mono<Response<PageBlobItem>> clearPages(PageRange pageRange,
      * @return
      *      A reactive response containing the information of the cleared pages.
      */
-    public Flux<PageRange> getPageRanges(BlobRange blobRange) {
+    public Mono<Response<PageList>> getPageRanges(BlobRange blobRange) {
         return this.getPageRanges(blobRange, null);
     }
 
@@ -352,15 +353,15 @@ public Flux<PageRange> getPageRanges(BlobRange blobRange) {
      * @return
      *      A reactive response emitting all the page ranges.
      */
-    public Flux<PageRange> getPageRanges(BlobRange blobRange, BlobAccessConditions accessConditions) {
+    public Mono<Response<PageList>> getPageRanges(BlobRange blobRange, BlobAccessConditions accessConditions) {
         blobRange = blobRange == null ? new BlobRange(0) : blobRange;
         accessConditions = accessConditions == null ? new BlobAccessConditions() : accessConditions;
 
         return postProcessResponse(this.azureBlobStorage.pageBlobs().getPageRangesWithRestResponseAsync(
             null, null, snapshot, null, null, blobRange.toHeaderValue(),
             null, accessConditions.leaseAccessConditions(), accessConditions.modifiedAccessConditions(),
             Context.NONE))
-            .flatMapMany(response -> Flux.fromIterable(response.value().pageRange()));
+            .map(response -> new SimpleResponse<>(response, response.value()));
     }
 
     /**
@@ -377,7 +378,7 @@ public Flux<PageRange> getPageRanges(BlobRange blobRange, BlobAccessConditions a
      * @return
      *      A reactive response emitting all the different page ranges.
      */
-    public Flux<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSnapshot) {
+    public Mono<Response<PageList>> getPageRangesDiff(BlobRange blobRange, String prevSnapshot) {
         return this.getPageRangesDiff(blobRange, prevSnapshot, null);
     }
 
@@ -397,7 +398,7 @@ public Flux<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSnapsho
      * @return A reactive response emitting all the different page ranges.
      * @throws IllegalArgumentException If {@code prevSnapshot} is {@code null}
      */
-    public Flux<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSnapshot, BlobAccessConditions accessConditions) {
+    public Mono<Response<PageList>> getPageRangesDiff(BlobRange blobRange, String prevSnapshot, BlobAccessConditions accessConditions) {
         blobRange = blobRange == null ? new BlobRange(0) : blobRange;
         accessConditions = accessConditions == null ? new BlobAccessConditions() : accessConditions;
 
@@ -409,7 +410,7 @@ public Flux<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSnapsho
             null, null, snapshot, null, null, prevSnapshot,
             blobRange.toHeaderValue(), null, accessConditions.leaseAccessConditions(),
             accessConditions.modifiedAccessConditions(), Context.NONE))
-            .flatMapMany(response -> Flux.fromIterable(response.value().pageRange()));
+            .map(response -> new SimpleResponse<>(response, response.value()));
     }
 
     /**

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

@@ -12,6 +12,7 @@
 import com.azure.storage.blob.models.ModifiedAccessConditions;
 import com.azure.storage.blob.models.PageBlobAccessConditions;
 import com.azure.storage.blob.models.PageBlobItem;
+import com.azure.storage.blob.models.PageList;
 import com.azure.storage.blob.models.PageRange;
 import com.azure.storage.blob.models.SequenceNumberActionType;
 import com.azure.storage.blob.models.SourceModifiedAccessConditions;
@@ -325,7 +326,7 @@ public Response<PageBlobItem> clearPages(PageRange pageRange,
      * @return
      *      The information of the cleared pages.
      */
-    public Iterable<PageRange> getPageRanges(BlobRange blobRange) {
+    public Response<PageList> getPageRanges(BlobRange blobRange) {
         return this.getPageRanges(blobRange, null, null);
     }
 
@@ -343,10 +344,8 @@ public Iterable<PageRange> getPageRanges(BlobRange blobRange) {
      * @return
      *      All the page ranges.
      */
-    public Iterable<PageRange> getPageRanges(BlobRange blobRange,
-            BlobAccessConditions accessConditions, Duration timeout) {
-        Flux<PageRange> response = pageBlobAsyncClient.getPageRanges(blobRange, accessConditions);
-        return timeout == null ? response.toIterable() : response.timeout(timeout).toIterable();
+    public Response<PageList> getPageRanges(BlobRange blobRange, BlobAccessConditions accessConditions, Duration timeout) {
+        return Utility.blockWithOptionalTimeout(pageBlobAsyncClient.getPageRanges(blobRange, accessConditions), timeout);
     }
 
     /**
@@ -363,7 +362,7 @@ public Iterable<PageRange> getPageRanges(BlobRange blobRange,
      * @return
      *      All the different page ranges.
      */
-    public Iterable<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSnapshot) {
+    public Response<PageList> getPageRangesDiff(BlobRange blobRange, String prevSnapshot) {
         return this.getPageRangesDiff(blobRange, prevSnapshot, null, null);
     }
 
@@ -385,10 +384,8 @@ public Iterable<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSna
      * @return
      *      All the different page ranges.
      */
-    public Iterable<PageRange> getPageRangesDiff(BlobRange blobRange, String prevSnapshot,
-            BlobAccessConditions accessConditions, Duration timeout) {
-        Flux<PageRange> response = pageBlobAsyncClient.getPageRangesDiff(blobRange, prevSnapshot, accessConditions);
-        return timeout == null ? response.toIterable() : response.timeout(timeout).toIterable();
+    public Response<PageList> getPageRangesDiff(BlobRange blobRange, String prevSnapshot, BlobAccessConditions accessConditions, Duration timeout) {
+        return Utility.blockWithOptionalTimeout(pageBlobAsyncClient.getPageRangesDiff(blobRange, prevSnapshot, accessConditions), timeout);
     }
 
     /**

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

@@ -17,6 +17,7 @@ import com.azure.core.util.configuration.ConfigurationManager
 import com.azure.identity.credential.EnvironmentCredential
 import com.azure.storage.blob.models.*
 import com.azure.storage.common.credentials.SharedKeyCredential
+import io.netty.buffer.ByteBuf
 import org.junit.Assume
 import org.spockframework.lang.ISpecificationContext
 import reactor.core.publisher.Flux
@@ -409,7 +410,7 @@ class APISpec extends Specification {
             response.value().contentEncoding() == contentEncoding &&
             response.value().contentLanguage() == contentLanguage &&
             response.value().contentMD5() == contentMD5 &&
-            response.headers().value("Content-Type") == (contentType == null ? "application/octet-stream" : contentType)
+            response.headers().value("Content-Type") == contentType
     }
 
     static Metadata getMetadataFromHeaders(HttpHeaders headers) {
@@ -502,43 +503,51 @@ class APISpec extends Specification {
     to play too nicely with mocked objects and the complex reflection stuff on both ends made it more difficult to work
     with than was worth it. Because this type is just for BlobDownload, we don't need to accept a header type.
      */
-    def getStubResponseForBlobDownload(int code, Flux<ByteBuffer> body, String etag) {
-        return new HttpResponse() {
+    static class MockDownloadHttpResponse extends HttpResponse {
+        private final int statusCode
+        private final HttpHeaders headers
+        private final Flux<ByteBuf> body
+
+        MockDownloadHttpResponse(HttpResponse response, int statusCode, Flux<ByteBuf> body) {
+            this.request(response.request())
+            this.statusCode = statusCode
+            this.headers = response.headers()
+            this.body = body
+        }
 
-            @Override
-            int statusCode() {
-                return code
-            }
+        @Override
+        int statusCode() {
+            return statusCode
+        }
 
-            @Override
-            String headerValue(String s) {
-                return null
-            }
+        @Override
+        String headerValue(String s) {
+            return headers.value(s)
+        }
 
-            @Override
-            HttpHeaders headers() {
-                return new HttpHeaders()
-            }
+        @Override
+        HttpHeaders headers() {
+            return headers
+        }
 
-            @Override
-            Flux<ByteBuffer> body() {
-                return body
-            }
+        @Override
+        Flux<ByteBuf> body() {
+            return body
+        }
 
-            @Override
-            Mono<byte[]> bodyAsByteArray() {
-                return null
-            }
+        @Override
+        Mono<byte[]> bodyAsByteArray() {
+            return Mono.error(new IOException())
+        }
 
-            @Override
-            Mono<String> bodyAsString() {
-                return null
-            }
+        @Override
+        Mono<String> bodyAsString() {
+            return Mono.error(new IOException())
+        }
 
-            @Override
-            Mono<String> bodyAsString(Charset charset) {
-                return null
-            }
+        @Override
+        Mono<String> bodyAsString(Charset charset) {
+            return Mono.error(new IOException())
         }
     }
 

sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/AadLoginTest.java

@@ -3,6 +3,7 @@
 
 package com.azure.storage.blob;
 
+import com.azure.core.util.configuration.ConfigurationManager;
 import com.azure.identity.credential.EnvironmentCredential;
 import com.azure.storage.blob.models.ContainerItem;
 import org.junit.BeforeClass;
@@ -15,8 +16,9 @@ public class AadLoginTest {
 
     @BeforeClass
     public static void setup() {
+        String endpoint = String.format("https://%s.blob.core.windows.net", ConfigurationManager.getConfiguration().get("PRIMARY_STORAGE_ACCOUNT_KEY"));
         storageClient = new BlobServiceClientBuilder()
-            .endpoint("https://" + System.getenv("ACCOUNT_NAME") + ".blob.core.windows.net")
+            .endpoint(endpoint)
             .credential(new EnvironmentCredential())
 //            .httpClient(HttpClient.createDefault().proxy(() -> new ProxyOptions(Type.HTTP, new InetSocketAddress("localhost", 8888))))
             .buildClient();

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

@@ -57,6 +57,9 @@ class AppendBlobAPITest extends APISpec {
         bu.create(headers, null, null, null)
         Response<BlobProperties> response = bu.getProperties()
 
+        // If the value isn't set the service will automatically set it
+        contentType = (contentType == null) ? "application/octet-stream" : contentType
+
         then:
         validateBlobProperties(response, cacheControl, contentDisposition, contentEncoding, contentLanguage, contentMD5, contentType)