Repository: googleapis/java-storage
Pull request #1820 (2022-12-20 20:47:13)

docs: document differing behavior of {get,list}{,default}Acl between HTTP and gRPC
Update ITAccessTest to run UBLA and IAM centric tests for gRPC. Split iam v1 and iam v3 tests out from ITAccessTest to ITBucketIamPolicyTest making them hermetic.



Patch:
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 386a348e..21e88426 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -17,7 +17,6 @@
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
     <pubsub-proto.version>1.104.2</pubsub-proto.version>
-    <kms.version>0.101.0</kms.version>
     <junit-platform.version>5.9.1</junit-platform.version>
   </properties>
   <dependencies>
@@ -143,10 +142,6 @@
       <artifactId>grpc-netty-shaded</artifactId>
       <scope>runtime</scope>
     </dependency>
-    <dependency>
-      <groupId>io.grpc</groupId>
-      <artifactId>grpc-auth</artifactId>
-    </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-stub</artifactId>
@@ -170,21 +165,17 @@
       <artifactId>grpc-xds</artifactId>
       <scope>runtime</scope>
     </dependency>
-    <dependency>
-      <groupId>com.google.api.grpc</groupId>
-      <artifactId>grpc-google-cloud-kms-v1</artifactId>
-      <version>${kms.version}</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>${kms.version}</version>
+      <version>0.101.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>com.google.api.grpc</groupId>
-      <artifactId>grpc-google-iam-v1</artifactId>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-kms</artifactId>
+      <version>2.10.0</version>
+      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.api.grpc</groupId>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
index c6f5e4da..3a29f16d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
@@ -33,6 +33,7 @@ import com.google.cloud.storage.BlobInfo.CustomerEncryption;
 import com.google.cloud.storage.BucketInfo.CustomPlacementConfig;
 import com.google.cloud.storage.BucketInfo.LifecycleRule;
 import com.google.cloud.storage.BucketInfo.LifecycleRule.AbortIncompleteMPUAction;
+import com.google.cloud.storage.BucketInfo.Logging;
 import com.google.cloud.storage.BucketInfo.PublicAccessPrevention;
 import com.google.cloud.storage.Conversions.Codec;
 import com.google.cloud.storage.HmacKey.HmacKeyState;
@@ -261,7 +262,10 @@ final class GrpcConversions {
       to.setCors(toImmutableListOf(corsCodec::decode).apply(corsList));
     }
     if (from.hasLogging()) {
-      to.setLogging(loggingCodec.decode(from.getLogging()));
+      Bucket.Logging logging = from.getLogging();
+      if (!logging.getLogBucket().isEmpty() || !logging.getLogObjectPrefix().isEmpty()) {
+        to.setLogging(loggingCodec.decode(logging));
+      }
     }
     if (from.hasOwner()) {
       to.setOwner(entityCodec.decode(from.getOwner().getEntity()));
@@ -357,7 +361,16 @@ final class GrpcConversions {
       }
       to.setLifecycle(lifecycleBuilder.build());
     }
-    ifNonNull(from.getLogging(), loggingCodec::encode, to::setLogging);
+
+    Logging logging = from.getLogging();
+    if (logging != null) {
+      // an empty bucket name is invalid, don't even attempt to encode if neither name or prefix
+      // are both empty
+      if ((logging.getLogBucket() != null && !logging.getLogBucket().isEmpty())
+          || (logging.getLogObjectPrefix() != null && !logging.getLogObjectPrefix().isEmpty())) {
+        to.setLogging(loggingCodec.encode(logging));
+      }
+    }
     ifNonNull(from.getCors(), toImmutableListOf(corsCodec::encode), to::addAllCors);
     ifNonNull(
         from.getOwner(),
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index 10f60230..97389f92 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -92,6 +92,7 @@ import com.google.storage.v2.GetServiceAccountRequest;
 import com.google.storage.v2.ListBucketsRequest;
 import com.google.storage.v2.ListHmacKeysRequest;
 import com.google.storage.v2.ListObjectsRequest;
+import com.google.storage.v2.ListObjectsResponse;
 import com.google.storage.v2.LockBucketRetentionPolicyRequest;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ObjectAccessControl;
@@ -101,12 +102,6 @@ import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.RewriteObjectRequest;
 import com.google.storage.v2.RewriteResponse;
 import com.google.storage.v2.StorageClient;
-import com.google.storage.v2.StorageClient.ListBucketsPage;
-import com.google.storage.v2.StorageClient.ListBucketsPagedResponse;
-import com.google.storage.v2.StorageClient.ListHmacKeysPage;
-import com.google.storage.v2.StorageClient.ListHmacKeysPagedResponse;
-import com.google.storage.v2.StorageClient.ListObjectsPage;
-import com.google.storage.v2.StorageClient.ListObjectsPagedResponse;
 import com.google.storage.v2.UpdateBucketRequest;
 import com.google.storage.v2.UpdateHmacKeyRequest;
 import com.google.storage.v2.UpdateObjectRequest;
@@ -131,6 +126,7 @@ import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
@@ -438,8 +434,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Page<Bucket> list(BucketListOption... options) {
-    UnaryCallable<ListBucketsRequest, ListBucketsPagedResponse> listBucketsCallable =
-        storageClient.listBucketsPagedCallable();
     Opts<BucketListOpt> opts = Opts.unwrap(options);
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
@@ -449,14 +443,17 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .andThen(opts.listBucketsRequest())
             .apply(ListBucketsRequest.newBuilder())
             .build();
-    ListBucketsPagedResponse call = listBucketsCallable.call(request, grpcCallContext);
     try {
-      ListBucketsPage page = call.getPage();
-      return new TransformingPageDecorator<>(
-          page,
-          syntaxDecoders.bucket.andThen(opts.clearBucketFields()),
+      return Retrying.run(
           getOptions(),
-          retryAlgorithmManager.getFor(request));
+          retryAlgorithmManager.getFor(request),
+          () -> storageClient.listBucketsPagedCallable().call(request, grpcCallContext),
+          resp ->
+              new TransformingPageDecorator<>(
+                  resp.getPage(),
+                  syntaxDecoders.bucket.andThen(opts.clearBucketFields()),
+                  getOptions(),
+                  retryAlgorithmManager.getFor(request)));
     } catch (Exception e) {
       throw StorageException.coalesce(e);
     }
@@ -464,8 +461,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Page<Blob> list(String bucket, BlobListOption... options) {
-    UnaryCallable<ListObjectsRequest, ListObjectsPagedResponse> listObjectsCallable =
-        storageClient.listObjectsPagedCallable();
     Opts<ObjectListOpt> opts = Opts.unwrap(options);
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
@@ -473,13 +468,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         ListObjectsRequest.newBuilder().setParent(bucketNameCodec.encode(bucket));
     ListObjectsRequest req = opts.listObjectsRequest().apply(builder).build();
     try {
-      ListObjectsPagedResponse call = listObjectsCallable.call(req, grpcCallContext);
-      ListObjectsPage page = call.getPage();
-      return new TransformingPageDecorator<>(
-          page,
-          syntaxDecoders.blob.andThen(opts.clearBlobFields()),
+      return Retrying.run(
           getOptions(),
-          retryAlgorithmManager.getFor(req));
+          retryAlgorithmManager.getFor(req),
+          () -> storageClient.listObjectsCallable().call(req, grpcCallContext),
+          resp -> new ListObjectsWithSyntheticDirectoriesPage(grpcCallContext, req, resp));
     } catch (Exception e) {
       throw StorageException.coalesce(e);
     }
@@ -1242,8 +1235,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options) {
-    UnaryCallable<ListHmacKeysRequest, ListHmacKeysPagedResponse> listHmacKeysCallable =
-        storageClient.listHmacKeysPagedCallable();
     Opts<HmacKeyListOpt> opts = Opts.unwrap(options);
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
@@ -1255,10 +1246,16 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(ListHmacKeysRequest.newBuilder())
             .build();
     try {
-      ListHmacKeysPagedResponse call = listHmacKeysCallable.call(request, grpcCallContext);
-      ListHmacKeysPage page = call.getPage();
-      return new TransformingPageDecorator<>(
-          page, codecs.hmacKeyMetadata(), getOptions(), retryAlgorithmManager.getFor(request));
+      return Retrying.run(
+          getOptions(),
+          retryAlgorithmManager.getFor(request),
+          () -> storageClient.listHmacKeysPagedCallable().call(request, grpcCallContext),
+          resp ->
+              new TransformingPageDecorator<>(
+                  resp.getPage(),
+                  codecs.hmacKeyMetadata(),
+                  getOptions(),
+                  retryAlgorithmManager.getFor(request)));
     } catch (Exception e) {
       throw StorageException.coalesce(e);
     }
@@ -1439,6 +1436,95 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         b -> codecs.bucketInfo().decode(b).asBucket(GrpcStorageImpl.this);
   }
 
+  /**
+   * Today {@link com.google.cloud.storage.spi.v1.HttpStorageRpc#list(String, Map)} creates
+   * synthetic objects to represent {@code prefixes} ("directories") returned as part of a list
+   * objects response. Specifically, a StorageObject with an `isDirectory` attribute added.
+   *
+   * <p>This approach is not sound, and presents an otherwise ephemeral piece of metadata as an
+   * actual piece of data. (A {@code prefix} is not actually an object, and therefor can't be
+   * queried for other object metadata.)
+   *
+   * <p>In an effort to preserve compatibility with the current public API, this class attempts to
+   * encapsulate the process of producing these Synthetic Directory Objects and lifting them into
+   * the Page.
+   *
+   * <p>This behavior should NOT be carried forward to any possible new API for the storage client.
+   */
+  private final class ListObjectsWithSyntheticDirectoriesPage implements Page<Blob> {
+
+    private final GrpcCallContext ctx;
+    private final ListObjectsRequest req;
+    private final ListObjectsResponse resp;
+
+    private ListObjectsWithSyntheticDirectoriesPage(
+        GrpcCallContext ctx, ListObjectsRequest req, ListObjectsResponse resp) {
+      this.ctx = ctx;
+      this.req = req;
+      this.resp = resp;
+    }
+
+    @Override
+    public boolean hasNextPage() {
+      return !resp.getNextPageToken().isEmpty();
+    }
+
+    @Override
+    public String getNextPageToken() {
+      return resp.getNextPageToken();
+    }
+
+    @Override
+    public Page<Blob> getNextPage() {
+      ListObjectsRequest nextPageReq =
+          req.toBuilder().setPageToken(resp.getNextPageToken()).build();
+      try {
+        ListObjectsResponse nextPageResp =
+            Retrying.run(
+                GrpcStorageImpl.this.getOptions(),
+                retryAlgorithmManager.getFor(nextPageReq),
+                () -> storageClient.listObjectsCallable().call(nextPageReq, ctx),
+                Decoder.identity());
+        return new ListObjectsWithSyntheticDirectoriesPage(ctx, nextPageReq, nextPageResp);
+      } catch (Exception e) {
+        throw StorageException.coalesce(e);
+      }
+    }
+
+    @Override
+    public Iterable<Blob> iterateAll() {
+      // drop to our interface type to help type inference below with the stream.
+      Page<Blob> curr = this;
+      Predicate<Page<Blob>> exhausted = p -> p != null && p.hasNextPage();
+      // Create a stream which will attempt to call getNextPage repeatedly until we meet our
+      // condition of exhaustion. By doing this we are able to rely on the retry logic in
+      // getNextPage
+      return () ->
+          streamIterate(curr, exhausted, Page::getNextPage)
+              .filter(Objects::nonNull)
+              .flatMap(p -> StreamSupport.stream(p.getValues().spliterator(), false))
+              .iterator();
+    }
+
+    @Override
+    public Iterable<Blob> getValues() {
+      return () -> {
+        String bucketName = bucketNameCodec.decode(req.getParent());
+        return Streams.concat(
+                resp.getObjectsList().stream().map(syntaxDecoders.blob::decode),
+                resp.getPrefixesList().stream()
+                    .map(
+                        prefix ->
+                            BlobInfo.newBuilder(bucketName, prefix)
+                                .setSize(0L)
+                                .setIsDirectory(true)
+                                .build())
+                    .map(info -> info.asBlob(GrpcStorageImpl.this)))
+            .iterator();
+      };
+    }
+  }
+
   static final class TransformingPageDecorator<
           RequestT,
           ResponseT,
@@ -1510,50 +1596,50 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
               .map(translator::decode)
               .iterator();
     }
+  }
 
-    private static <T> Stream<T> streamIterate(
-        T seed, Predicate<? super T> shouldComputeNext, UnaryOperator<T> computeNext) {
-      requireNonNull(seed, "seed must be non null");
-      requireNonNull(shouldComputeNext, "shouldComputeNext must be non null");
-      requireNonNull(computeNext, "computeNext must be non null");
-      Spliterator<T> spliterator =
-          new AbstractSpliterator<T>(Long.MAX_VALUE, 0) {
-            T prev;
-            boolean started = false;
-            boolean done = false;
-
-            @Override
-            public boolean tryAdvance(Consumer<? super T> action) {
-              // if we haven't started, emit our seed and return
-              if (!started) {
-                started = true;
-                action.accept(seed);
-                prev = seed;
+  private static <T> Stream<T> streamIterate(
+      T seed, Predicate<? super T> shouldComputeNext, UnaryOperator<T> computeNext) {
+    requireNonNull(seed, "seed must be non null");
+    requireNonNull(shouldComputeNext, "shouldComputeNext must be non null");
+    requireNonNull(computeNext, "computeNext must be non null");
+    Spliterator<T> spliterator =
+        new AbstractSpliterator<T>(Long.MAX_VALUE, 0) {
+          T prev;
+          boolean started = false;
+          boolean done = false;
+
+          @Override
+          public boolean tryAdvance(Consumer<? super T> action) {
+            // if we haven't started, emit our seed and return
+            if (!started) {
+              started = true;
+              action.accept(seed);
+              prev = seed;
+              return true;
+            }
+            // if we've previously finished quickly return
+            if (done) {
+              return false;
+            }
+            // test whether we should try and compute the next value
+            if (shouldComputeNext.test(prev)) {
+              // compute the next value and figure out if we can use it
+              T next = computeNext.apply(prev);
+              if (next != null) {
+                action.accept(next);
+                prev = next;
                 return true;
               }
-              // if we've previously finished quickly return
-              if (done) {
-                return false;
-              }
-              // test whether we should try and compute the next value
-              if (shouldComputeNext.test(prev)) {
-                // compute the next value and figure out if we can use it
-                T next = computeNext.apply(prev);
-                if (next != null) {
-                  action.accept(next);
-                  prev = next;
-                  return true;
-                }
-              }
-
-              // fallthrough, if we haven't taken an action by now consider the stream done and
-              // return
-              done = true;
-              return false;
             }
-          };
-      return StreamSupport.stream(spliterator, false);
-    }
+
+            // fallthrough, if we haven't taken an action by now consider the stream done and
+            // return
+            done = true;
+            return false;
+          }
+        };
+    return StreamSupport.stream(spliterator, false);
   }
 
   static <T> T throwHttpJsonOnly(String methodName) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 5b3c06da..bfc2cdb0 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -3363,6 +3363,16 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * Acl acl = storage.getAcl(bucketName, new User(userEmail), userProjectOption);
    * }</pre>
    *
+   * <h4>Behavioral Differences between HTTP and gRPC</h4>
+   *
+   * <ol>
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
+   *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
+   *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
+   *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
+   *       gRPC, an empty list will be returned.
+   * </ol>
+   *
    * @param bucket name of the bucket where the getAcl operation takes place
    * @param entity ACL entity to fetch
    * @param options extra parameters to apply to this operation
@@ -3496,6 +3506,16 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * }
    * }</pre>
    *
+   * <h4>Behavioral Differences between HTTP and gRPC</h4>
+   *
+   * <ol>
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
+   *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
+   *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
+   *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
+   *       gRPC, an empty list will be returned.
+   * </ol>
+   *
    * @param bucket the name of the bucket to list ACLs for
    * @param options any number of BucketSourceOptions to apply to this operation
    * @throws StorageException upon failure
@@ -3521,6 +3541,16 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * Acl acl = storage.getDefaultAcl(bucketName, User.ofAllAuthenticatedUsers());
    * }</pre>
    *
+   * <h4>Behavioral Differences between HTTP and gRPC</h4>
+   *
+   * <ol>
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
+   *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
+   *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
+   *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
+   *       gRPC, an empty list will be returned.
+   * </ol>
+   *
    * @throws StorageException upon failure
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
@@ -3604,6 +3634,16 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * }
    * }</pre>
    *
+   * <h4>Behavioral Differences between HTTP and gRPC</h4>
+   *
+   * <ol>
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
+   *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
+   *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
+   *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
+   *       gRPC, an empty list will be returned.
+   * </ol>
+   *
    * @throws StorageException upon failure
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
diff --git a/pom.xml b/pom.xml
index 8fd9c3a7..a26499fe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -92,7 +92,7 @@
       <dependency>
         <groupId>org.mockito</groupId>
         <artifactId>mockito-core</artifactId>
-        <version>4.10.0</version>
+        <version>4.11.0</version>
         <scope>test</scope>
         <exclusions>
           <exclusion>
