Crypto Async Tests (#42546)

* wip

* async tests and recordings

* removing unused import

* addressing comments

* updating recordings and making sure inputstream async tests actually use input streams

* updating recordings

Author: ibrandes

sdk/storage/azure-storage-blob-cryptography/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "java",
   "TagPrefix": "java/storage/azure-storage-blob-cryptography",
-  "Tag": "java/storage/azure-storage-blob-cryptography_da06ce0fc7"
+  "Tag": "java/storage/azure-storage-blob-cryptography_28278320dd"
 }

sdk/storage/azure-storage-blob-cryptography/src/test/java/com/azure/storage/blob/specialized/cryptography/BlobCryptographyTestBase.java

@@ -4,20 +4,31 @@
 package com.azure.storage.blob.specialized.cryptography;
 
 import com.azure.core.client.traits.HttpTrait;
+import com.azure.core.credential.TokenCredential;
 import com.azure.core.cryptography.AsyncKeyEncryptionKey;
 import com.azure.core.cryptography.AsyncKeyEncryptionKeyResolver;
 import com.azure.core.http.HttpClient;
 import com.azure.core.http.HttpHeaderName;
 import com.azure.core.http.HttpHeaders;
+import com.azure.core.http.HttpPipeline;
+import com.azure.core.http.HttpPipelineBuilder;
 import com.azure.core.http.HttpResponse;
+import com.azure.core.http.policy.BearerTokenAuthenticationPolicy;
+import com.azure.core.http.policy.ExponentialBackoff;
 import com.azure.core.http.policy.HttpPipelinePolicy;
+import com.azure.core.http.policy.HttpPolicyProviders;
+import com.azure.core.http.policy.RetryPolicy;
+import com.azure.core.http.policy.RetryStrategy;
+import com.azure.core.http.policy.UserAgentPolicy;
 import com.azure.core.http.rest.Response;
 import com.azure.core.test.TestMode;
 import com.azure.core.test.TestProxyTestBase;
 import com.azure.core.test.models.CustomMatcher;
 import com.azure.core.test.models.TestProxySanitizer;
 import com.azure.core.test.models.TestProxySanitizerType;
+import com.azure.core.util.Configuration;
 import com.azure.core.util.CoreUtils;
+import com.azure.security.keyvault.keys.KeyServiceVersion;
 import com.azure.security.keyvault.keys.cryptography.models.KeyWrapAlgorithm;
 import com.azure.storage.blob.BlobAsyncClient;
 import com.azure.storage.blob.BlobClient;
@@ -47,10 +58,13 @@
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.OffsetDateTime;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Collectors;
 
 import static com.azure.core.test.utils.TestUtils.assertArraysEqual;
 import static com.azure.core.test.utils.TestUtils.assertByteBuffersEqual;
@@ -246,10 +260,16 @@ protected String setupBlobMatchCondition(EncryptedBlobClient ebbc, String match)
             : match;
     }
 
-    protected String setupBlobMatchCondition(EncryptedBlobAsyncClient ebbac, String match) {
+    protected Mono<String> setupBlobMatchCondition(EncryptedBlobAsyncClient ebbac, String match) {
         return RECEIVED_ETAG.equals(match)
-            ? ebbac.getPropertiesWithResponse(null).block().getHeaders().getValue(HttpHeaderName.ETAG)
-            : match;
+            ? ebbac.getProperties().map(BlobProperties::getETag)
+            : Mono.justOrEmpty(match).defaultIfEmpty("null");
+    }
+
+    protected static List<String> convertNulls(String... conditions) {
+        return Arrays.stream(conditions)
+            .map(condition -> "null".equals(condition) ? null : condition)
+            .collect(Collectors.toList());
     }
 
     /**
@@ -272,17 +292,21 @@ protected String setupBlobLeaseCondition(BlobClient bc, String leaseID) {
         return RECEIVED_LEASE_ID.equals(leaseID) ? responseLeaseId : leaseID;
     }
 
-    protected String setupBlobLeaseCondition(BlobAsyncClient bac, String leaseID) {
-        String responseLeaseId = null;
+    protected Mono<String> setupBlobLeaseCondition(BlobAsyncClient bac, String leaseID) {
+        Mono<String> responseLeaseId = null;
         if (RECEIVED_LEASE_ID.equals(leaseID) || GARBAGE_LEASE_ID.equals(leaseID)) {
             responseLeaseId = new BlobLeaseClientBuilder()
                 .blobAsyncClient(bac)
                 .buildAsyncClient()
-                .acquireLease(-1)
-                .block();
+                .acquireLease(-1);
         }
 
-        return RECEIVED_LEASE_ID.equals(leaseID) ? responseLeaseId : leaseID;
+        if (responseLeaseId == null) {
+            return Mono.justOrEmpty(leaseID).defaultIfEmpty("null");
+        }
+
+        return responseLeaseId.map(returnedLeaseId -> RECEIVED_LEASE_ID.equals(leaseID)
+            ? returnedLeaseId : (leaseID == null ? "null" : leaseID));
     }
 
     protected static BlobLeaseClient createLeaseClient(BlobClient blobClient) {
@@ -421,4 +445,42 @@ protected <T extends HttpTrait<T>, E extends Enum<E>> T instrument(T builder) {
     protected HttpClient getHttpClient() {
         return StorageCommonTestUtils.getHttpClient(interceptorManager);
     }
+
+    protected HttpPipeline getHttpPipeline(KeyServiceVersion serviceVersion) {
+        Configuration global = Configuration.getGlobalConfiguration().clone();
+        TokenCredential credential;
+
+        credential = StorageCommonTestUtils.getTokenCredential(interceptorManager);
+
+        // Closest to API goes first, closest to wire goes last.
+        final List<HttpPipelinePolicy> policies = new ArrayList<>();
+        policies.add(new UserAgentPolicy("client_name", "client_version", global, serviceVersion));
+        HttpPolicyProviders.addBeforeRetryPolicies(policies);
+        RetryStrategy strategy = new ExponentialBackoff(5, Duration.ofSeconds(2), Duration.ofSeconds(16));
+        policies.add(new RetryPolicy(strategy));
+        policies.add(new BearerTokenAuthenticationPolicy(credential, "https://vault.azure.net/.default"));
+        HttpPolicyProviders.addAfterRetryPolicies(policies);
+
+        if (getTestMode() == TestMode.RECORD) {
+            policies.add(interceptorManager.getRecordPolicy());
+        }
+
+        return new HttpPipelineBuilder()
+            .policies(policies.toArray(new HttpPipelinePolicy[0]))
+            .httpClient(getHttpClient())
+            .build();
+    }
+
+    protected static void compareListToBuffer(List<ByteBuffer> buffers, ByteBuffer result) {
+        result.position(0);
+
+        for (ByteBuffer buffer : buffers) {
+            buffer.position(0);
+            result.limit(result.position() + buffer.remaining());
+            assertByteBuffersEqual(buffer, result);
+            result.position(result.position() + buffer.remaining());
+        }
+
+        assertEquals(0, result.remaining());
+    }
 }

sdk/storage/azure-storage-blob-cryptography/src/test/java/com/azure/storage/blob/specialized/cryptography/EncryptedBlobOutputStreamTests.java

@@ -25,7 +25,6 @@
 public class EncryptedBlobOutputStreamTests extends BlobCryptographyTestBase {
     private static final String KEY_ID = "keyId";
     private EncryptedBlobClient bec; // encrypted client
-    private EncryptedBlobAsyncClient beac; // encrypted async client
     private BlobContainerClient cc;
 
     @Override
@@ -40,11 +39,6 @@ protected void beforeTest() {
 
         String blobName = generateBlobName();
 
-        beac = getEncryptedClientBuilder(fakeKey, null, ENV.getPrimaryAccount().getCredential(),
-            cc.getBlobContainerUrl())
-            .blobName(blobName)
-            .buildEncryptedBlobAsyncClient();
-
         bec = getEncryptedClientBuilder(fakeKey, null, ENV.getPrimaryAccount().getCredential(),
             cc.getBlobContainerUrl())
             .blobName(blobName)
@@ -112,7 +106,7 @@ public void encryptedBlobOutputStreamDefaultNoOverwriteInterrupted() throws IOEx
     @Test
     public void encryptedBlobOutputStreamOverwrite() throws IOException {
         byte[] randomData = getRandomByteArray(10 * Constants.MB);
-        beac.upload(DATA.getDefaultFlux(), null).block();
+        bec.upload(DATA.getDefaultInputStream());
 
         try (OutputStream outputStream = bec.getBlobOutputStream(true)) {
             outputStream.write(randomData);