diff --git a/renovate.json b/renovate.json
index 8201f595..b5e21946 100644
--- a/renovate.json
+++ b/renovate.json
@@ -72,6 +72,7 @@
     {
       "groupName": "cross product test dependencies",
       "packagePatterns": [
+        "^com.google.cloud:google-cloud-kms",
         "^com.google.cloud:google-cloud-pubsub",
         "^com.google.api.grpc:grpc-google-cloud-kms-v1",
         "^com.google.api.grpc:proto-google-cloud-kms-v1",


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java
index 99b29818..a46aad53 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage.it;
 
+import static com.google.cloud.storage.TestUtils.assertAll;
 import static com.google.cloud.storage.TestUtils.retry429s;
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -26,8 +27,6 @@ import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import com.google.cloud.Condition;
-import com.google.cloud.Identity;
 import com.google.cloud.Policy;
 import com.google.cloud.storage.Acl;
 import com.google.cloud.storage.Acl.Entity;
@@ -46,30 +45,19 @@ import com.google.cloud.storage.Storage.BucketField;
 import com.google.cloud.storage.Storage.BucketGetOption;
 import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
-import com.google.cloud.storage.StorageRoles;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
 import com.google.cloud.storage.it.runner.annotations.Backend;
-import com.google.cloud.storage.it.runner.annotations.BucketFixture;
-import com.google.cloud.storage.it.runner.annotations.BucketType;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.function.Predicate;
-import java.util.stream.Collector;
 import java.util.stream.Collectors;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -84,13 +72,9 @@ public class ITAccessTest {
 
   @Inject public Storage storage;
 
-  @Inject
-  @BucketFixture(BucketType.DEFAULT)
-  public BucketInfo bucket;
+  @Inject public Transport transport;
 
-  @Inject
-  @BucketFixture(BucketType.REQUESTER_PAYS)
-  public BucketInfo requesterPaysBucket;
+  @Inject public BucketInfo bucket;
 
   @Inject public Generator generator;
 
@@ -310,367 +294,53 @@ public class ITAccessTest {
     }
   }
 
+  /** Validate legacy deprecated field is redirected correctly */
   @Test
-  @Ignore("Make hermetic, previously dependant on external transitive state")
-  public void testBucketPolicyV1RequesterPays() {
-    String projectId = storage.getOptions().getProjectId();
-
-    Storage.BucketSourceOption[] bucketOptions =
-        new Storage.BucketSourceOption[] {Storage.BucketSourceOption.userProject(projectId)};
-    Identity projectOwner = Identity.projectOwner(projectId);
-    Identity projectEditor = Identity.projectEditor(projectId);
-    Identity projectViewer = Identity.projectViewer(projectId);
-    Map<com.google.cloud.Role, Set<Identity>> bindingsWithoutPublicRead =
-        ImmutableMap.of(
-            StorageRoles.legacyBucketOwner(),
-            ImmutableSet.of(projectOwner, projectEditor),
-            StorageRoles.legacyBucketReader(),
-            ImmutableSet.of(projectViewer));
-    Map<com.google.cloud.Role, Set<Identity>> bindingsWithPublicRead =
-        ImmutableMap.of(
-            StorageRoles.legacyBucketOwner(),
-            ImmutableSet.of(projectOwner, projectEditor),
-            StorageRoles.legacyBucketReader(),
-            ImmutableSet.of(projectViewer),
-            StorageRoles.legacyObjectReader(),
-            ImmutableSet.of(Identity.allUsers()));
-
-    // Validate getting policy.
-    Policy currentPolicy = storage.getIamPolicy(requesterPaysBucket.getName(), bucketOptions);
-    assertEquals(bindingsWithoutPublicRead, currentPolicy.getBindings());
-
-    // Validate updating policy.
-    Policy updatedPolicy =
-        storage.setIamPolicy(
-            requesterPaysBucket.getName(),
-            currentPolicy
-                .toBuilder()
-                .addIdentity(StorageRoles.legacyObjectReader(), Identity.allUsers())
-                .build(),
-            bucketOptions);
-    assertEquals(bindingsWithPublicRead, updatedPolicy.getBindings());
-    Policy revertedPolicy =
-        storage.setIamPolicy(
-            requesterPaysBucket.getName(),
-            updatedPolicy
-                .toBuilder()
-                .removeIdentity(StorageRoles.legacyObjectReader(), Identity.allUsers())
-                .build(),
-            bucketOptions);
-    assertEquals(bindingsWithoutPublicRead, revertedPolicy.getBindings());
-
-    // Validate testing permissions.
-    List<Boolean> expectedPermissions = ImmutableList.of(true, true);
-    assertEquals(
-        expectedPermissions,
-        storage.testIamPermissions(
-            requesterPaysBucket.getName(),
-            ImmutableList.of("storage.buckets.getIamPolicy", "storage.buckets.setIamPolicy"),
-            bucketOptions));
+  @SuppressWarnings("deprecation")
+  public void testBucketWithBucketPolicyOnlyEnabled() throws Exception {
+    doTestUniformBucketLevelAccessAclImpact(
+        BucketInfo.IamConfiguration.newBuilder().setIsBucketPolicyOnlyEnabled(true).build());
   }
 
   @Test
-  @Ignore("Make hermetic, previously dependant on external transitive state")
-  public void testBucketPolicyV1() {
-    String projectId = storage.getOptions().getProjectId();
-
-    Storage.BucketSourceOption[] bucketOptions = new Storage.BucketSourceOption[] {};
-    Identity projectOwner = Identity.projectOwner(projectId);
-    Identity projectEditor = Identity.projectEditor(projectId);
-    Identity projectViewer = Identity.projectViewer(projectId);
-    Map<com.google.cloud.Role, Set<Identity>> bindingsWithoutPublicRead =
-        ImmutableMap.of(
-            StorageRoles.legacyBucketOwner(),
-            ImmutableSet.of(projectOwner, projectEditor),
-            StorageRoles.legacyBucketReader(),
-            ImmutableSet.of(projectViewer));
-    Map<com.google.cloud.Role, Set<Identity>> bindingsWithPublicRead =
-        ImmutableMap.of(
-            StorageRoles.legacyBucketOwner(),
-            ImmutableSet.of(projectOwner, projectEditor),
-            StorageRoles.legacyBucketReader(),
-            ImmutableSet.of(projectViewer),
-            StorageRoles.legacyObjectReader(),
-            ImmutableSet.of(Identity.allUsers()));
-
-    // Validate getting policy.
-    Policy currentPolicy = storage.getIamPolicy(bucket.getName(), bucketOptions);
-    assertEquals(bindingsWithoutPublicRead, currentPolicy.getBindings());
-
-    // Validate updating policy.
-    Policy updatedPolicy =
-        storage.setIamPolicy(
-            bucket.getName(),
-            currentPolicy
-                .toBuilder()
-                .addIdentity(StorageRoles.legacyObjectReader(), Identity.allUsers())
-                .build(),
-            bucketOptions);
-    assertEquals(bindingsWithPublicRead, updatedPolicy.getBindings());
-    Policy revertedPolicy =
-        storage.setIamPolicy(
-            bucket.getName(),
-            updatedPolicy
-                .toBuilder()
-                .removeIdentity(StorageRoles.legacyObjectReader(), Identity.allUsers())
-                .build(),
-            bucketOptions);
-    assertEquals(bindingsWithoutPublicRead, revertedPolicy.getBindings());
-
-    // Validate testing permissions.
-    List<Boolean> expectedPermissions = ImmutableList.of(true, true);
-    assertEquals(
-        expectedPermissions,
-        storage.testIamPermissions(
-            bucket.getName(),
-            ImmutableList.of("storage.buckets.getIamPolicy", "storage.buckets.setIamPolicy"),
-            bucketOptions));
+  public void testBucketWithUniformBucketLevelAccessEnabled() throws Exception {
+    doTestUniformBucketLevelAccessAclImpact(
+        BucketInfo.IamConfiguration.newBuilder()
+            .setIsUniformBucketLevelAccessEnabled(true)
+            .build());
   }
 
-  @Test
-  @Ignore("Make hermetic, previously dependant on external transitive state")
-  public void testBucketPolicyV3() throws Exception {
-    String projectId = storage.getOptions().getProjectId();
-    BucketInfo bucketInfo =
-        BucketInfo.newBuilder(generator.randomBucketName())
-            .setIamConfiguration(
-                BucketInfo.IamConfiguration.newBuilder()
-                    .setIsUniformBucketLevelAccessEnabled(true)
-                    .build())
-            .build();
-    Storage.BucketSourceOption[] bucketOptions =
-        new Storage.BucketSourceOption[] {Storage.BucketSourceOption.requestedPolicyVersion(3)};
-    Identity projectOwner = Identity.projectOwner(projectId);
-    Identity projectEditor = Identity.projectEditor(projectId);
-    Identity projectViewer = Identity.projectViewer(projectId);
-    List<com.google.cloud.Binding> bindingsWithoutPublicRead =
-        ImmutableList.of(
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyBucketOwner().toString())
-                .setMembers(ImmutableList.of(projectEditor.strValue(), projectOwner.strValue()))
-                .build(),
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyBucketReader().toString())
-                .setMembers(ImmutableList.of(projectViewer.strValue()))
-                .build());
-    List<com.google.cloud.Binding> bindingsWithPublicRead =
-        ImmutableList.of(
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyBucketReader().toString())
-                .setMembers(ImmutableList.of(projectViewer.strValue()))
-                .build(),
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyBucketOwner().toString())
-                .setMembers(ImmutableList.of(projectEditor.strValue(), projectOwner.strValue()))
-                .build(),
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyObjectReader().toString())
-                .setMembers(ImmutableList.of("allUsers"))
-                .build());
-
-    List<com.google.cloud.Binding> bindingsWithConditionalPolicy =
-        ImmutableList.of(
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyBucketReader().toString())
-                .setMembers(ImmutableList.of(projectViewer.strValue()))
-                .build(),
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyBucketOwner().toString())
-                .setMembers(ImmutableList.of(projectEditor.strValue(), projectOwner.strValue()))
-                .build(),
-            com.google.cloud.Binding.newBuilder()
-                .setRole(StorageRoles.legacyObjectReader().toString())
-                .setMembers(
-                    ImmutableList.of(
-                        "serviceAccount:storage-python@spec-test-ruby-samples.iam.gserviceaccount.com"))
-                .setCondition(
-                    Condition.newBuilder()
-                        .setTitle("Title")
-                        .setDescription("Description")
-                        .setExpression(
-                            "resource.name.startsWith(\"projects/_/buckets/bucket-name/objects/prefix-a-\")")
-                        .build())
-                .build());
-
+  private void doTestUniformBucketLevelAccessAclImpact(IamConfiguration iamConfiguration)
+      throws Exception {
+    String bucketName = generator.randomBucketName();
     try (TemporaryBucket tempB =
-        TemporaryBucket.newBuilder().setBucketInfo(bucketInfo).setStorage(storage).build()) {
+        TemporaryBucket.newBuilder()
+            .setBucketInfo(
+                Bucket.newBuilder(bucketName).setIamConfiguration(iamConfiguration).build())
+            .setStorage(storage)
+            .build()) {
       BucketInfo bucket = tempB.getBucket();
 
-      // Validate getting policy.
-      Policy currentPolicy = storage.getIamPolicy(bucket.getName(), bucketOptions);
-      Collector<CharSequence, ?, String> joining = Collectors.joining(",\n\t", "[\n\t", "\n]");
-      String s = currentPolicy.getBindingsList().stream().map(Object::toString).collect(joining);
-      String ss = bindingsWithoutPublicRead.stream().map(Object::toString).collect(joining);
-      assertThat(s).isEqualTo(ss);
-      // assertEquals(bindingsWithoutPublicRead, currentPolicy.getBindingsList());
-
-      // Validate updating policy.
-      List<com.google.cloud.Binding> currentBindings =
-          new ArrayList(currentPolicy.getBindingsList());
-      currentBindings.add(
-          com.google.cloud.Binding.newBuilder()
-              .setRole(StorageRoles.legacyObjectReader().getValue())
-              .addMembers(Identity.allUsers().strValue())
-              .build());
-      Policy updatedPolicy =
-          storage.setIamPolicy(
-              bucket.getName(),
-              currentPolicy.toBuilder().setBindings(currentBindings).build(),
-              bucketOptions);
-      assertTrue(
-          bindingsWithPublicRead.size() == updatedPolicy.getBindingsList().size()
-              && bindingsWithPublicRead.containsAll(updatedPolicy.getBindingsList()));
-
-      // Remove a member
-      List<com.google.cloud.Binding> updatedBindings =
-          new ArrayList(updatedPolicy.getBindingsList());
-      for (int i = 0; i < updatedBindings.size(); i++) {
-        com.google.cloud.Binding binding = updatedBindings.get(i);
-        if (binding.getRole().equals(StorageRoles.legacyObjectReader().toString())) {
-          List<String> members = new ArrayList(binding.getMembers());
-          members.remove(Identity.allUsers().strValue());
-          updatedBindings.set(i, binding.toBuilder().setMembers(members).build());
-          break;
-        }
+      assertTrue(bucket.getIamConfiguration().isUniformBucketLevelAccessEnabled());
+      assertNotNull(
+          bucket.getIamConfiguration().getUniformBucketLevelAccessLockedTimeOffsetDateTime());
+
+      if (transport == Transport.HTTP) {
+        StorageException listAclsError =
+            assertThrows(StorageException.class, () -> storage.listAcls(bucketName));
+        assertAll(
+            () -> assertThat(listAclsError.getCode()).isEqualTo(400),
+            () -> assertThat(listAclsError.getReason()).isEqualTo("invalid"));
+
+        StorageException listDefaultAclsError =
+            assertThrows(StorageException.class, () -> storage.listDefaultAcls(bucketName));
+        assertAll(
+            () -> assertThat(listDefaultAclsError.getCode()).isEqualTo(400),
+            () -> assertThat(listDefaultAclsError.getReason()).isEqualTo("invalid"));
+      } else if (transport == Transport.GRPC) {
+        assertThat(storage.listAcls(bucketName)).isEmpty();
+        assertThat(storage.listDefaultAcls(bucketName)).isEmpty();
       }
-
-      Policy revertedPolicy =
-          storage.setIamPolicy(
-              bucket.getName(),
-              updatedPolicy.toBuilder().setBindings(updatedBindings).build(),
-              bucketOptions);
-
-      assertEquals(bindingsWithoutPublicRead, revertedPolicy.getBindingsList());
-      assertTrue(
-          bindingsWithoutPublicRead.size() == revertedPolicy.getBindingsList().size()
-              && bindingsWithoutPublicRead.containsAll(revertedPolicy.getBindingsList()));
-
-      // Add Conditional Policy
-      List<com.google.cloud.Binding> conditionalBindings =
-          new ArrayList(revertedPolicy.getBindingsList());
-      conditionalBindings.add(
-          com.google.cloud.Binding.newBuilder()
-              .setRole(StorageRoles.legacyObjectReader().toString())
-              .addMembers(
-                  "serviceAccount:storage-python@spec-test-ruby-samples.iam.gserviceaccount.com")
-              .setCondition(
-                  Condition.newBuilder()
-                      .setTitle("Title")
-                      .setDescription("Description")
-                      .setExpression(
-                          "resource.name.startsWith(\"projects/_/buckets/bucket-name/objects/prefix-a-\")")
-                      .build())
-              .build());
-      Policy conditionalPolicy =
-          storage.setIamPolicy(
-              bucket.getName(),
-              revertedPolicy.toBuilder().setBindings(conditionalBindings).setVersion(3).build(),
-              bucketOptions);
-      assertTrue(
-          bindingsWithConditionalPolicy.size() == conditionalPolicy.getBindingsList().size()
-              && bindingsWithConditionalPolicy.containsAll(conditionalPolicy.getBindingsList()));
-
-      // Remove Conditional Policy
-      conditionalPolicy =
-          storage.setIamPolicy(
-              bucket.getName(),
-              conditionalPolicy.toBuilder().setBindings(updatedBindings).setVersion(3).build(),
-              bucketOptions);
-
-      // Validate testing permissions.
-      List<Boolean> expectedPermissions = ImmutableList.of(true, true);
-      assertEquals(
-          expectedPermissions,
-          storage.testIamPermissions(
-              bucket.getName(),
-              ImmutableList.of("storage.buckets.getIamPolicy", "storage.buckets.setIamPolicy"),
-              bucketOptions));
-    }
-  }
-
-  @Test
-  @SuppressWarnings({"unchecked", "deprecation"})
-  @CrossRun.Ignore(transports = Transport.GRPC)
-  public void testBucketWithBucketPolicyOnlyEnabled() throws Exception {
-    // TODO: break this test up into each of the respective scenarios
-    //   1. Create bucket with BucketPolicyOnly enabled
-    //   2. Get bucket with BucketPolicyOnly enabled
-    //   3. Expect failure when attempting to list ACLs for BucketPolicyOnly bucket
-    //   4. Expect failure when attempting to list default ACLs for BucketPolicyOnly bucket
-
-    // TODO: temp bucket
-    String randBucketName = generator.randomBucketName();
-    try {
-      storage.create(
-          Bucket.newBuilder(randBucketName)
-              .setIamConfiguration(
-                  BucketInfo.IamConfiguration.newBuilder()
-                      .setIsBucketPolicyOnlyEnabled(true)
-                      .build())
-              .build());
-
-      Bucket remoteBucket =
-          storage.get(randBucketName, Storage.BucketGetOption.fields(BucketField.IAMCONFIGURATION));
-
-      assertTrue(remoteBucket.getIamConfiguration().isBucketPolicyOnlyEnabled());
-      assertNotNull(remoteBucket.getIamConfiguration().getBucketPolicyOnlyLockedTime());
-
-      try {
-        remoteBucket.listAcls();
-        fail("StorageException was expected.");
-      } catch (StorageException e) {
-        // Expected: Listing legacy ACLs should fail on a BPO enabled bucket
-      }
-      try {
-        remoteBucket.listDefaultAcls();
-        fail("StorageException was expected");
-      } catch (StorageException e) {
-        // Expected: Listing legacy ACLs should fail on a BPO enabled bucket
-      }
-    } finally {
-      BucketCleaner.doCleanup(randBucketName, storage);
-    }
-  }
-
-  @Test
-  @CrossRun.Ignore(transports = Transport.GRPC)
-  public void testBucketWithUniformBucketLevelAccessEnabled() throws Exception {
-    // TODO: break this test up into each of the respective scenarios
-    //   1. Create bucket with UniformBucketLevelAccess enabled
-    //   2. Get bucket with UniformBucketLevelAccess enabled
-    //   3. Expect failure when attempting to list ACLs for UniformBucketLevelAccess bucket
-    //   4. Expect failure when attempting to list default ACLs for UniformBucketLevelAccess bucket
-
-    // TODO: temp bucket
-    String randBucketName = generator.randomBucketName();
-    try {
-      storage.create(
-          Bucket.newBuilder(randBucketName)
-              .setIamConfiguration(
-                  BucketInfo.IamConfiguration.newBuilder()
-                      .setIsUniformBucketLevelAccessEnabled(true)
-                      .build())
-              .build());
-
-      Bucket remoteBucket =
-          storage.get(randBucketName, Storage.BucketGetOption.fields(BucketField.IAMCONFIGURATION));
-
-      assertTrue(remoteBucket.getIamConfiguration().isUniformBucketLevelAccessEnabled());
-      assertNotNull(remoteBucket.getIamConfiguration().getUniformBucketLevelAccessLockedTime());
-      try {
-        remoteBucket.listAcls();
-        fail("StorageException was expected.");
-      } catch (StorageException e) {
-        // Expected: Listing legacy ACLs should fail on a BPO enabled bucket
-      }
-      try {
-        remoteBucket.listDefaultAcls();
-        fail("StorageException was expected");
-      } catch (StorageException e) {
-        // Expected: Listing legacy ACLs should fail on a BPO enabled bucket
-      }
-    } finally {
-      BucketCleaner.doCleanup(randBucketName, storage);
     }
   }
 
@@ -910,35 +580,6 @@ public class ITAccessTest {
     }
   }
 
-  @Test
-  public void testListBucketRequesterPaysFails() throws InterruptedException {
-    String projectId = storage.getOptions().getProjectId();
-    Iterator<Bucket> bucketIterator =
-        storage
-            .list(
-                Storage.BucketListOption.prefix(bucket.getName()),
-                Storage.BucketListOption.fields(),
-                Storage.BucketListOption.userProject(projectId))
-            .iterateAll()
-            .iterator();
-    while (!bucketIterator.hasNext()) {
-      Thread.sleep(500);
-      bucketIterator =
-          storage
-              .list(
-                  Storage.BucketListOption.prefix(bucket.getName()),
-                  Storage.BucketListOption.fields())
-              .iterateAll()
-              .iterator();
-    }
-    while (bucketIterator.hasNext()) {
-      Bucket remoteBucket = bucketIterator.next();
-      assertTrue(remoteBucket.getName().startsWith(bucket.getName()));
-      assertNull(remoteBucket.getCreateTime());
-      assertNull(remoteBucket.getSelfLink());
-    }
-  }
-
   @Test
   public void testRetentionPolicyNoLock() throws Exception {
     String bucketName = generator.randomBucketName();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
index ae37b3c9..a4657db7 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
@@ -40,7 +40,9 @@ import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
 import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.CrossRun.Exclude;
 import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.it.runner.registry.ObjectsFixture;
 import com.google.cloud.storage.it.runner.registry.ObjectsFixture.ObjectAndContent;
 import com.google.common.io.BaseEncoding;
@@ -53,7 +55,6 @@ import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.channels.WritableByteChannel;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.nio.file.StandardOpenOption;
@@ -64,7 +65,6 @@ import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -80,8 +80,6 @@ public final class ITBlobReadChannelTest {
       BaseEncoding.base64()
           .decode("H4sIAAAAAAAAAPNIzcnJV3DPz0/PSVVwzskvTVEILskvSkxPVQQA/LySchsAAAA=");
 
-  @Rule public final TestName testName = new TestName();
-
   @Rule public final DataGeneration dataGeneration = new DataGeneration(new Random(872364872));
 
   @Rule public final TemporaryFolder tmp = new TemporaryFolder();
@@ -89,6 +87,7 @@ public final class ITBlobReadChannelTest {
   @Inject public Storage storage;
   @Inject public BucketInfo bucket;
   @Inject public ObjectsFixture objectsFixture;
+  @Inject public Generator generator;
 
   @Test
   public void testLimit_smallerThanOneChunk() throws IOException {
@@ -138,7 +137,7 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void testLimit_downloadToFile() throws IOException {
-    String blobName = String.format("%s/src", testName.getMethodName());
+    String blobName = String.format("%s/src", generator.randomObjectName());
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     ByteBuffer content = dataGeneration.randByteBuffer(108);
     try (WriteChannel writer = storage.writer(BlobInfo.newBuilder(blobId).build())) {
@@ -172,12 +171,12 @@ public final class ITBlobReadChannelTest {
   @Test
   public void
       testReadChannel_preconditionFailureResultsInIOException_metagenerationMatch_specified() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
     Blob remoteBlob = storage.create(blob);
     assertNotNull(remoteBlob);
     try (ReadChannel reader =
-        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.metagenerationMatch(-1L))) {
+        storage.reader(blob.getBlobId(), BlobSourceOption.metagenerationMatch(-1L))) {
       reader.read(ByteBuffer.allocate(42));
       fail("IOException was expected");
     } catch (IOException ex) {
@@ -187,12 +186,12 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void testReadChannel_preconditionFailureResultsInIOException_generationMatch_specified() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
     Blob remoteBlob = storage.create(blob);
     assertNotNull(remoteBlob);
     try (ReadChannel reader =
-        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.generationMatch(-1L))) {
+        storage.reader(blob.getBlobId(), BlobSourceOption.generationMatch(-1L))) {
       reader.read(ByteBuffer.allocate(42));
       fail("IOException was expected");
     } catch (IOException ex) {
@@ -202,13 +201,13 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void testReadChannel_preconditionFailureResultsInIOException_generationMatch_extractor() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
     Blob remoteBlob = storage.create(blob);
     assertNotNull(remoteBlob);
     BlobId blobIdWrongGeneration = BlobId.of(bucket.getName(), blobName, -1L);
     try (ReadChannel reader =
-        storage.reader(blobIdWrongGeneration, Storage.BlobSourceOption.generationMatch())) {
+        storage.reader(blobIdWrongGeneration, BlobSourceOption.generationMatch())) {
       reader.read(ByteBuffer.allocate(42));
       fail("IOException was expected");
     } catch (IOException ex) {
@@ -218,7 +217,7 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void ensureReaderReturnsCompressedBytesByDefault() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobInfo blobInfo =
         BlobInfo.newBuilder(bucket, blobName)
             .setContentType("text/plain")
@@ -244,7 +243,7 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void ensureReaderCanAutoDecompressWhenReturnRawInputStream_false() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobInfo blobInfo =
         BlobInfo.newBuilder(bucket, blobName)
             .setContentType("text/plain")
@@ -265,7 +264,7 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void returnRawInputStream_true() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobInfo blobInfo =
         BlobInfo.newBuilder(bucket, blobName)
             .setContentType("text/plain")
@@ -293,7 +292,7 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void channelIsConsideredOpenUponConstruction() {
     ReadChannel reader = storage.reader(objectsFixture.getInfo1().getBlobId());
     assertThat(reader.isOpen()).isTrue();
@@ -302,9 +301,9 @@ public final class ITBlobReadChannelTest {
 
   @Test
   public void optionsWork() {
-    byte[] bytes1 = "A".getBytes(StandardCharsets.UTF_8);
+    byte[] bytes1 = "A".getBytes(UTF_8);
 
-    BlobInfo info = BlobInfo.newBuilder(bucket, testName.getMethodName()).build();
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
     Blob gen1 = storage.create(info, bytes1, BlobTargetOption.doesNotExist());
 
     // attempt to read generation=1 && ifGenerationNotMatch=1
@@ -321,31 +320,31 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void captureAndRestore_position_Limit() throws IOException {
     captureAndRestoreTest(26, 51);
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void captureAndRestore_position_noLimit() throws IOException {
     captureAndRestoreTest(26, null);
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void captureAndRestore_noPosition_limit() throws IOException {
     captureAndRestoreTest(null, 51);
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void captureAndRestore_noPosition_noLimit() throws IOException {
     captureAndRestoreTest(null, null);
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void seekAfterReadWorks() throws IOException {
     ObjectAndContent obj512KiB = objectsFixture.getObj512KiB();
     BlobInfo gen1 = obj512KiB.getInfo();
@@ -375,7 +374,7 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void limitAfterReadWorks() throws IOException {
     ObjectAndContent obj512KiB = objectsFixture.getObj512KiB();
     BlobInfo gen1 = obj512KiB.getInfo();
@@ -405,12 +404,12 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void readingLastByteReturnsOneByte_seekOnly() throws IOException {
     int length = 10;
     byte[] bytes = DataGenerator.base64Characters().genBytes(length);
 
-    BlobInfo info1 = BlobInfo.newBuilder(bucket, testName.getMethodName()).build();
+    BlobInfo info1 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
     Blob gen1 = storage.create(info1, bytes, BlobTargetOption.doesNotExist());
 
     byte[] expected1 = Arrays.copyOfRange(bytes, 9, 10);
@@ -427,12 +426,12 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void readingLastByteReturnsOneByte_seekAndLimit() throws IOException {
     int length = 10;
     byte[] bytes = DataGenerator.base64Characters().genBytes(length);
 
-    BlobInfo info1 = BlobInfo.newBuilder(bucket, testName.getMethodName()).build();
+    BlobInfo info1 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
     Blob gen1 = storage.create(info1, bytes, BlobTargetOption.doesNotExist());
 
     byte[] expected1 = Arrays.copyOfRange(bytes, 9, 10);
@@ -455,12 +454,12 @@ public final class ITBlobReadChannelTest {
    * <p>This is behavior is a bug, and should be fixed at the next major version
    */
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void responseWith416ReturnsZeroAndLeavesTheChannelOpen() throws IOException {
     int length = 10;
     byte[] bytes = DataGenerator.base64Characters().genBytes(length);
 
-    BlobInfo info1 = BlobInfo.newBuilder(bucket, testName.getMethodName()).build();
+    BlobInfo info1 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
     Blob gen1 = storage.create(info1, bytes, BlobTargetOption.doesNotExist());
 
     try (ReadChannel reader = storage.reader(gen1.getBlobId())) {
@@ -520,7 +519,7 @@ public final class ITBlobReadChannelTest {
 
   private void doLimitTest(int srcContentSize, int rangeBegin, int rangeEnd, int chunkSize)
       throws IOException {
-    String blobName = String.format("%s/src", testName.getMethodName());
+    String blobName = String.format("%s/src", generator.randomObjectName());
     BlobInfo src = BlobInfo.newBuilder(bucket, blobName).build();
     ByteBuffer content = dataGeneration.randByteBuffer(srcContentSize);
     ByteBuffer dup = content.duplicate();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java
index d3094358..e5c2cae7 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java
@@ -36,6 +36,7 @@ import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.annotations.SingleBackend;
 import com.google.cloud.storage.it.runner.annotations.StorageFixture;
+import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.it.runner.registry.TestBench;
 import com.google.cloud.storage.it.runner.registry.TestBench.RetryTestResource;
 import com.google.common.collect.ImmutableList;
@@ -44,9 +45,7 @@ import com.google.gson.JsonArray;
 import com.google.gson.JsonObject;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -55,8 +54,6 @@ public final class ITBlobReadChannelV2RetryTest {
 
   private static final int _512KiB = 512 * 1024;
 
-  @Rule public TestName testName = new TestName();
-
   @Inject public TestBench testBench;
 
   @Inject
@@ -64,6 +61,7 @@ public final class ITBlobReadChannelV2RetryTest {
   public Storage storage;
 
   @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
 
   @Test
   public void generationIsLockedForRetries() throws Exception {
@@ -71,7 +69,7 @@ public final class ITBlobReadChannelV2RetryTest {
     StorageOptions baseOptions = storage.getOptions();
     byte[] bytes = DataGenerator.base64Characters().genBytes(_512KiB);
 
-    BlobId id = BlobId.of(bucket.getName(), testName.getMethodName());
+    BlobId id = BlobId.of(bucket.getName(), generator.randomObjectName());
     Blob gen1 =
         storage.create(BlobInfo.newBuilder(id).build(), bytes, BlobTargetOption.doesNotExist());
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadMaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadMaskTest.java
index 6a42c3ee..96b478d3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadMaskTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadMaskTest.java
@@ -114,7 +114,7 @@ public final class ITBlobReadMaskTest {
     public ImmutableList<?> parameters() {
       ImmutableList<Args<BlobField, BlobInfo>> args =
           ImmutableList.of(
-              new Args<>(BlobField.ACL, LazyAssertion.skip("Waiting for b/217922398 to be fixed")),
+              new Args<>(BlobField.ACL, LazyAssertion.equal()),
               new Args<>(BlobField.BUCKET, LazyAssertion.equal()),
               new Args<>(
                   BlobField.CACHE_CONTROL,
@@ -155,7 +155,7 @@ public final class ITBlobReadMaskTest {
               new Args<>(BlobField.CRC32C, LazyAssertion.equal()),
               new Args<>(BlobField.CUSTOMER_ENCRYPTION, LazyAssertion.equal()),
               new Args<>(BlobField.CUSTOM_TIME, LazyAssertion.equal()),
-              new Args<>(BlobField.ETAG, LazyAssertion.skip("Waiting for b/217922398 to be fixed")),
+              new Args<>(BlobField.ETAG, LazyAssertion.equal()),
               new Args<>(
                   BlobField.EVENT_BASED_HOLD,
                   LazyAssertion.apiaryNullGrpcDefault(false, BlobInfo::getEventBasedHold)),
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java
index 14e42a62..0de35093 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java
@@ -45,6 +45,7 @@ import com.google.cloud.storage.it.runner.StorageITRunner;
 import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.it.runner.registry.TestBench;
 import com.google.cloud.storage.it.runner.registry.TestBench.RetryTestResource;
 import com.google.cloud.storage.spi.StorageRpcFactory;
@@ -62,7 +63,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
 import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 import org.threeten.bp.Clock;
 import org.threeten.bp.Instant;
@@ -73,6 +73,7 @@ import org.threeten.bp.format.DateTimeFormatter;
 @RunWith(StorageITRunner.class)
 @SingleBackend(Backend.TEST_BENCH)
 public final class ITBlobWriteChannelTest {
+
   private static final Logger LOGGER = Logger.getLogger(ITBlobWriteChannelTest.class.getName());
   private static final String NOW_STRING;
   private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
@@ -86,7 +87,7 @@ public final class ITBlobWriteChannelTest {
 
   @Inject public TestBench testBench;
 
-  @Rule public final TestName testName = new TestName();
+  @Inject public Generator generator;
 
   @Rule public final DataGeneration dataGeneration = new DataGeneration(new Random(1234567890));
 
@@ -146,7 +147,7 @@ public final class ITBlobWriteChannelTest {
         DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.from(ZoneOffset.UTC));
     String nowString = formatter.format(now);
     BucketInfo bucketInfo = BucketInfo.of(dataGeneration.getBucketName());
-    String blobPath = String.format("%s/%s/blob", testName.getMethodName(), nowString);
+    String blobPath = String.format("%s/%s/blob", generator.randomObjectName(), nowString);
     BlobId blobId = BlobId.of(dataGeneration.getBucketName(), blobPath);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
     storage.create(bucketInfo);
@@ -161,7 +162,7 @@ public final class ITBlobWriteChannelTest {
   }
 
   private void doJsonUnexpectedEOFTest(int contentSize, int cappedByteCount) throws IOException {
-    String blobPath = String.format("%s/%s/blob", testName.getMethodName(), NOW_STRING);
+    String blobPath = String.format("%s/%s/blob", generator.randomObjectName(), NOW_STRING);
 
     BucketInfo bucketInfo = BucketInfo.of(dataGeneration.getBucketName());
     BlobInfo blobInfoGen0 = BlobInfo.newBuilder(bucketInfo, blobPath, 0L).build();
@@ -260,7 +261,7 @@ public final class ITBlobWriteChannelTest {
         DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.from(ZoneOffset.UTC));
     String nowString = formatter.format(now);
     BucketInfo bucketInfo = BucketInfo.of(dataGeneration.getBucketName());
-    String blobPath = String.format("%s/%s/blob", testName.getMethodName(), nowString);
+    String blobPath = String.format("%s/%s/blob", generator.randomObjectName(), nowString);
     BlobId blobId = BlobId.of(dataGeneration.getBucketName(), blobPath);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketIamPolicyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketIamPolicyTest.java
new file mode 100644
index 00000000..ded7e96b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketIamPolicyTest.java
@@ -0,0 +1,225 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.Binding;
+import com.google.cloud.Condition;
+import com.google.cloud.Identity;
+import com.google.cloud.Policy;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BucketSourceOption;
+import com.google.cloud.storage.StorageRoles;
+import com.google.cloud.storage.TestUtils;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.annotations.ParallelFriendly;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.util.List;
+import java.util.stream.Collector;
+import java.util.stream.Collectors;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    transports = {Transport.HTTP, Transport.GRPC},
+    backends = {Backend.PROD})
+@ParallelFriendly
+public final class ITBucketIamPolicyTest {
+  @Inject public Storage storage;
+
+  @Inject public BucketInfo bucketInfo;
+
+  @Inject public Generator generator;
+
+  private Identity projectOwner;
+  private Identity projectEditor;
+  private Identity projectViewer;
+
+  @Before
+  public void setUp() throws Exception {
+    String projectId = storage.getOptions().getProjectId();
+    projectOwner = Identity.projectOwner(projectId);
+    projectEditor = Identity.projectEditor(projectId);
+    projectViewer = Identity.projectViewer(projectId);
+  }
+
+  /**
+   * In order to define an IAM Condition, policy version 3 and Uniform Bucket Level Access must both
+   * be used.
+   *
+   * <p>Define a policy with a condition and verify it can be read back and decoded equivalently.
+   */
+  @Test
+  public void iamPolicyWithCondition() throws Exception {
+    BucketSourceOption opt = BucketSourceOption.requestedPolicyVersion(3);
+    Policy policy =
+        Policy.newBuilder()
+            .setVersion(3)
+            .setBindings(
+                ImmutableList.of(
+                    Binding.newBuilder()
+                        .setRole(StorageRoles.legacyBucketReader().toString())
+                        .setMembers(ImmutableList.of(projectViewer.strValue()))
+                        .build(),
+                    Binding.newBuilder()
+                        .setRole(StorageRoles.legacyBucketOwner().toString())
+                        .setMembers(
+                            ImmutableList.of(projectEditor.strValue(), projectOwner.strValue()))
+                        .build(),
+                    Binding.newBuilder()
+                        .setRole(StorageRoles.legacyObjectReader().toString())
+                        .setMembers(
+                            ImmutableList.of(
+                                "serviceAccount:storage-python@spec-test-ruby-samples.iam.gserviceaccount.com"))
+                        .setCondition(
+                            Condition.newBuilder()
+                                .setTitle("Title")
+                                .setDescription("Description")
+                                .setExpression(
+                                    "resource.name.startsWith(\"projects/_/buckets/bucket-name/objects/prefix-a-\")")
+                                .build())
+                        .build()))
+            .build();
+
+    try (TemporaryBucket tempB =
+        TemporaryBucket.newBuilder()
+            .setBucketInfo(
+                // create a bucket with UBLA set to true
+                BucketInfo.newBuilder(generator.randomBucketName())
+                    .setIamConfiguration(
+                        BucketInfo.IamConfiguration.newBuilder()
+                            .setIsUniformBucketLevelAccessEnabled(true)
+                            .build())
+                    .build())
+            .setStorage(storage)
+            .build()) {
+      BucketInfo bucket = tempB.getBucket();
+      String bucketName = bucket.getName();
+
+      // Set the policy on the bucket
+      Policy setResult =
+          storage.setIamPolicy(
+              bucketName,
+              policy,
+              BucketSourceOption.metagenerationMatch(bucket.getMetageneration()),
+              opt);
+      assertPolicyEqual(policy, setResult);
+
+      Policy actual = storage.getIamPolicy(bucketName, opt);
+      assertPolicyEqual(policy, actual);
+    }
+  }
+
+  @Test
+  public void iamPolicyWithoutCondition() throws Exception {
+    BucketSourceOption opt = BucketSourceOption.requestedPolicyVersion(1);
+    Policy policy =
+        Policy.newBuilder()
+            .setVersion(1)
+            .setBindings(
+                ImmutableMap.of(
+                    StorageRoles.legacyBucketOwner(),
+                    ImmutableSet.of(projectOwner, projectEditor),
+                    StorageRoles.legacyBucketReader(),
+                    ImmutableSet.of(projectViewer)))
+            .build();
+
+    try (TemporaryBucket tempB =
+        TemporaryBucket.newBuilder()
+            .setBucketInfo(
+                // create a bucket without UBLA
+                BucketInfo.newBuilder(generator.randomBucketName())
+                    .setIamConfiguration(
+                        BucketInfo.IamConfiguration.newBuilder()
+                            .setIsUniformBucketLevelAccessEnabled(false)
+                            .build())
+                    .build())
+            .setStorage(storage)
+            .build()) {
+      BucketInfo bucket = tempB.getBucket();
+      String bucketName = bucket.getName();
+
+      // Set the policy on the bucket
+      Policy setResult =
+          storage.setIamPolicy(
+              bucketName,
+              policy,
+              BucketSourceOption.metagenerationMatch(bucket.getMetageneration()),
+              opt);
+      assertPolicyEqual(policy, setResult);
+
+      Policy actual = storage.getIamPolicy(bucketName, opt);
+      assertPolicyEqual(policy, actual);
+    }
+  }
+
+  @Test
+  public void testIamPermissions() {
+    List<Boolean> expectedResult = ImmutableList.of(true, true);
+    ImmutableList<String> permissions =
+        ImmutableList.of("storage.buckets.getIamPolicy", "storage.buckets.setIamPolicy");
+    List<Boolean> actual = storage.testIamPermissions(bucketInfo.getName(), permissions);
+    assertThat(actual).isEqualTo(expectedResult);
+  }
+
+  private static void assertPolicyEqual(Policy expected, Policy actual) throws Exception {
+    TestUtils.assertAll(
+        () -> assertThat(actual.getVersion()).isEqualTo(expected.getVersion()),
+        () -> assertBindingsEqual(expected.getBindingsList(), actual.getBindingsList()));
+  }
+
+  private static void assertBindingsEqual(List<Binding> expected, List<Binding> actual) {
+
+    // pre-stringify the value to be compared to make it easier to diff if there is a failure
+    // ordering is not necessarily maintained across RPCs, after stringification sort before
+    // comparison
+    String e = stringifyBindings(expected);
+    String a = stringifyBindings(actual);
+
+    assertThat(a).isEqualTo(e);
+  }
+
+  private static String stringifyBindings(List<Binding> bindings) {
+    Collector<CharSequence, ?, String> joining = Collectors.joining(",\n\t", "[\n\t", "\n]");
+    // ordering is not necessarily maintained across RPCs
+    // Sort any lists before stringification
+    return bindings.stream()
+        .map(
+            b -> {
+              Binding.Builder builder = b.toBuilder();
+              builder.setRole(b.getRole());
+              builder.setCondition(b.getCondition());
+              builder.setMembers(
+                  b.getMembers().stream().sorted().collect(ImmutableList.toImmutableList()));
+              return builder.build();
+            })
+        .map(Object::toString)
+        .sorted()
+        .collect(joining);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java
index f29ff4e0..6e7eeadf 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java
@@ -33,9 +33,7 @@ import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -44,10 +42,9 @@ import org.junit.runner.RunWith;
     backends = {Backend.PROD})
 public final class ITBucketLifecycleRulesTest {
 
-  @Rule public final TestName testName = new TestName();
+  @Inject public Generator generator;
 
   @Inject public Storage storage;
-  @Inject public Generator generator;
 
   @Test
   public void deleteRule_addingALabelToABucketWithASingleDeleteRuleOnlyModifiesTheLabels()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketReadMaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketReadMaskTest.java
index d3fa6d76..e0dfdc81 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketReadMaskTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketReadMaskTest.java
@@ -97,8 +97,7 @@ public final class ITBucketReadMaskTest {
     public ImmutableList<?> parameters() {
       ImmutableList<Args<BucketField, BucketInfo>> args =
           ImmutableList.of(
-              new Args<>(
-                  BucketField.ACL, LazyAssertion.skip("Waiting for b/217922398 to be fixed")),
+              new Args<>(BucketField.ACL, LazyAssertion.equal()),
               new Args<>(BucketField.AUTOCLASS, LazyAssertion.equal()),
               new Args<>(BucketField.BILLING, LazyAssertion.equal()),
               new Args<>(BucketField.CORS, LazyAssertion.equal()),
@@ -116,8 +115,7 @@ public final class ITBucketReadMaskTest {
                     assertThat(grpcT.getDefaultAcl()).isNull(); // workaround for b/261771961
                   }),
               new Args<>(BucketField.ENCRYPTION, LazyAssertion.equal()),
-              new Args<>(
-                  BucketField.ETAG, LazyAssertion.skip("Waiting for b/217922398 to be fixed")),
+              new Args<>(BucketField.ETAG, LazyAssertion.equal()),
               new Args<>(BucketField.IAMCONFIGURATION, LazyAssertion.equal()),
               new Args<>(BucketField.ID, LazyAssertion.equal()),
               new Args<>(BucketField.LABELS, LazyAssertion.equal()),
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketTest.java
index 2b75de1b..7f451d9c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketTest.java
@@ -107,8 +107,6 @@ public class ITBucketTest {
   }
 
   @Test
-  // FieldMask not supported in GRPC get currently.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testGetBucketSelectedFields() {
     Bucket remoteBucket =
         storage.get(bucket.getName(), Storage.BucketGetOption.fields(BucketField.ID));
@@ -118,19 +116,14 @@ public class ITBucketTest {
   }
 
   @Test
-  // FieldMask not supported in GRPC currently.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testGetBucketAllSelectedFields() {
     Bucket remoteBucket =
         storage.get(bucket.getName(), Storage.BucketGetOption.fields(BucketField.values()));
     assertEquals(bucket.getName(), remoteBucket.getName());
     assertNotNull(remoteBucket.getCreateTime());
-    assertNotNull(remoteBucket.getSelfLink());
   }
 
   @Test
-  // Cannot turn on for GRPC until b/246634709 is resolved, verified locally.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testBucketLocationType() throws Exception {
     String bucketName = generator.randomBucketName();
     BucketInfo bucketInfo = BucketInfo.newBuilder(bucketName).setLocation("us").build();
@@ -143,8 +136,6 @@ public class ITBucketTest {
   }
 
   @Test
-  // Cannot turn on for GRPC until creation bug b/246634709 is resolved, verified locally.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testBucketCustomPlacmentConfigDualRegion() throws Exception {
     String bucketName = generator.randomBucketName();
     List<String> locations = new ArrayList<>();
@@ -167,8 +158,6 @@ public class ITBucketTest {
   }
 
   @Test
-  // Cannot turn on until GRPC Update logic bug is fixed b/247133805
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testBucketLogging() throws Exception {
     String logsBucketName = generator.randomBucketName();
     String loggingBucketName = generator.randomBucketName();
@@ -210,8 +199,6 @@ public class ITBucketTest {
   }
 
   @Test
-  // GRPC Update logic bug b/247133805
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testRemoveBucketCORS() {
     String bucketName = generator.randomBucketName();
     List<Cors.Origin> origins = ImmutableList.of(Cors.Origin.of("http://cloud.google.com"));
@@ -260,8 +247,6 @@ public class ITBucketTest {
   }
 
   @Test
-  // Cannot turn on for GRPC until b/246634709 is resolved, verified locally.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testRpoConfig() {
     String rpoBucket = generator.randomBucketName();
     try {
@@ -383,7 +368,6 @@ public class ITBucketTest {
   }
 
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testEnableDisableBucketDefaultEventBasedHold() {
     String bucketName = generator.randomBucketName();
     Bucket remoteBucket =
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
index e702f3ff..79c6cf35 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
@@ -52,9 +52,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.stream.IntStream;
 import java.util.stream.StreamSupport;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -69,8 +67,6 @@ public final class ITGrpcTest {
 
   @Inject public Generator generator;
 
-  @Rule public final TestName testName = new TestName();
-
   @Test
   public void testCreateBucket() {
     String bucketName = generator.randomBucketName();
@@ -81,7 +77,7 @@ public final class ITGrpcTest {
   @Test
   public void listBlobs() {
     byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
-    String prefix = testName.getMethodName();
+    String prefix = generator.randomObjectName();
     List<Blob> blobs =
         IntStream.rangeClosed(1, 10)
             .mapToObj(i -> String.format("%s/%02d", prefix, i))
@@ -182,7 +178,7 @@ public final class ITGrpcTest {
 
   @Test
   public void objectWrite_storage_create() {
-    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, generator.randomObjectName()).build();
     byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
     Blob blob = storage.create(info, content, BlobTargetOption.doesNotExist());
     byte[] actual = blob.getContent();
@@ -191,7 +187,7 @@ public final class ITGrpcTest {
 
   @Test
   public void objectWrite_storage_create_stream() {
-    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, generator.randomObjectName()).build();
     byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
     Blob blob =
         storage.create(info, new ByteArrayInputStream(content), BlobWriteOption.doesNotExist());
@@ -201,7 +197,7 @@ public final class ITGrpcTest {
 
   @Test
   public void objectWrite_storage_writer() throws IOException {
-    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, generator.randomObjectName()).build();
     byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
     try (WriteChannel c = storage.writer(info, BlobWriteOption.doesNotExist())) {
       c.write(ByteBuffer.wrap(content));
@@ -216,10 +212,12 @@ public final class ITGrpcTest {
 
     byte[] expected = "Hello, World!".getBytes(StandardCharsets.UTF_8);
 
-    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName() + "copy/src").build();
+    BlobInfo info =
+        BlobInfo.newBuilder(bucketInfo, generator.randomObjectName() + "copy/src").build();
     Blob cpySrc = s.create(info, expected, BlobTargetOption.doesNotExist());
 
-    BlobInfo dst = BlobInfo.newBuilder(bucketInfo, testName.getMethodName() + "copy/dst").build();
+    BlobInfo dst =
+        BlobInfo.newBuilder(bucketInfo, generator.randomObjectName() + "copy/dst").build();
 
     CopyRequest copyRequest =
         CopyRequest.newBuilder()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java
index cdbaa2e2..e49acf88 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java
@@ -24,19 +24,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
-import com.google.api.gax.paging.Page;
-import com.google.auth.Credentials;
 import com.google.cloud.WriteChannel;
-import com.google.cloud.kms.v1.CreateCryptoKeyRequest;
-import com.google.cloud.kms.v1.CreateKeyRingRequest;
-import com.google.cloud.kms.v1.CryptoKey;
-import com.google.cloud.kms.v1.CryptoKeyName;
-import com.google.cloud.kms.v1.GetCryptoKeyRequest;
-import com.google.cloud.kms.v1.GetKeyRingRequest;
-import com.google.cloud.kms.v1.KeyManagementServiceGrpc;
-import com.google.cloud.kms.v1.KeyManagementServiceGrpc.KeyManagementServiceBlockingStub;
-import com.google.cloud.kms.v1.KeyRingName;
-import com.google.cloud.kms.v1.LocationName;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
@@ -44,44 +32,23 @@ import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.CopyWriter;
 import com.google.cloud.storage.DataGeneration;
-import com.google.cloud.storage.ServiceAccount;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobField;
-import com.google.cloud.storage.Storage.BucketField;
 import com.google.cloud.storage.StorageException;
-import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
 import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.it.runner.registry.KmsFixture;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Iterators;
 import com.google.common.io.BaseEncoding;
-import com.google.iam.v1.Binding;
-import com.google.iam.v1.IAMPolicyGrpc;
-import com.google.iam.v1.SetIamPolicyRequest;
-import io.grpc.ClientInterceptor;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
-import io.grpc.Metadata;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
-import io.grpc.auth.MoreCallCredentials;
-import io.grpc.stub.MetadataUtils;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.security.Key;
-import java.util.Iterator;
 import java.util.Random;
-import java.util.Set;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 import javax.crypto.spec.SecretKeySpec;
-import org.junit.After;
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -95,19 +62,6 @@ public class ITKmsTest {
   private static final long seed = -7071346537822433445L;
   @Rule public final DataGeneration dataGeneration = new DataGeneration(new Random(seed));
 
-  // TODO: either pull the scope of these resources up higher, or make them per-test distinct
-  //   to insulate them from each other
-  private String kmsKeyOneResourcePath;
-  private String kmsKeyTwoResourcePath;
-  private static final String KMS_KEY_RING_NAME = "gcs_test_kms_key_ring";
-  private static final String KMS_KEY_RING_LOCATION = "us";
-  private static final String KMS_KEY_ONE_NAME = "gcs_kms_key_one";
-  private static final String KMS_KEY_TWO_NAME = "gcs_kms_key_two";
-  private Metadata requestParamsHeader = new Metadata();
-  private static Metadata.Key<String> requestParamsKey =
-      Metadata.Key.of("x-goog-request-params", Metadata.ASCII_STRING_MARSHALLER);
-  private ManagedChannel kmsChannel;
-  private static final Logger log = Logger.getLogger(ITKmsTest.class.getName());
   private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
   private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
   private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
@@ -118,179 +72,19 @@ public class ITKmsTest {
   @Inject public Storage storage;
   @Inject public BucketInfo bucket;
   @Inject public Generator generator;
-
-  @Before
-  public void setup() {
-    // Prepare KMS KeyRing for CMEK tests
-    // https://cloud.google.com/storage/docs/encryption/using-customer-managed-keys
-    // We don't care currently if we are using HTTP or gRPC because
-    // these values should be the same.
-    StorageOptions storageOptions = storage.getOptions();
-    String projectId = storageOptions.getProjectId();
-
-    Credentials credentials = storageOptions.getCredentials();
-    // TODO: can we switch all this manual grpc stuff for
-    //   com.google.cloud.kms.v1.KeyManagementServiceClient?
-    kmsChannel = ManagedChannelBuilder.forTarget("cloudkms.googleapis.com:443").build();
-    KeyManagementServiceBlockingStub kmsStub =
-        KeyManagementServiceGrpc.newBlockingStub(kmsChannel)
-            .withCallCredentials(MoreCallCredentials.from(credentials));
-    IAMPolicyGrpc.IAMPolicyBlockingStub iamStub =
-        IAMPolicyGrpc.newBlockingStub(kmsChannel)
-            .withCallCredentials(MoreCallCredentials.from(credentials));
-    ensureKmsKeyRingExistsForTests(kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME);
-    ensureKmsKeyRingIamPermissionsForTests(
-        iamStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME);
-    kmsKeyOneResourcePath =
-        ensureKmsKeyExistsForTests(
-            kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME, KMS_KEY_ONE_NAME);
-    kmsKeyTwoResourcePath =
-        ensureKmsKeyExistsForTests(
-            kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME, KMS_KEY_TWO_NAME);
-  }
-
-  @After
-  public void after() {
-    if (kmsChannel != null) {
-      try {
-        kmsChannel.shutdownNow();
-      } catch (Exception e) {
-        log.log(Level.WARNING, "Error while trying to shutdown kms channel", e);
-      }
-      kmsChannel = null;
-    }
-  }
-
-  private String ensureKmsKeyRingExistsForTests(
-      KeyManagementServiceBlockingStub kmsStub,
-      String projectId,
-      String location,
-      String keyRingName)
-      throws StatusRuntimeException {
-    String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
-    try {
-      // Attempt to Get KeyRing
-      GetKeyRingRequest getKeyRingRequest =
-          GetKeyRingRequest.newBuilder().setName(kmsKeyRingResourcePath).build();
-      requestParamsHeader.put(requestParamsKey, "name=" + kmsKeyRingResourcePath);
-      ClientInterceptor headersInterceptor =
-          MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
-      KeyManagementServiceBlockingStub stubForGetKeyRing =
-          kmsStub.withInterceptors(headersInterceptor);
-      stubForGetKeyRing.getKeyRing(getKeyRingRequest);
-    } catch (StatusRuntimeException ex) {
-      if (ex.getStatus().getCode() == Status.Code.NOT_FOUND) {
-        // Create KmsKeyRing
-        String keyRingParent = LocationName.of(projectId, location).toString();
-        CreateKeyRingRequest createKeyRingRequest =
-            CreateKeyRingRequest.newBuilder()
-                .setParent(keyRingParent)
-                .setKeyRingId(keyRingName)
-                .build();
-        requestParamsHeader.put(requestParamsKey, "parent=" + keyRingParent);
-        ClientInterceptor headersInterceptor =
-            MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
-        KeyManagementServiceBlockingStub stubForCreateKeyRing =
-            kmsStub.withInterceptors(headersInterceptor);
-        stubForCreateKeyRing.createKeyRing(createKeyRingRequest);
-      } else {
-        throw ex;
-      }
-    }
-
-    return kmsKeyRingResourcePath;
-  }
-
-  private void ensureKmsKeyRingIamPermissionsForTests(
-      IAMPolicyGrpc.IAMPolicyBlockingStub iamStub,
-      String projectId,
-      String location,
-      String keyRingName)
-      throws StatusRuntimeException {
-    ServiceAccount serviceAccount = storage.getServiceAccount(projectId);
-    String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
-    Binding binding =
-        Binding.newBuilder()
-            .setRole("roles/cloudkms.cryptoKeyEncrypterDecrypter")
-            .addMembers("serviceAccount:" + serviceAccount.getEmail())
-            .build();
-    com.google.iam.v1.Policy policy =
-        com.google.iam.v1.Policy.newBuilder().addBindings(binding).build();
-    SetIamPolicyRequest setIamPolicyRequest =
-        SetIamPolicyRequest.newBuilder()
-            .setResource(kmsKeyRingResourcePath)
-            .setPolicy(policy)
-            .build();
-    requestParamsHeader.put(requestParamsKey, "parent=" + kmsKeyRingResourcePath);
-    ClientInterceptor headersInterceptor =
-        MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
-    iamStub = iamStub.withInterceptors(headersInterceptor);
-    try {
-      iamStub.setIamPolicy(setIamPolicyRequest);
-    } catch (StatusRuntimeException e) {
-      if (log.isLoggable(Level.WARNING)) {
-        log.log(Level.WARNING, "Unable to set IAM policy: {0}", e.getMessage());
-      }
-    }
-  }
-
-  private String ensureKmsKeyExistsForTests(
-      KeyManagementServiceBlockingStub kmsStub,
-      String projectId,
-      String location,
-      String keyRingName,
-      String keyName)
-      throws StatusRuntimeException {
-    String kmsKeyResourcePath =
-        CryptoKeyName.of(projectId, location, keyRingName, keyName).toString();
-    try {
-      // Attempt to Get CryptoKey
-      requestParamsHeader.put(requestParamsKey, "name=" + kmsKeyResourcePath);
-      GetCryptoKeyRequest getCryptoKeyRequest =
-          GetCryptoKeyRequest.newBuilder().setName(kmsKeyResourcePath).build();
-      ClientInterceptor headersInterceptor =
-          MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
-      KeyManagementServiceGrpc.KeyManagementServiceBlockingStub stubForGetCryptoKey =
-          kmsStub.withInterceptors(headersInterceptor);
-      stubForGetCryptoKey.getCryptoKey(getCryptoKeyRequest);
-    } catch (StatusRuntimeException ex) {
-      if (ex.getStatus().getCode() == Status.Code.NOT_FOUND) {
-        String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
-        CryptoKey cryptoKey =
-            CryptoKey.newBuilder().setPurpose(CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT).build();
-        CreateCryptoKeyRequest createCryptoKeyRequest =
-            CreateCryptoKeyRequest.newBuilder()
-                .setCryptoKeyId(keyName)
-                .setParent(kmsKeyRingResourcePath)
-                .setCryptoKey(cryptoKey)
-                .build();
-
-        requestParamsHeader.put(requestParamsKey, "parent=" + kmsKeyRingResourcePath);
-        ClientInterceptor headersInterceptor =
-            MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
-        KeyManagementServiceGrpc.KeyManagementServiceBlockingStub stubForCreateCryptoKey =
-            kmsStub.withInterceptors(headersInterceptor);
-        stubForCreateCryptoKey.createCryptoKey(createCryptoKeyRequest);
-      } else {
-        throw ex;
-      }
-    }
-    return kmsKeyResourcePath;
-  }
+  @Inject public KmsFixture kms;
 
   @Test
   public void testClearBucketDefaultKmsKeyName() {
     String bucketName = generator.randomBucketName();
-    // TODO: replace with storage
-    // b/246634709
     Bucket remoteBucket =
         storage.create(
             BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
+                .setDefaultKmsKeyName(kms.getKey1().getName())
+                .setLocation(kms.getKeyRingLocation())
                 .build());
     try {
-      assertEquals(kmsKeyOneResourcePath, remoteBucket.getDefaultKmsKeyName());
+      assertEquals(kms.getKey1().getName(), remoteBucket.getDefaultKmsKeyName());
       Bucket updatedBucket = remoteBucket.toBuilder().setDefaultKmsKeyName(null).build().update();
       assertNull(updatedBucket.getDefaultKmsKeyName());
     } finally {
@@ -300,21 +94,18 @@ public class ITKmsTest {
 
   @Test
   public void testUpdateBucketDefaultKmsKeyName() {
-    // TODO: replace with storage
-    // b/246634709
     String bucketName = generator.randomBucketName();
     Bucket remoteBucket =
         storage.create(
             BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
+                .setDefaultKmsKeyName(kms.getKey1().getName())
+                .setLocation(kms.getKeyRingLocation())
                 .build());
-
     try {
-      assertEquals(kmsKeyOneResourcePath, remoteBucket.getDefaultKmsKeyName());
+      assertEquals(kms.getKey1().getName(), remoteBucket.getDefaultKmsKeyName());
       Bucket updatedBucket =
-          remoteBucket.toBuilder().setDefaultKmsKeyName(kmsKeyTwoResourcePath).build().update();
-      assertEquals(kmsKeyTwoResourcePath, updatedBucket.getDefaultKmsKeyName());
+          remoteBucket.toBuilder().setDefaultKmsKeyName(kms.getKey2().getName()).build().update();
+      assertEquals(kms.getKey2().getName(), updatedBucket.getDefaultKmsKeyName());
     } finally {
       BucketCleaner.doCleanup(bucketName, storage);
     }
@@ -327,12 +118,12 @@ public class ITKmsTest {
     BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
     Blob remoteBlob =
         storage.create(
-            blob, BLOB_BYTE_CONTENT, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+            blob, BLOB_BYTE_CONTENT, Storage.BlobTargetOption.kmsKeyName(kms.getKey1().getName()));
     assertNotNull(remoteBlob);
     assertEquals(blob.getBucket(), remoteBlob.getBucket());
     assertEquals(blob.getName(), remoteBlob.getName());
     assertNotNull(remoteBlob.getKmsKeyName());
-    assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    assertTrue(remoteBlob.getKmsKeyName().startsWith(kms.getKey1().getName()));
     byte[] readBytes = storage.readAllBytes(bucketName, blobName);
     assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
   }
@@ -345,21 +136,19 @@ public class ITKmsTest {
         blob,
         BLOB_BYTE_CONTENT,
         Storage.BlobTargetOption.encryptionKey(KEY),
-        Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+        Storage.BlobTargetOption.kmsKeyName(kms.getKey1().getName()));
   }
 
   @Test
   public void testCreateBlobWithDefaultKmsKeyName() {
     String bucketName = generator.randomBucketName();
-    // TODO: replace with storage
-    // b/246634709
     Bucket bucket =
         storage.create(
             BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
+                .setDefaultKmsKeyName(kms.getKey1().getName())
+                .setLocation(kms.getKeyRingLocation())
                 .build());
-    assertEquals(bucket.getDefaultKmsKeyName(), kmsKeyOneResourcePath);
+    assertEquals(bucket.getDefaultKmsKeyName(), kms.getKey1().getName());
 
     try {
       String blobName = "test-create-with-default-kms-key-name-blob";
@@ -369,7 +158,7 @@ public class ITKmsTest {
       assertEquals(blob.getBucket(), remoteBlob.getBucket());
       assertEquals(blob.getName(), remoteBlob.getName());
       assertNotNull(remoteBlob.getKmsKeyName());
-      assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+      assertTrue(remoteBlob.getKmsKeyName().startsWith(kms.getKey1().getName()));
       byte[] readBytes = storage.readAllBytes(bucketName, blobName);
       assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
     } finally {
@@ -381,58 +170,17 @@ public class ITKmsTest {
   public void testGetBlobKmsKeyNameField() {
     String blobName = "test-get-selected-kms-key-name-field-blob";
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).setContentType(CONTENT_TYPE).build();
-    assertNotNull(storage.create(blob, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath)));
+    assertNotNull(
+        storage.create(blob, Storage.BlobTargetOption.kmsKeyName(kms.getKey1().getName())));
     Blob remoteBlob =
         storage.get(blob.getBlobId(), Storage.BlobGetOption.fields(BlobField.KMS_KEY_NAME));
     assertEquals(blob.getBlobId(), remoteBlob.getBlobId());
-    assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    assertTrue(remoteBlob.getKmsKeyName().startsWith(kms.getKey1().getName()));
     assertNull(remoteBlob.getContentType());
   }
 
-  @Test(timeout = 5000)
-  public void testListBlobsKmsKeySelectedFields() throws InterruptedException {
-    String[] blobNames = {
-      "test-list-blobs-selected-field-kms-key-name-blob1",
-      "test-list-blobs-selected-field-kms-key-name-blob2"
-    };
-    BlobInfo blob1 = BlobInfo.newBuilder(bucket, blobNames[0]).setContentType(CONTENT_TYPE).build();
-    BlobInfo blob2 = BlobInfo.newBuilder(bucket, blobNames[1]).setContentType(CONTENT_TYPE).build();
-    Blob remoteBlob1 =
-        storage.create(blob1, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-    Blob remoteBlob2 =
-        storage.create(blob2, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-    assertNotNull(remoteBlob1);
-    assertNotNull(remoteBlob2);
-    Page<Blob> page =
-        storage.list(
-            bucket.getName(),
-            Storage.BlobListOption.prefix("test-list-blobs-selected-field-kms-key-name-blob"),
-            Storage.BlobListOption.fields(BlobField.KMS_KEY_NAME));
-    // Listing blobs is eventually consistent, we loop until the list is of the expected size. The
-    // test fails if timeout is reached.
-    while (Iterators.size(page.iterateAll().iterator()) != 2) {
-      Thread.sleep(500);
-      page =
-          storage.list(
-              bucket.getName(),
-              Storage.BlobListOption.prefix("test-list-blobs-selected-field-kms-key-name-blob"),
-              Storage.BlobListOption.fields(BlobField.KMS_KEY_NAME));
-    }
-    Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1]);
-    Iterator<Blob> iterator = page.iterateAll().iterator();
-    while (iterator.hasNext()) {
-      Blob remoteBlob = iterator.next();
-      assertEquals(bucket.getName(), remoteBlob.getBucket());
-      assertTrue(blobSet.contains(remoteBlob.getName()));
-      assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
-      assertNull(remoteBlob.getContentType());
-    }
-  }
-
   @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testRotateFromCustomerEncryptionToKmsKey() {
-    // Bucket attribute extration on allowlist bug b/246634709
     String sourceBlobName = "test-copy-blob-encryption-key-source";
     BlobId source = BlobId.of(bucket.getName(), sourceBlobName);
     ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
@@ -452,14 +200,14 @@ public class ITKmsTest {
         Storage.CopyRequest.newBuilder()
             .setSource(source)
             .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
-            .setTarget(target, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath))
+            .setTarget(target, Storage.BlobTargetOption.kmsKeyName(kms.getKey1().getName()))
             .build();
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
     assertEquals(targetBlobName, copyWriter.getResult().getName());
     assertEquals(CONTENT_TYPE, copyWriter.getResult().getContentType());
     assertNotNull(copyWriter.getResult().getKmsKeyName());
-    assertTrue(copyWriter.getResult().getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    assertTrue(copyWriter.getResult().getKmsKeyName().startsWith(kms.getKey1().getName()));
     assertArrayEquals(BLOB_BYTE_CONTENT, copyWriter.getResult().getContent());
     assertEquals(metadata, copyWriter.getResult().getMetadata());
     assertTrue(copyWriter.isDone());
@@ -490,60 +238,18 @@ public class ITKmsTest {
             .setTarget(
                 target,
                 Storage.BlobTargetOption.encryptionKey(KEY),
-                Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath))
+                Storage.BlobTargetOption.kmsKeyName(kms.getKey1().getName()))
             .build();
     storage.copy(req);
   }
 
-  @Test
-  public void testListBucketDefaultKmsKeyName() throws InterruptedException {
-    String bucketName = generator.randomBucketName();
-    // TODO: replace with storage
-    // b/246634709
-    Bucket remoteBucket =
-        storage.create(
-            BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
-                .build());
-    assertNotNull(remoteBucket);
-    assertTrue(remoteBucket.getDefaultKmsKeyName().startsWith(kmsKeyOneResourcePath));
-    try {
-      Iterator<Bucket> bucketIterator =
-          storage
-              .list(
-                  Storage.BucketListOption.prefix(bucketName),
-                  Storage.BucketListOption.fields(BucketField.ENCRYPTION))
-              .iterateAll()
-              .iterator();
-      while (!bucketIterator.hasNext()) {
-        Thread.sleep(500);
-        bucketIterator =
-            storage
-                .list(
-                    Storage.BucketListOption.prefix(bucketName),
-                    Storage.BucketListOption.fields(BucketField.ENCRYPTION))
-                .iterateAll()
-                .iterator();
-      }
-      while (bucketIterator.hasNext()) {
-        Bucket bucket = bucketIterator.next();
-        assertTrue(bucket.getName().startsWith(bucketName));
-        assertNotNull(bucket.getDefaultKmsKeyName());
-        assertTrue(bucket.getDefaultKmsKeyName().startsWith(kmsKeyOneResourcePath));
-      }
-    } finally {
-      BucketCleaner.doCleanup(bucketName, storage);
-    }
-  }
-
   @Test
   public void testWriterWithKmsKeyName() throws IOException {
     // Write an empty object with a kmsKeyName.
     String blobName = "test-empty-blob";
     BlobInfo blobInfo = BlobInfo.newBuilder(bucket, blobName).build();
     Blob blob =
-        storage.create(blobInfo, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+        storage.create(blobInfo, Storage.BlobTargetOption.kmsKeyName(kms.getKey1().getName()));
 
     // Create a writer using blob that already has metadata received from Storage API.
     int numberOfBytes;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectAclTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectAclTest.java
index 8d2256f5..65a43cab 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectAclTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectAclTest.java
@@ -44,15 +44,14 @@ import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.annotations.ParallelFriendly;
+import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.it.runner.registry.ObjectsFixture;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import org.junit.Before;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -62,7 +61,7 @@ import org.junit.runner.RunWith;
 @ParallelFriendly
 public final class ITObjectAclTest {
 
-  @Rule public final TestName testName = new TestName();
+  @Inject public Generator generator;
 
   @Inject public Storage storage;
 
@@ -325,7 +324,7 @@ public final class ITObjectAclTest {
   }
 
   private BlobId tmpId() {
-    return BlobId.of(bucketInfo.getName(), testName.getMethodName());
+    return BlobId.of(bucketInfo.getName(), generator.randomObjectName());
   }
 
   private Blob tmpObject() {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
index 5538626d..2bcb78e6 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
@@ -37,15 +37,14 @@ import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.annotations.Parameterized;
 import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
 import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider;
+import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.ByteStreams;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -55,7 +54,7 @@ import org.junit.runner.RunWith;
 @Parameterized(ChecksummedTestContentProvider.class)
 public final class ITObjectChecksumSupportTest {
 
-  @Rule public final TestName testName = new TestName();
+  @Inject public Generator generator;
 
   @Inject public Storage storage;
   @Inject public BucketInfo bucket;
@@ -65,6 +64,7 @@ public final class ITObjectChecksumSupportTest {
   @Parameter public ChecksummedTestContent content;
 
   public static final class ChecksummedTestContentProvider implements ParametersProvider {
+
     @Override
     public ImmutableList<?> parameters() {
       DataGenerator gen = DataGenerator.base64Characters();
@@ -83,7 +83,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testCrc32cValidated_createFrom_expectFailure() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
 
@@ -102,7 +102,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testCrc32cValidated_createFrom_expectSuccess() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
 
@@ -118,7 +118,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testCrc32cValidated_writer_expectFailure() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
 
@@ -141,7 +141,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testCrc32cValidated_writer_expectSuccess() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
 
@@ -160,7 +160,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testMd5Validated_createFrom_expectFailure() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setMd5(content.getMd5Base64()).build();
 
@@ -179,7 +179,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testMd5Validated_createFrom_expectSuccess() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setMd5(content.getMd5Base64()).build();
 
@@ -195,7 +195,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testMd5Validated_writer_expectFailure() {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setMd5(content.getMd5Base64()).build();
 
@@ -216,7 +216,7 @@ public final class ITObjectChecksumSupportTest {
 
   @Test
   public void testMd5Validated_writer_expectSuccess() throws IOException {
-    String blobName = testName.getMethodName();
+    String blobName = generator.randomObjectName();
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setMd5(content.getMd5Base64()).build();
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
index 56f0a37c..56f671f9 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
@@ -41,18 +41,25 @@ import com.google.cloud.storage.CopyWriter;
 import com.google.cloud.storage.PackagePrivateMethodWorkarounds;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
 import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.BucketField;
 import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.Storage.ComposeRequest;
+import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.cloud.storage.Storage.PredefinedAcl;
 import com.google.cloud.storage.StorageClass;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
 import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.BucketFixture;
 import com.google.cloud.storage.it.runner.annotations.BucketType;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.CrossRun.Exclude;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.common.collect.ImmutableList;
@@ -75,14 +82,15 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 import javax.crypto.spec.SecretKeySpec;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
@@ -90,6 +98,7 @@ import org.junit.runner.RunWith;
     transports = {Transport.HTTP, Transport.GRPC},
     backends = {Backend.PROD})
 public class ITObjectTest {
+
   private static final String CONTENT_TYPE = "text/plain";
   private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
   private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
@@ -106,18 +115,17 @@ public class ITObjectTest {
   private static final Long RETENTION_PERIOD = 5L;
   private static final Long RETENTION_PERIOD_IN_MILLISECONDS = RETENTION_PERIOD * 1000;
 
-  @Rule public final TestName testName = new TestName();
+  @Inject public Generator generator;
 
   @Inject
-  @com.google.cloud.storage.it.runner.annotations.BucketFixture(BucketType.DEFAULT)
+  @BucketFixture(BucketType.DEFAULT)
   public BucketInfo bucket;
 
   @Inject
-  @com.google.cloud.storage.it.runner.annotations.BucketFixture(BucketType.REQUESTER_PAYS)
+  @BucketFixture(BucketType.REQUESTER_PAYS)
   public BucketInfo requesterPaysBucket;
 
   @Inject public Storage storage;
-  @Inject public Generator generator;
 
   @Test
   public void testCreateBlob() {
@@ -158,20 +166,19 @@ public class ITObjectTest {
   public void testCreateGetBlobWithEncryptionKey() {
     String blobName = "test-create-with-customer-key-blob";
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
-    Blob remoteBlob =
-        storage.create(blob, BLOB_BYTE_CONTENT, Storage.BlobTargetOption.encryptionKey(KEY));
+    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT, BlobTargetOption.encryptionKey(KEY));
     assertNotNull(remoteBlob);
     assertEquals(blob.getBucket(), remoteBlob.getBucket());
     assertEquals(blob.getName(), remoteBlob.getName());
     byte[] readBytes =
         storage.readAllBytes(
-            bucket.getName(), blobName, Storage.BlobSourceOption.decryptionKey(BASE64_KEY));
+            bucket.getName(), blobName, BlobSourceOption.decryptionKey(BASE64_KEY));
     assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
     remoteBlob =
         storage.get(
             blob.getBlobId(),
-            Storage.BlobGetOption.decryptionKey(BASE64_KEY),
-            Storage.BlobGetOption.fields(BlobField.CRC32C, BlobField.MD5HASH));
+            BlobGetOption.decryptionKey(BASE64_KEY),
+            BlobGetOption.fields(BlobField.CRC32C, BlobField.MD5HASH));
     assertNotNull(remoteBlob.getCrc32c());
     assertNotNull(remoteBlob.getMd5());
   }
@@ -226,8 +233,7 @@ public class ITObjectTest {
     assertNotNull(remoteBlob);
     BlobInfo wrongGenerationBlob = BlobInfo.newBuilder(bucket, blobName, -1L).build();
     try {
-      storage.create(
-          wrongGenerationBlob, BLOB_BYTE_CONTENT, Storage.BlobTargetOption.generationMatch());
+      storage.create(wrongGenerationBlob, BLOB_BYTE_CONTENT, BlobTargetOption.generationMatch());
       fail("StorageException was expected");
     } catch (StorageException ex) {
       // expected
@@ -235,21 +241,17 @@ public class ITObjectTest {
   }
 
   @Test
-  // FieldMask on get not supported by GRPC yet.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testGetBlobEmptySelectedFields() {
 
     String blobName = "test-get-empty-selected-fields-blob";
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).setContentType(CONTENT_TYPE).build();
     assertNotNull(storage.create(blob));
-    Blob remoteBlob = storage.get(blob.getBlobId(), Storage.BlobGetOption.fields());
+    Blob remoteBlob = storage.get(blob.getBlobId(), BlobGetOption.fields());
     assertEquals(blob.getBlobId(), remoteBlob.getBlobId());
     assertNull(remoteBlob.getContentType());
   }
 
   @Test
-  // FieldMask on get not supported by GRPC yet.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testGetBlobSelectedFields() {
 
     String blobName = "test-get-selected-fields-blob";
@@ -259,16 +261,13 @@ public class ITObjectTest {
             .setMetadata(ImmutableMap.of("k", "v"))
             .build();
     assertNotNull(storage.create(blob));
-    Blob remoteBlob =
-        storage.get(blob.getBlobId(), Storage.BlobGetOption.fields(BlobField.METADATA));
+    Blob remoteBlob = storage.get(blob.getBlobId(), BlobGetOption.fields(BlobField.METADATA));
     assertEquals(blob.getBlobId(), remoteBlob.getBlobId());
     assertEquals(ImmutableMap.of("k", "v"), remoteBlob.getMetadata());
     assertNull(remoteBlob.getContentType());
   }
 
   @Test
-  // FieldMask on get not supported by GRPC yet.
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testGetBlobAllSelectedFields() {
 
     String blobName = "test-get-all-selected-fields-blob";
@@ -278,13 +277,10 @@ public class ITObjectTest {
             .setMetadata(ImmutableMap.of("k", "v"))
             .build();
     assertNotNull(storage.create(blob));
-    Blob remoteBlob =
-        storage.get(blob.getBlobId(), Storage.BlobGetOption.fields(BlobField.values()));
+    Blob remoteBlob = storage.get(blob.getBlobId(), BlobGetOption.fields(BlobField.values()));
     assertEquals(blob.getBucket(), remoteBlob.getBucket());
     assertEquals(blob.getName(), remoteBlob.getName());
     assertEquals(ImmutableMap.of("k", "v"), remoteBlob.getMetadata());
-    assertNotNull(remoteBlob.getGeneratedId());
-    assertNotNull(remoteBlob.getSelfLink());
   }
 
   @Test
@@ -295,7 +291,7 @@ public class ITObjectTest {
     assertNotNull(remoteBlob);
     BlobId wrongGenerationBlob = BlobId.of(bucket.getName(), blobName);
     try {
-      storage.get(wrongGenerationBlob, Storage.BlobGetOption.generationMatch(-1));
+      storage.get(wrongGenerationBlob, BlobGetOption.generationMatch(-1));
       fail("StorageException was expected");
     } catch (StorageException ex) {
       // expected
@@ -317,92 +313,36 @@ public class ITObjectTest {
     }
   }
 
-  @Test(timeout = 5000)
-  // FieldMask on get not supported by GRPC yet.
-  @CrossRun.Exclude(transports = Transport.GRPC)
-  public void testListBlobsSelectedFields() throws InterruptedException {
+  @Test
+  public void testListBlobsSelectedFields() {
+    String baseName = generator.randomObjectName();
+
+    String name1 = baseName + "1";
+    String name2 = baseName + "2";
 
-    String[] blobNames = {
-      "test-list-blobs-selected-fields-blob1", "test-list-blobs-selected-fields-blob2"
-    };
     ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
-    BlobInfo blob1 =
-        BlobInfo.newBuilder(bucket, blobNames[0])
-            .setContentType(CONTENT_TYPE)
-            .setMetadata(metadata)
-            .build();
-    BlobInfo blob2 =
-        BlobInfo.newBuilder(bucket, blobNames[1])
-            .setContentType(CONTENT_TYPE)
-            .setMetadata(metadata)
-            .build();
+    BlobInfo blob1 = BlobInfo.newBuilder(bucket, name1).setMetadata(metadata).build();
+    BlobInfo blob2 = BlobInfo.newBuilder(bucket, name2).setMetadata(metadata).build();
     Blob remoteBlob1 = storage.create(blob1);
     Blob remoteBlob2 = storage.create(blob2);
-    assertNotNull(remoteBlob1);
-    assertNotNull(remoteBlob2);
+
+    ImmutableSet<Map<String, String>> expected =
+        Stream.of(remoteBlob1, remoteBlob2)
+            .map(BlobInfo::getMetadata)
+            .collect(ImmutableSet.toImmutableSet());
+
     Page<Blob> page =
         storage.list(
             bucket.getName(),
-            Storage.BlobListOption.prefix("test-list-blobs-selected-fields-blob"),
+            Storage.BlobListOption.prefix(baseName),
             Storage.BlobListOption.fields(BlobField.METADATA));
-    // Listing blobs is eventually consistent, we loop until the list is of the expected size. The
-    // test fails if timeout is reached.
-    while (Iterators.size(page.iterateAll().iterator()) != 2) {
-      Thread.sleep(500);
-      page =
-          storage.list(
-              bucket.getName(),
-              Storage.BlobListOption.prefix("test-list-blobs-selected-fields-blob"),
-              Storage.BlobListOption.fields(BlobField.METADATA));
-    }
-    Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1]);
-    Iterator<Blob> iterator = page.iterateAll().iterator();
-    while (iterator.hasNext()) {
-      Blob remoteBlob = iterator.next();
-      assertEquals(bucket.getName(), remoteBlob.getBucket());
-      assertTrue(blobSet.contains(remoteBlob.getName()));
-      assertEquals(metadata, remoteBlob.getMetadata());
-      assertNull(remoteBlob.getContentType());
-    }
-  }
 
-  @Test(timeout = 5000)
-  // FieldMask on get not supported by GRPC yet.
-  @CrossRun.Exclude(transports = Transport.GRPC)
-  public void testListBlobsEmptySelectedFields() throws InterruptedException {
+    ImmutableSet<Blob> blobs = ImmutableSet.copyOf(page.iterateAll());
 
-    String[] blobNames = {
-      "test-list-blobs-empty-selected-fields-blob1", "test-list-blobs-empty-selected-fields-blob2"
-    };
-    BlobInfo blob1 = BlobInfo.newBuilder(bucket, blobNames[0]).setContentType(CONTENT_TYPE).build();
-    BlobInfo blob2 = BlobInfo.newBuilder(bucket, blobNames[1]).setContentType(CONTENT_TYPE).build();
-    Blob remoteBlob1 = storage.create(blob1);
-    Blob remoteBlob2 = storage.create(blob2);
-    assertNotNull(remoteBlob1);
-    assertNotNull(remoteBlob2);
-    Page<Blob> page =
-        storage.list(
-            bucket.getName(),
-            Storage.BlobListOption.prefix("test-list-blobs-empty-selected-fields-blob"),
-            Storage.BlobListOption.fields());
-    // Listing blobs is eventually consistent, we loop until the list is of the expected size. The
-    // test fails if timeout is reached.
-    while (Iterators.size(page.iterateAll().iterator()) != 2) {
-      Thread.sleep(500);
-      page =
-          storage.list(
-              bucket.getName(),
-              Storage.BlobListOption.prefix("test-list-blobs-empty-selected-fields-blob"),
-              Storage.BlobListOption.fields());
-    }
-    Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1]);
-    Iterator<Blob> iterator = page.iterateAll().iterator();
-    while (iterator.hasNext()) {
-      Blob remoteBlob = iterator.next();
-      assertEquals(bucket.getName(), remoteBlob.getBucket());
-      assertTrue(blobSet.contains(remoteBlob.getName()));
-      assertNull(remoteBlob.getContentType());
-    }
+    ImmutableSet<Map<String, String>> actual =
+        blobs.stream().map(BlobInfo::getMetadata).collect(ImmutableSet.toImmutableSet());
+
+    assertThat(actual).isEqualTo(expected);
   }
 
   @Test
@@ -429,7 +369,7 @@ public class ITObjectTest {
     Bucket remoteBucket =
         storage.get(
             requesterPaysBucket.getName(),
-            Storage.BucketGetOption.fields(BucketField.ID, BucketField.BILLING),
+            BucketGetOption.fields(BucketField.ID, BucketField.BILLING),
             BucketGetOption.userProject(projectId));
 
     assertTrue(remoteBucket.requesterPays());
@@ -439,9 +379,9 @@ public class ITObjectTest {
     try {
       storage.list(
           requesterPaysBucket.getName(),
-          Storage.BlobListOption.prefix("test-list-blobs-empty-selected-fields-blob"),
-          Storage.BlobListOption.fields(),
-          Storage.BlobListOption.userProject("fakeBillingProjectId"));
+          BlobListOption.prefix("test-list-blobs-empty-selected-fields-blob"),
+          BlobListOption.fields(),
+          BlobListOption.userProject("fakeBillingProjectId"));
       fail("Expected bad user project error.");
     } catch (StorageException e) {
       assertTrue(e.getMessage().contains("User project specified in the request is invalid"));
@@ -480,8 +420,8 @@ public class ITObjectTest {
       Page<Blob> page =
           storage.list(
               bucketName,
-              Storage.BlobListOption.prefix("test-list-blobs-versioned-blob"),
-              Storage.BlobListOption.versions(true));
+              BlobListOption.prefix("test-list-blobs-versioned-blob"),
+              BlobListOption.versions(true));
       // https://cloud.google.com/storage/docs/consistency#strongly_consistent_operations
       // enabling versioning on an existing bucket seems to have some backpressure on when new
       // versions can safely be made, but listing is not eventually consistent.
@@ -494,8 +434,8 @@ public class ITObjectTest {
         page =
             storage.list(
                 bucketName,
-                Storage.BlobListOption.prefix("test-list-blobs-versioned-blob"),
-                Storage.BlobListOption.versions(true));
+                BlobListOption.prefix("test-list-blobs-versioned-blob"),
+                BlobListOption.versions(true));
       }
       Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1]);
       Iterator<Blob> iterator = page.iterateAll().iterator();
@@ -537,69 +477,95 @@ public class ITObjectTest {
       assertEquals(3, Iterators.size(page1.iterateAll().iterator()));
 
       // Listing blobs with startOffset.
-      Page<Blob> page2 =
-          storage.list(bucketName, Storage.BlobListOption.startOffset("startOffset"));
+      Page<Blob> page2 = storage.list(bucketName, BlobListOption.startOffset("startOffset"));
       assertEquals(2, Iterators.size(page2.iterateAll().iterator()));
 
       // Listing blobs with endOffset.
-      Page<Blob> page3 = storage.list(bucketName, Storage.BlobListOption.endOffset("endOffset"));
+      Page<Blob> page3 = storage.list(bucketName, BlobListOption.endOffset("endOffset"));
       assertEquals(1, Iterators.size(page3.iterateAll().iterator()));
 
       // Listing blobs with startOffset and endOffset.
       Page<Blob> page4 =
           storage.list(
               bucketName,
-              Storage.BlobListOption.startOffset("startOffset"),
-              Storage.BlobListOption.endOffset("endOffset"));
+              BlobListOption.startOffset("startOffset"),
+              BlobListOption.endOffset("endOffset"));
       assertEquals(0, Iterators.size(page4.iterateAll().iterator()));
     } finally {
       BucketCleaner.doCleanup(bucketName, storage);
     }
   }
 
-  @Test(timeout = 5000)
-  // This test is currently timing out for GRPC
-  @CrossRun.Exclude(transports = Transport.GRPC)
-  public void testListBlobsCurrentDirectory() throws InterruptedException {
+  @Test
+  public void testListBlobsCurrentDirectoryIncludesBothObjectsAndSyntheticDirectories() {
+    String bucketName = bucket.getName();
+    String directoryName = generator.randomObjectName();
+    String subdirectoryName = "subdirectory";
+
+    String uriSubDir = String.format("gs://%s/%s/%s/", bucketName, directoryName, subdirectoryName);
+    String uri1 = String.format("gs://%s/%s/%s/blob1", bucketName, directoryName, subdirectoryName);
+    String uri2 = String.format("gs://%s/%s/blob2", bucketName, directoryName);
 
-    String directoryName = "test-list-blobs-current-directory/";
-    String subdirectoryName = "subdirectory/";
-    String[] blobNames = {directoryName + subdirectoryName + "blob1", directoryName + "blob2"};
-    BlobInfo blob1 = BlobInfo.newBuilder(bucket, blobNames[0]).setContentType(CONTENT_TYPE).build();
-    BlobInfo blob2 = BlobInfo.newBuilder(bucket, blobNames[1]).setContentType(CONTENT_TYPE).build();
-    Blob remoteBlob1 = storage.create(blob1, BLOB_BYTE_CONTENT);
-    Blob remoteBlob2 = storage.create(blob2, BLOB_BYTE_CONTENT);
-    assertNotNull(remoteBlob1);
-    assertNotNull(remoteBlob2);
-    Page<Blob> page =
+    BlobId id1 = BlobId.fromGsUtilUri(uri1);
+    BlobId id2 = BlobId.fromGsUtilUri(uri2);
+    BlobId idSubDir = BlobId.fromGsUtilUri(uriSubDir);
+
+    BlobInfo blob1 = BlobInfo.newBuilder(id1).build();
+    BlobInfo blob2 = BlobInfo.newBuilder(id2).build();
+    BlobInfo obj1Gen1 = storage.create(blob1, BLOB_BYTE_CONTENT).asBlobInfo();
+    BlobInfo obj2Gen1 = storage.create(blob2, BLOB_BYTE_CONTENT).asBlobInfo();
+
+    Page<Blob> page1 =
         storage.list(
-            bucket.getName(),
-            Storage.BlobListOption.prefix("test-list-blobs-current-directory/"),
-            Storage.BlobListOption.currentDirectory());
-    // Listing blobs is eventually consistent, we loop until the list is of the expected size. The
-    // test fails if timeout is reached.
-    while (Iterators.size(page.iterateAll().iterator()) != 2) {
-      Thread.sleep(500);
-      page =
-          storage.list(
-              bucket.getName(),
-              Storage.BlobListOption.prefix("test-list-blobs-current-directory/"),
-              Storage.BlobListOption.currentDirectory());
-    }
-    Iterator<Blob> iterator = page.iterateAll().iterator();
-    while (iterator.hasNext()) {
-      Blob remoteBlob = iterator.next();
-      assertEquals(bucket.getName(), remoteBlob.getBucket());
-      if (remoteBlob.getName().equals(blobNames[1])) {
-        assertEquals(CONTENT_TYPE, remoteBlob.getContentType());
-        assertEquals(BLOB_BYTE_CONTENT.length, (long) remoteBlob.getSize());
-        assertFalse(remoteBlob.isDirectory());
-      } else if (remoteBlob.getName().equals(directoryName + subdirectoryName)) {
-        assertEquals(0L, (long) remoteBlob.getSize());
-        assertTrue(remoteBlob.isDirectory());
-      } else {
-        fail("Unexpected blob with name " + remoteBlob.getName());
-      }
+            bucketName,
+            BlobListOption.prefix(directoryName + "/"),
+            BlobListOption.currentDirectory());
+
+    ImmutableList<Blob> blobs = ImmutableList.copyOf(page1.iterateAll());
+
+    ImmutableSet<BlobInfo> actual =
+        blobs.stream()
+            .map(Blob::asBlobInfo)
+            .map(info -> PackagePrivateMethodWorkarounds.noAcl(info))
+            .collect(ImmutableSet.toImmutableSet());
+
+    // obj1Gen1 is "in subdirectory" and we don't expect to receive it as a result when listing
+    // object in "the current directory"
+    assertThat(actual).doesNotContain(obj1Gen1);
+
+    // make sure one of the results we received is the "subdirectory" blob1 is "in"
+    Optional<BlobInfo> first = actual.stream().filter(BlobInfo::isDirectory).findFirst();
+    assertThat(first.isPresent()).isTrue();
+    assertThat(first.get().getBlobId()).isEqualTo(idSubDir);
+
+    assertThat(actual).contains(PackagePrivateMethodWorkarounds.noAcl(obj2Gen1));
+  }
+
+  @Test
+  public void testListBlobsMultiplePages() {
+    String basePath = generator.randomObjectName();
+
+    ImmutableList<BlobInfo> expected =
+        IntStream.rangeClosed(1, 10)
+            .mapToObj(i -> String.format("%s/%2d", basePath, i))
+            .map(name -> BlobInfo.newBuilder(bucket, name).build())
+            .map(info -> storage.create(info, BlobTargetOption.doesNotExist()))
+            .map(info1 -> PackagePrivateMethodWorkarounds.noAcl(info1))
+            .collect(ImmutableList.toImmutableList());
+
+    Page<Blob> page =
+        storage.list(bucket.getName(), BlobListOption.prefix(basePath), BlobListOption.pageSize(3));
+
+    ImmutableList<BlobInfo> actual =
+        ImmutableList.copyOf(page.iterateAll()).stream()
+            .map(info -> PackagePrivateMethodWorkarounds.noAcl(info))
+            .collect(ImmutableList.toImmutableList());
+
+    try {
+      assertThat(actual).isEqualTo(expected);
+    } finally {
+      // delete all the objects we created
+      expected.stream().map(BlobInfo::getBlobId).forEach(storage::delete);
     }
   }
 
@@ -659,7 +625,7 @@ public class ITObjectTest {
 
   @Test
   // Metadata update bug b/230510191
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testUpdateBlobUnsetMetadata() {
 
     String blobName = "test-update-blob-unset-metadata";
@@ -691,7 +657,7 @@ public class ITObjectTest {
     BlobInfo wrongGenerationBlob =
         BlobInfo.newBuilder(bucket, blobName, -1L).setContentType(CONTENT_TYPE).build();
     try {
-      storage.update(wrongGenerationBlob, Storage.BlobTargetOption.generationMatch());
+      storage.update(wrongGenerationBlob, BlobTargetOption.generationMatch());
       fail("StorageException was expected");
     } catch (StorageException ex) {
       // expected
@@ -724,8 +690,7 @@ public class ITObjectTest {
     Blob remoteBlob = storage.create(blob);
     assertNotNull(remoteBlob);
     try {
-      storage.delete(
-          bucket.getName(), blob.getName(), Storage.BlobSourceOption.generationMatch(-1L));
+      storage.delete(bucket.getName(), blob.getName(), BlobSourceOption.generationMatch(-1L));
       fail("StorageException was expected");
     } catch (StorageException ex) {
       // expected
@@ -745,8 +710,8 @@ public class ITObjectTest {
     assertNotNull(remoteSourceBlob2);
     String targetBlobName = "test-compose-blob-target";
     BlobInfo targetBlob = BlobInfo.newBuilder(bucket, targetBlobName).build();
-    Storage.ComposeRequest req =
-        Storage.ComposeRequest.of(ImmutableList.of(sourceBlobName1, sourceBlobName2), targetBlob);
+    ComposeRequest req =
+        ComposeRequest.of(ImmutableList.of(sourceBlobName1, sourceBlobName2), targetBlob);
     Blob remoteTargetBlob = storage.compose(req);
     assertNotNull(remoteTargetBlob);
     assertEquals(targetBlob.getName(), remoteTargetBlob.getName());
@@ -771,8 +736,8 @@ public class ITObjectTest {
     String targetBlobName = "test-compose-blob-with-content-type-target";
     BlobInfo targetBlob =
         BlobInfo.newBuilder(bucket, targetBlobName).setContentType(CONTENT_TYPE).build();
-    Storage.ComposeRequest req =
-        Storage.ComposeRequest.of(ImmutableList.of(sourceBlobName1, sourceBlobName2), targetBlob);
+    ComposeRequest req =
+        ComposeRequest.of(ImmutableList.of(sourceBlobName1, sourceBlobName2), targetBlob);
     Blob remoteTargetBlob = storage.compose(req);
     assertNotNull(remoteTargetBlob);
     assertEquals(targetBlob.getName(), remoteTargetBlob.getName());
@@ -797,8 +762,8 @@ public class ITObjectTest {
     assertNotNull(remoteSourceBlob2);
     String targetBlobName = "test-compose-blob-fail-target";
     BlobInfo targetBlob = BlobInfo.newBuilder(bucket, targetBlobName).build();
-    Storage.ComposeRequest req =
-        Storage.ComposeRequest.newBuilder()
+    ComposeRequest req =
+        ComposeRequest.newBuilder()
             .addSource(sourceBlobName1, -1L)
             .addSource(sourceBlobName2, -1L)
             .setTarget(targetBlob)
@@ -813,7 +778,7 @@ public class ITObjectTest {
 
   @Test
   // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlob() {
 
     String sourceBlobName = "test-copy-blob-source";
@@ -824,8 +789,7 @@ public class ITObjectTest {
     Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
     assertNotNull(remoteBlob);
     String targetBlobName = "test-copy-blob-target";
-    Storage.CopyRequest req =
-        Storage.CopyRequest.of(source, BlobId.of(bucket.getName(), targetBlobName));
+    CopyRequest req = CopyRequest.of(source, BlobId.of(bucket.getName(), targetBlobName));
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
     assertEquals(targetBlobName, copyWriter.getResult().getName());
@@ -838,7 +802,7 @@ public class ITObjectTest {
 
   @Test
   // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlobWithPredefinedAcl() {
 
     String sourceBlobName = "test-copy-blob-source";
@@ -849,12 +813,12 @@ public class ITObjectTest {
     Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
     assertNotNull(remoteBlob);
     String targetBlobName = "test-copy-blob-target";
-    Storage.CopyRequest req =
-        Storage.CopyRequest.newBuilder()
+    CopyRequest req =
+        CopyRequest.newBuilder()
             .setSource(source)
             .setTarget(
                 BlobId.of(bucket.getName(), targetBlobName),
-                Storage.BlobTargetOption.predefinedAcl(Storage.PredefinedAcl.PUBLIC_READ))
+                BlobTargetOption.predefinedAcl(PredefinedAcl.PUBLIC_READ))
             .build();
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
@@ -869,7 +833,7 @@ public class ITObjectTest {
 
   @Test
   // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlobWithEncryptionKeys() {
 
     String sourceBlobName = "test-copy-blob-encryption-key-source";
@@ -879,7 +843,7 @@ public class ITObjectTest {
         storage.create(
             BlobInfo.newBuilder(source).build(),
             BLOB_BYTE_CONTENT,
-            Storage.BlobTargetOption.encryptionKey(KEY));
+            BlobTargetOption.encryptionKey(KEY));
     assertNotNull(remoteBlob);
     String targetBlobName = "test-copy-blob-encryption-key-target";
     BlobInfo target =
@@ -887,11 +851,11 @@ public class ITObjectTest {
             .setContentType(CONTENT_TYPE)
             .setMetadata(metadata)
             .build();
-    Storage.CopyRequest req =
-        Storage.CopyRequest.newBuilder()
+    CopyRequest req =
+        CopyRequest.newBuilder()
             .setSource(source)
-            .setTarget(target, Storage.BlobTargetOption.encryptionKey(OTHER_BASE64_KEY))
-            .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
+            .setTarget(target, BlobTargetOption.encryptionKey(OTHER_BASE64_KEY))
+            .setSourceOptions(BlobSourceOption.decryptionKey(BASE64_KEY))
             .build();
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
@@ -903,10 +867,10 @@ public class ITObjectTest {
     assertEquals(metadata, copyWriter.getResult().getMetadata());
     assertTrue(copyWriter.isDone());
     req =
-        Storage.CopyRequest.newBuilder()
+        CopyRequest.newBuilder()
             .setSource(source)
             .setTarget(target)
-            .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
+            .setSourceOptions(BlobSourceOption.decryptionKey(BASE64_KEY))
             .build();
     copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
@@ -921,7 +885,7 @@ public class ITObjectTest {
 
   @Test
   // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlobUpdateMetadata() {
 
     String sourceBlobName = "test-copy-blob-update-metadata-source";
@@ -935,7 +899,7 @@ public class ITObjectTest {
             .setContentType(CONTENT_TYPE)
             .setMetadata(metadata)
             .build();
-    Storage.CopyRequest req = Storage.CopyRequest.of(source, target);
+    CopyRequest req = CopyRequest.of(source, target);
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
     assertEquals(targetBlobName, copyWriter.getResult().getName());
@@ -948,7 +912,7 @@ public class ITObjectTest {
 
   // Re-enable this test when it stops failing
   // @Test
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlobUpdateStorageClass() {
     String sourceBlobName = "test-copy-blob-update-storage-class-source";
     BlobId source = BlobId.of(bucket.getName(), sourceBlobName);
@@ -961,7 +925,7 @@ public class ITObjectTest {
     String targetBlobName = "test-copy-blob-update-storage-class-target";
     BlobInfo targetInfo =
         BlobInfo.newBuilder(bucket, targetBlobName).setStorageClass(StorageClass.COLDLINE).build();
-    Storage.CopyRequest req = Storage.CopyRequest.of(source, targetInfo);
+    CopyRequest req = CopyRequest.of(source, targetInfo);
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
     assertEquals(targetBlobName, copyWriter.getResult().getName());
@@ -973,7 +937,7 @@ public class ITObjectTest {
 
   @Test
   // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlobNoContentType() {
 
     String sourceBlobName = "test-copy-blob-no-content-type-source";
@@ -983,7 +947,7 @@ public class ITObjectTest {
     String targetBlobName = "test-copy-blob-no-content-type-target";
     ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
     BlobInfo target = BlobInfo.newBuilder(bucket, targetBlobName).setMetadata(metadata).build();
-    Storage.CopyRequest req = Storage.CopyRequest.of(source, target);
+    CopyRequest req = CopyRequest.of(source, target);
     CopyWriter copyWriter = storage.copy(req);
     assertEquals(bucket.getName(), copyWriter.getResult().getBucket());
     assertEquals(targetBlobName, copyWriter.getResult().getName());
@@ -997,7 +961,7 @@ public class ITObjectTest {
   @Test
   // Verified against testbench
   // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testCopyBlobFail() {
 
     String sourceBlobName = "test-copy-blob-source-fail";
@@ -1007,10 +971,10 @@ public class ITObjectTest {
     String targetBlobName = "test-copy-blob-target-fail";
     BlobInfo target =
         BlobInfo.newBuilder(bucket, targetBlobName).setContentType(CONTENT_TYPE).build();
-    Storage.CopyRequest req =
-        Storage.CopyRequest.newBuilder()
+    CopyRequest req =
+        CopyRequest.newBuilder()
             .setSource(bucket.getName(), sourceBlobName)
-            .setSourceOptions(Storage.BlobSourceOption.generationMatch(-1L))
+            .setSourceOptions(BlobSourceOption.generationMatch(-1L))
             .setTarget(target)
             .build();
     try {
@@ -1019,10 +983,10 @@ public class ITObjectTest {
     } catch (StorageException ex) {
       // expected
     }
-    Storage.CopyRequest req2 =
-        Storage.CopyRequest.newBuilder()
+    CopyRequest req2 =
+        CopyRequest.newBuilder()
             .setSource(source)
-            .setSourceOptions(Storage.BlobSourceOption.generationMatch())
+            .setSourceOptions(BlobSourceOption.generationMatch())
             .setTarget(target)
             .build();
     try {
@@ -1038,8 +1002,7 @@ public class ITObjectTest {
     String blobName = "test-read-write-channel-with-customer-key-blob";
     BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
     byte[] stringBytes;
-    try (WriteChannel writer =
-        storage.writer(blob, Storage.BlobWriteOption.encryptionKey(BASE64_KEY))) {
+    try (WriteChannel writer = storage.writer(blob, BlobWriteOption.encryptionKey(BASE64_KEY))) {
       stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
       writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
       writer.write(ByteBuffer.wrap(stringBytes));
@@ -1047,7 +1010,7 @@ public class ITObjectTest {
     ByteBuffer readBytes;
     ByteBuffer readStringBytes;
     try (ReadChannel reader =
-        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.decryptionKey(KEY))) {
+        storage.reader(blob.getBlobId(), BlobSourceOption.decryptionKey(KEY))) {
       readBytes = ByteBuffer.allocate(BLOB_BYTE_CONTENT.length);
       readStringBytes = ByteBuffer.allocate(stringBytes.length);
       reader.read(readBytes);
@@ -1113,7 +1076,7 @@ public class ITObjectTest {
 
   @Test
   // Capture not implemented yet
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testReadAndWriteCaptureChannels() throws IOException {
 
     String blobName = "test-read-and-write-capture-channels-blob";
@@ -1145,7 +1108,7 @@ public class ITObjectTest {
 
   @Test
   // Only supported in JSON right now
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testGetBlobs() {
     String sourceBlobName1 = "test-get-blobs-1";
     String sourceBlobName2 = "test-get-blobs-2";
@@ -1162,7 +1125,7 @@ public class ITObjectTest {
 
   @Test
   // Only supported in JSON right now
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testGetBlobsFail() {
 
     String sourceBlobName1 = "test-get-blobs-fail-1";
@@ -1178,7 +1141,7 @@ public class ITObjectTest {
 
   @Test
   // Only supported in JSON right now
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testDeleteBlobs() {
 
     String sourceBlobName1 = "test-delete-blobs-1";
@@ -1194,7 +1157,7 @@ public class ITObjectTest {
 
   @Test
   // Only supported in JSON right now
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testDeleteBlobsFail() {
     String sourceBlobName1 = "test-delete-blobs-fail-1";
     String sourceBlobName2 = "test-delete-blobs-fail-2";
@@ -1217,7 +1180,7 @@ public class ITObjectTest {
 
   @Test
   // Only supported in JSON right now
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testUpdateBlobs() {
 
     String sourceBlobName1 = "test-update-blobs-1";
@@ -1242,7 +1205,7 @@ public class ITObjectTest {
 
   @Test
   // Only supported in JSON right now
-  @CrossRun.Exclude(transports = Transport.GRPC)
+  @Exclude(transports = Transport.GRPC)
   public void testUpdateBlobsFail() {
 
     String sourceBlobName1 = "test-update-blobs-fail-1";
@@ -1291,7 +1254,7 @@ public class ITObjectTest {
     Blob remoteBlob = storage.create(blobInfo);
     assertTrue(remoteBlob.getTemporaryHold());
     remoteBlob =
-        storage.get(remoteBlob.getBlobId(), Storage.BlobGetOption.fields(BlobField.TEMPORARY_HOLD));
+        storage.get(remoteBlob.getBlobId(), BlobGetOption.fields(BlobField.TEMPORARY_HOLD));
     assertTrue(remoteBlob.getTemporaryHold());
     remoteBlob = remoteBlob.toBuilder().setTemporaryHold(false).build().update();
     assertFalse(remoteBlob.getTemporaryHold());
@@ -1365,7 +1328,7 @@ public class ITObjectTest {
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
 
     ByteArrayInputStream content = new ByteArrayInputStream(BLOB_BYTE_CONTENT);
-    Blob blob = storage.createFrom(blobInfo, content, Storage.BlobWriteOption.encryptionKey(KEY));
+    Blob blob = storage.createFrom(blobInfo, content, BlobWriteOption.encryptionKey(KEY));
 
     try {
       blob.getContent();
@@ -1384,16 +1347,16 @@ public class ITObjectTest {
     switch (method) {
       case "create":
         return detectType
-            ? storage.create(blobInfo, Storage.BlobTargetOption.detectContentType())
+            ? storage.create(blobInfo, BlobTargetOption.detectContentType())
             : storage.create(blobInfo);
       case "createFrom":
         InputStream inputStream = new ByteArrayInputStream(BLOB_BYTE_CONTENT);
         return detectType
-            ? storage.createFrom(blobInfo, inputStream, Storage.BlobWriteOption.detectContentType())
+            ? storage.createFrom(blobInfo, inputStream, BlobWriteOption.detectContentType())
             : storage.createFrom(blobInfo, inputStream);
       case "writer":
         if (detectType) {
-          storage.writer(blobInfo, Storage.BlobWriteOption.detectContentType()).close();
+          storage.writer(blobInfo, BlobWriteOption.detectContentType()).close();
         } else {
           storage.writer(blobInfo).close();
         }
@@ -1438,8 +1401,6 @@ public class ITObjectTest {
   }
 
   @Test
-  // Bucket attribute extration on allowlist bug b/246634709
-  @CrossRun.Exclude(transports = Transport.GRPC)
   public void testBlobTimeStorageClassUpdated() {
 
     String blobName = "test-blob-with-storage-class";
@@ -1456,8 +1417,8 @@ public class ITObjectTest {
     // We can't change an object's storage class directly, the only way is to rewrite the object
     // with the desired storage class.
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
-    Storage.CopyRequest request =
-        Storage.CopyRequest.newBuilder()
+    CopyRequest request =
+        CopyRequest.newBuilder()
             .setSource(blobId)
             .setTarget(BlobInfo.newBuilder(blobId).setStorageClass(StorageClass.STANDARD).build())
             .build();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITOptionRegressionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITOptionRegressionTest.java
index b3846dfe..0ca27c52 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITOptionRegressionTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITOptionRegressionTest.java
@@ -16,7 +16,7 @@
 
 package com.google.cloud.storage.it;
 
-import static org.junit.Assume.assumeTrue;
+import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.storage.Blob;
@@ -929,7 +929,10 @@ public final class ITOptionRegressionTest {
 
   @Test
   public void bucket_BucketSourceOption_userProject_String() {
-    assumeTrue(false);
+    // attempt to delete a bucket that doesn't exist
+    boolean delete = s.delete(bucketName(), BucketSourceOption.userProject("proj"));
+    assertThat(delete).isFalse();
+    requestAuditing.assertQueryParam("userProject", "proj");
   }
 
   @Test
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java
index 754ebf9c..814c5da6 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage.it.runner;
 
 import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.CrossRun.AllowClassRule;
 import com.google.cloud.storage.it.runner.annotations.ParallelFriendly;
 import com.google.cloud.storage.it.runner.annotations.Parameterized;
 import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
@@ -31,6 +32,7 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 import java.util.stream.Stream;
+import org.junit.ClassRule;
 import org.junit.runner.Description;
 import org.junit.runner.Runner;
 import org.junit.runner.manipulation.Filter;
@@ -136,9 +138,13 @@ public final class StorageITRunner extends Suite {
     }
 
     if (crossRun != null) {
-      // TODO
-      //   add warning message if @ClassRule or @BeforeClass/AfterClass are used, as these will run
-      //   multiple times due to the class being run for each crossRun result.
+      List<FrameworkField> classRules = testClass.getAnnotatedFields(ClassRule.class);
+      AllowClassRule allowClassRule = testClass.getAnnotation(AllowClassRule.class);
+      if (allowClassRule == null && !classRules.isEmpty()) {
+        String msg =
+            "@CrossRun used along with @ClassRule. This can be dangerous, multiple class scopes will be created for cross running, possibly breaking expectations on rule scope. If the use of a @ClassRule is still desirable, please annotate your class with @CrossRun.AllowClassRule";
+        throw new InitializationError(msg);
+      }
       return SneakyException.unwrap(
           () ->
               ImmutableSet.copyOf(crossRun.backends()).stream()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/annotations/CrossRun.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/annotations/CrossRun.java
index c5a0d48e..960e56e9 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/annotations/CrossRun.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/annotations/CrossRun.java
@@ -70,4 +70,18 @@ public @interface CrossRun {
 
     TransportCompatibility.Transport[] transports() default {};
   }
+
+  /**
+   * When using {@link CrossRun} a class scope will be created for each permutation, this can break
+   * expectations of scope/lifecycle for {@link org.junit.ClassRule}s. In an abundance of caution,
+   * we consider the use of a {@link org.junit.ClassRule} along with {@link CrossRun} an invalid
+   * class definition.
+   *
+   * <p>In order to allow the use of a {@link org.junit.ClassRule} along with the caveats mentioned
+   * above, a class can be annotated with {@link AllowClassRule} to suppress the error and proceed
+   * running the test class with the rule.
+   */
+  @Target({ElementType.TYPE})
+  @Retention(RetentionPolicy.RUNTIME)
+  @interface AllowClassRule {}
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
index 59699b8c..3f37bb3c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
@@ -46,7 +46,8 @@ final class BackendResources implements ManagedLifecycle {
       TestRunScopedInstance<StorageInstance> storageGrpc,
       TestRunScopedInstance<BucketInfoShim> bucket,
       TestRunScopedInstance<BucketInfoShim> bucketRequesterPays,
-      TestRunScopedInstance<ObjectsFixture> objectsFixture) {
+      TestRunScopedInstance<ObjectsFixture> objectsFixture,
+      TestRunScopedInstance<KmsFixture> kmsFixture) {
     this.backend = backend;
     this.protectedBucketNames = protectedBucketNames;
     this.registryEntries =
@@ -62,7 +63,8 @@ final class BackendResources implements ManagedLifecycle {
                 backendIs(backend).and(isRequesterPaysBucket())),
             RegistryEntry.of(
                 7, BucketInfo.class, bucket, backendIs(backend).and(isDefaultBucket())),
-            RegistryEntry.of(8, ObjectsFixture.class, objectsFixture, backendIs(backend)));
+            RegistryEntry.of(8, ObjectsFixture.class, objectsFixture, backendIs(backend)),
+            RegistryEntry.of(9, KmsFixture.class, kmsFixture, backendIs(backend)));
   }
 
   public ImmutableList<RegistryEntry<?>> getRegistryEntries() {
@@ -139,8 +141,18 @@ final class BackendResources implements ManagedLifecycle {
         TestRunScopedInstance.of(
             "OBJECTS_FIXTURE_" + backend.name(),
             () -> new ObjectsFixture(storageJson.get().getStorage(), bucket.get().getBucketInfo()));
+    TestRunScopedInstance<KmsFixture> kmsFixture =
+        TestRunScopedInstance.of(
+            "KMS_FIXTURE_" + backend.name(), () -> KmsFixture.of(storageJson.get().getStorage()));
 
     return new BackendResources(
-        backend, protectedBucketNames, storageJson, storageGrpc, bucket, bucketRp, objectsFixture);
+        backend,
+        protectedBucketNames,
+        storageJson,
+        storageGrpc,
+        bucket,
+        bucketRp,
+        objectsFixture,
+        kmsFixture);
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java
index 3ffd383e..383053fa 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java
@@ -17,20 +17,37 @@
 package com.google.cloud.storage.it.runner.registry;
 
 import java.util.UUID;
+import java.util.WeakHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.junit.runner.Description;
 
 /** Scoped instance of a class which generates common things for tests. */
 public final class Generator implements ManagedLifecycle {
 
-  /**
-   * Generate a new random bucket name
-   *
-   * @return
-   */
+  private final WeakHashMap<Description, AtomicInteger> counters;
+
+  Generator() {
+    counters = new WeakHashMap<>();
+  }
+
+  /** Generate a new random bucket name */
+  @NonNull
   public String randomBucketName() {
     // TODO: track their creation and detect if the bucket is "leaked" and fail the test
     return "java-storage-grpc-rand-" + UUID.randomUUID();
   }
 
+  @NonNull
+  public String randomObjectName() {
+    Description currentTest = Registry.getInstance().getCurrentTest();
+    if (currentTest == null) {
+      throw new IllegalStateException("No actively running test in registry.");
+    }
+    AtomicInteger counter = counters.computeIfAbsent(currentTest, (d) -> new AtomicInteger(1));
+    return String.format("%s-%04d", currentTest.getMethodName(), counter.getAndIncrement());
+  }
+
   @Override
   public Object get() {
     return this;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/KmsFixture.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/KmsFixture.java
new file mode 100644
index 00000000..e7099da0
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/KmsFixture.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it.runner.registry;
+
+import com.google.api.gax.rpc.NotFoundException;
+import com.google.cloud.kms.v1.CreateCryptoKeyRequest;
+import com.google.cloud.kms.v1.CreateKeyRingRequest;
+import com.google.cloud.kms.v1.CryptoKey;
+import com.google.cloud.kms.v1.CryptoKey.CryptoKeyPurpose;
+import com.google.cloud.kms.v1.CryptoKeyName;
+import com.google.cloud.kms.v1.CryptoKeyVersion.CryptoKeyVersionAlgorithm;
+import com.google.cloud.kms.v1.CryptoKeyVersionTemplate;
+import com.google.cloud.kms.v1.KeyManagementServiceClient;
+import com.google.cloud.kms.v1.KeyRing;
+import com.google.cloud.kms.v1.KeyRingName;
+import com.google.cloud.kms.v1.LocationName;
+import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.Storage;
+import com.google.iam.v1.Binding;
+import com.google.iam.v1.Policy;
+import com.google.iam.v1.SetIamPolicyRequest;
+import io.grpc.StatusRuntimeException;
+import java.io.IOException;
+
+public class KmsFixture implements ManagedLifecycle {
+
+  private final Storage storage;
+  private final String keyRingLocation;
+  private final String keyRingName;
+  private final String key1Name;
+  private final String key2Name;
+
+  private KeyRing keyRing;
+  private CryptoKey key1;
+  private CryptoKey key2;
+
+  private KmsFixture(
+      Storage storage,
+      String keyRingLocation,
+      String keyRingName,
+      String key1Name,
+      String key2Name) {
+    this.storage = storage;
+    this.keyRingLocation = keyRingLocation;
+    this.keyRingName = keyRingName;
+    this.key1Name = key1Name;
+    this.key2Name = key2Name;
+  }
+
+  public String getKeyRingLocation() {
+    return keyRingLocation;
+  }
+
+  public KeyRing getKeyRing() {
+    return keyRing;
+  }
+
+  public CryptoKey getKey1() {
+    return key1;
+  }
+
+  public CryptoKey getKey2() {
+    return key2;
+  }
+
+  @Override
+  public Object get() {
+    return this;
+  }
+
+  @Override
+  public void start() {
+    try (KeyManagementServiceClient kms = KeyManagementServiceClient.create()) {
+      keyRing = resolveKeyRing(kms);
+      Policy ignore = grantStorageServiceAccountRolesToKeyRing(kms);
+      key1 = resolveKey(kms, key1Name);
+      key2 = resolveKey(kms, key2Name);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public void stop() {
+    // KMS prevents key and ring deletion, https://cloud.google.com/kms/docs/faq#cannot_delete
+  }
+
+  static KmsFixture of(Storage s) {
+    // KMS prevents key and ring deletion, https://cloud.google.com/kms/docs/faq#cannot_delete
+    // therefore we instead prefer stable names to not blow out the number of keys and rings
+    // in a project.
+    return new KmsFixture(s, "us", "gcs_test_kms_key_ring", "gcs_kms_key_one", "gcs_kms_key_two");
+  }
+
+  private KeyRing resolveKeyRing(KeyManagementServiceClient kms) throws StatusRuntimeException {
+    String projectId = storage.getOptions().getProjectId();
+    KeyRingName ringName = KeyRingName.of(projectId, keyRingLocation, keyRingName);
+    try {
+      return kms.getKeyRing(ringName.toString());
+    } catch (NotFoundException ex) {
+      CreateKeyRingRequest req =
+          CreateKeyRingRequest.newBuilder()
+              .setParent(LocationName.of(projectId, keyRingLocation).toString())
+              .setKeyRingId(keyRingName)
+              .setKeyRing(KeyRing.newBuilder().build())
+              .build();
+      System.out.println("req = " + req);
+      return kms.createKeyRing(req);
+    }
+  }
+
+  private Policy grantStorageServiceAccountRolesToKeyRing(KeyManagementServiceClient kms) {
+    String projectId = storage.getOptions().getProjectId();
+    ServiceAccount serviceAccount = storage.getServiceAccount(projectId);
+    Binding binding =
+        Binding.newBuilder()
+            .setRole("roles/cloudkms.cryptoKeyEncrypterDecrypter")
+            .addMembers("serviceAccount:" + serviceAccount.getEmail())
+            .build();
+    SetIamPolicyRequest setIamPolicyRequest =
+        SetIamPolicyRequest.newBuilder()
+            .setResource(keyRing.getName())
+            .setPolicy(Policy.newBuilder().addBindings(binding).build())
+            .build();
+
+    return kms.setIamPolicy(setIamPolicyRequest);
+  }
+
+  private CryptoKey resolveKey(KeyManagementServiceClient kms, String keyName) {
+    CryptoKeyName cryptoKeyName = cryptoKeyNameOnRing(keyRing, keyName);
+    try {
+      return kms.getCryptoKey(cryptoKeyName);
+    } catch (NotFoundException ex) {
+      CreateCryptoKeyRequest req =
+          CreateCryptoKeyRequest.newBuilder()
+              .setParent(keyRing.getName())
+              .setCryptoKeyId(keyName)
+              .setCryptoKey(
+                  CryptoKey.newBuilder()
+                      .setPurpose(CryptoKeyPurpose.ENCRYPT_DECRYPT)
+                      .setVersionTemplate(
+                          CryptoKeyVersionTemplate.newBuilder()
+                              .setAlgorithm(CryptoKeyVersionAlgorithm.GOOGLE_SYMMETRIC_ENCRYPTION)))
+              .build();
+      return kms.createCryptoKey(req);
+    }
+  }
+
+  private static CryptoKeyName cryptoKeyNameOnRing(KeyRing keyRing, String keyName) {
+    KeyRingName krn = KeyRingName.parse(keyRing.getName());
+    return CryptoKeyName.of(krn.getProject(), krn.getLocation(), krn.getKeyRing(), keyName);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Registry.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Registry.java
index 1404f02b..e8c43b20 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Registry.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Registry.java
@@ -32,6 +32,8 @@ import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.function.Function;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.runner.Description;
 import org.junit.runner.notification.RunListener;
 import org.junit.runners.model.FrameworkField;
 import org.junit.runners.model.RunnerScheduler;
@@ -85,6 +87,7 @@ public final class Registry extends RunListener {
   private final ImmutableSet<Class<?>> injectableTypes =
       entries.stream().map(RegistryEntry::getType).collect(ImmutableSet.toImmutableSet());
   private final String injectableTypesString = Joiner.on("|").join(injectableTypes);
+  private final ThreadLocal<Description> currentTest = new ThreadLocal<>();
 
   private Registry() {}
 
@@ -112,6 +115,21 @@ public final class Registry extends RunListener {
     return testBench.get();
   }
 
+  @Nullable
+  public Description getCurrentTest() {
+    return currentTest.get();
+  }
+
+  @Override
+  public void testStarted(Description description) {
+    currentTest.set(description);
+  }
+
+  @Override
+  public void testFinished(Description description) {
+    currentTest.remove();
+  }
+
   public RunnerScheduler parallelScheduler() {
     // Schedulers can't be shared, create a new one with the shared thread pool
     return new ParallelScheduler(exec);