CSFLE auto encryption tests improvements

driver-core/src/test/functional/com/mongodb/ClusterFixture.java

@@ -88,6 +88,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static com.mongodb.assertions.Assertions.assertNotNull;
 import static com.mongodb.connection.ClusterConnectionMode.LOAD_BALANCED;
@@ -278,6 +279,11 @@ public static String getEnv(final String name, final String defaultValue) {
         return value == null ? defaultValue : value;
     }
 
+    public static Optional<String> cryptSharedLibPathSysPropValue() {
+        String value = getEnv("CRYPT_SHARED_LIB_PATH", "");
+        return value.isEmpty() ? Optional.empty() : Optional.of(value);
+    }
+
     @Nullable
     public static String getEnv(final String name) {
         return System.getenv(name);

driver-core/src/test/functional/com/mongodb/client/test/CollectionHelper.java

@@ -176,6 +176,14 @@ public void create(final WriteConcern writeConcern, final BsonDocument createOpt
                 case "size":
                     createCollectionOptions.sizeInBytes(createOptions.getNumber("size").longValue());
                     break;
+                case "encryptedFields":
+                    createCollectionOptions.encryptedFields(createOptions.getDocument("encryptedFields"));
+                    break;
+                case "validator":
+                    ValidationOptions validationOptions = new ValidationOptions();
+                    validationOptions.validator(createOptions.getDocument("validator"));
+                    createCollectionOptions.validationOptions(validationOptions);
+                    break;
                 default:
                     throw new UnsupportedOperationException("Unsupported create collection option: " + option);
             }
@@ -195,6 +203,10 @@ public void create(final String collectionName, final CreateCollectionOptions op
         if (indexOptionDefaults.getStorageEngine() != null) {
             operation.indexOptionDefaults(new BsonDocument("storageEngine", toBsonDocument(indexOptionDefaults.getStorageEngine())));
         }
+        Bson encryptedFields = options.getEncryptedFields();
+        if (encryptedFields != null) {
+            operation.encryptedFields(encryptedFields.toBsonDocument());
+        }
         ValidationOptions validationOptions = options.getValidationOptions();
         if (validationOptions.getValidator() != null) {
             operation.validator(toBsonDocument(validationOptions.getValidator()));

driver-core/src/test/unit/com/mongodb/internal/connection/TestClusterListener.java

@@ -32,7 +32,6 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
 
-import static com.mongodb.assertions.Assertions.isTrue;
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.internal.Locks.withLock;
 
@@ -48,15 +47,17 @@ public final class TestClusterListener implements ClusterListener {
 
     @Override
     public void clusterOpening(final ClusterOpeningEvent event) {
-        isTrue("clusterOpeningEvent is null", clusterOpeningEvent == null);
-        clusterOpeningEvent = event;
+        if (clusterOpeningEvent == null) {
+            clusterOpeningEvent = event;
+        }
     }
 
     @Override
     public void clusterClosed(final ClusterClosedEvent event) {
-        isTrue("clusterClosingEvent is null", clusterClosingEvent == null);
-        closedLatch.countDown();
-        clusterClosingEvent = event;
+        if (clusterClosingEvent == null) {
+            closedLatch.countDown();
+            clusterClosingEvent = event;
+        }
     }
 
     @Override

driver-scala/src/integrationTest/scala/org/mongodb/scala/syncadapter/SyncClientEncryption.scala

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 org.mongodb.scala.syncadapter
+
+import com.mongodb.ClusterFixture.TIMEOUT_DURATION
+import com.mongodb.client.model.{ CreateCollectionOptions, CreateEncryptedCollectionParams }
+import com.mongodb.client.model.vault.{
+  DataKeyOptions,
+  EncryptOptions,
+  RewrapManyDataKeyOptions,
+  RewrapManyDataKeyResult
+}
+import com.mongodb.client.result.DeleteResult
+import com.mongodb.client.vault.{ ClientEncryption => JClientEncryption }
+import com.mongodb.client.{ MongoDatabase => JMongoDatabase }
+import org.bson.{ BsonBinary, BsonDocument, BsonValue }
+import org.bson.conversions.Bson
+import org.mongodb.scala.vault.ClientEncryption
+import reactor.core.publisher.Mono
+
+import java.util.Objects.requireNonNull
+
+case class SyncClientEncryption(wrapped: ClientEncryption) extends JClientEncryption {
+
+  override def createDataKey(kmsProvider: String): BsonBinary =
+    requireNonNull(Mono.from(wrapped.createDataKey(kmsProvider, new DataKeyOptions)).block(TIMEOUT_DURATION))
+
+  override def createDataKey(kmsProvider: String, dataKeyOptions: DataKeyOptions): BsonBinary =
+    requireNonNull(Mono.from(wrapped.createDataKey(kmsProvider, dataKeyOptions)).block(TIMEOUT_DURATION))
+
+  override def encrypt(value: BsonValue, options: EncryptOptions): BsonBinary =
+    requireNonNull(Mono.from(wrapped.encrypt(value, options)).block(TIMEOUT_DURATION))
+
+  override def encryptExpression(expression: Bson, options: EncryptOptions): BsonDocument =
+    requireNonNull(Mono.from(wrapped
+      .encryptExpression(expression.toBsonDocument, options)).block(TIMEOUT_DURATION).toBsonDocument)
+
+  override def decrypt(value: BsonBinary): BsonValue =
+    requireNonNull(Mono.from(wrapped.decrypt(value)).block(TIMEOUT_DURATION))
+
+  override def deleteKey(id: BsonBinary): DeleteResult =
+    requireNonNull(Mono.from(wrapped.deleteKey(id)).block(TIMEOUT_DURATION))
+
+  override def getKey(id: BsonBinary): BsonDocument = Mono.from(wrapped.getKey(id)).block(TIMEOUT_DURATION)
+
+  override def getKeys = new SyncFindIterable[BsonDocument](wrapped.keys)
+
+  override def addKeyAltName(id: BsonBinary, keyAltName: String): BsonDocument =
+    Mono.from(wrapped.addKeyAltName(id, keyAltName)).block(TIMEOUT_DURATION)
+
+  override def removeKeyAltName(id: BsonBinary, keyAltName: String): BsonDocument =
+    Mono.from(wrapped.removeKeyAltName(id, keyAltName)).block(TIMEOUT_DURATION)
+
+  override def getKeyByAltName(keyAltName: String): BsonDocument =
+    Mono.from(wrapped.getKeyByAltName(keyAltName)).block(TIMEOUT_DURATION)
+
+  override def rewrapManyDataKey(filter: Bson): RewrapManyDataKeyResult =
+    requireNonNull(Mono.from(wrapped.rewrapManyDataKey(filter)).block(TIMEOUT_DURATION))
+
+  override def rewrapManyDataKey(filter: Bson, options: RewrapManyDataKeyOptions): RewrapManyDataKeyResult =
+    requireNonNull(Mono.from(wrapped.rewrapManyDataKey(filter, options)).block(TIMEOUT_DURATION))
+
+  override def createEncryptedCollection(
+      database: JMongoDatabase,
+      collectionName: String,
+      createCollectionOptions: CreateCollectionOptions,
+      createEncryptedCollectionParams: CreateEncryptedCollectionParams
+  ): BsonDocument = {
+    database match {
+      case syncMongoDatabase: SyncMongoDatabase =>
+        requireNonNull(Mono.from(wrapped.createEncryptedCollection(
+          syncMongoDatabase.wrapped,
+          collectionName,
+          createCollectionOptions,
+          createEncryptedCollectionParams
+        )).block(TIMEOUT_DURATION))
+      case _ => throw new AssertionError(s"Unexpected database type: ${database.getClass}")
+    }
+  }
+
+  override def close(): Unit = {
+    wrapped.close()
+  }
+}

driver-scala/src/integrationTest/scala/org/mongodb/scala/syncadapter/SyncMongoCollection.scala

@@ -33,7 +33,6 @@ import org.mongodb.scala.MongoCollection
 import org.mongodb.scala.bson.DefaultHelper.DefaultsTo
 import org.mongodb.scala.result.{ InsertManyResult, InsertOneResult }
 
-import java.util
 import java.util.concurrent.TimeUnit
 import scala.collection.JavaConverters._
 import scala.concurrent.duration.{ Duration, MILLISECONDS }
@@ -588,25 +587,25 @@ case class SyncMongoCollection[T](wrapped: MongoCollection[T]) extends JMongoCol
     )
 
   override def renameCollection(newCollectionNamespace: MongoNamespace): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.renameCollection(newCollectionNamespace).toFuture().get()
   }
 
   override def renameCollection(
       newCollectionNamespace: MongoNamespace,
       renameCollectionOptions: RenameCollectionOptions
   ): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.renameCollection(newCollectionNamespace, renameCollectionOptions).toFuture().get()
   }
 
   override def renameCollection(clientSession: ClientSession, newCollectionNamespace: MongoNamespace): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.renameCollection(unwrap(clientSession), newCollectionNamespace).toFuture().get()
   }
 
   override def renameCollection(
       clientSession: ClientSession,
       newCollectionNamespace: MongoNamespace,
       renameCollectionOptions: RenameCollectionOptions
   ): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.renameCollection(unwrap(clientSession), newCollectionNamespace, renameCollectionOptions).toFuture().get()
   }
 }

driver-scala/src/integrationTest/scala/org/mongodb/scala/syncadapter/SyncMongoDatabase.scala

@@ -51,13 +51,15 @@ case class SyncMongoDatabase(wrapped: MongoDatabase) extends JMongoDatabase {
   override def withCodecRegistry(codecRegistry: CodecRegistry) =
     SyncMongoDatabase(wrapped.withCodecRegistry(codecRegistry))
 
-  override def withReadPreference(readPreference: ReadPreference) = throw new UnsupportedOperationException
+  override def withReadPreference(readPreference: ReadPreference) =
+    SyncMongoDatabase(wrapped.withReadPreference(readPreference))
 
-  override def withWriteConcern(writeConcern: WriteConcern) = throw new UnsupportedOperationException
+  override def withWriteConcern(writeConcern: WriteConcern) = SyncMongoDatabase(wrapped.withWriteConcern(writeConcern))
 
-  override def withReadConcern(readConcern: ReadConcern) = throw new UnsupportedOperationException
+  override def withReadConcern(readConcern: ReadConcern) = SyncMongoDatabase(wrapped.withReadConcern(readConcern))
 
-  override def withTimeout(timeout: Long, timeUnit: TimeUnit) = throw new UnsupportedOperationException
+  override def withTimeout(timeout: Long, timeUnit: TimeUnit) =
+    SyncMongoDatabase(wrapped.withTimeout(timeout, timeUnit))
 
   override def getCollection(collectionName: String) =
     SyncMongoCollection[Document](wrapped.getCollection(collectionName))
@@ -170,7 +172,7 @@ case class SyncMongoDatabase(wrapped: MongoDatabase) extends JMongoDatabase {
   }
 
   override def createView(viewName: String, viewOn: String, pipeline: java.util.List[_ <: Bson]): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.createView(viewName, viewOn, pipeline.asScala.toList).toFuture().get()
   }
 
   override def createView(
@@ -179,7 +181,7 @@ case class SyncMongoDatabase(wrapped: MongoDatabase) extends JMongoDatabase {
       pipeline: java.util.List[_ <: Bson],
       createViewOptions: CreateViewOptions
   ): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.createView(viewName, viewOn, pipeline.asScala.toList, createViewOptions).toFuture().get()
   }
 
   override def createView(
@@ -188,7 +190,7 @@ case class SyncMongoDatabase(wrapped: MongoDatabase) extends JMongoDatabase {
       viewOn: String,
       pipeline: java.util.List[_ <: Bson]
   ): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.createView(unwrap(clientSession), viewName, viewOn, pipeline.asScala.toList).toFuture().get()
   }
 
   override def createView(
@@ -198,7 +200,13 @@ case class SyncMongoDatabase(wrapped: MongoDatabase) extends JMongoDatabase {
       pipeline: java.util.List[_ <: Bson],
       createViewOptions: CreateViewOptions
   ): Unit = {
-    throw new UnsupportedOperationException
+    wrapped.createView(
+      unwrap(clientSession),
+      viewName,
+      viewOn,
+      pipeline.asScala.toList,
+      createViewOptions
+    ).toFuture().get()
   }
 
   override def watch = new SyncChangeStreamIterable[Document](wrapped.watch[Document]())

driver-scala/src/integrationTest/scala/org/mongodb/scala/unified/ClientEncryptionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 org.mongodb.scala.unified
+
+object ClientEncryptionTest extends UnifiedTest {
+  val directory = "client-side-encryption/tests/unified"
+}

driver-scala/src/integrationTest/scala/org/mongodb/scala/unified/UnifiedCrudTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 org.mongodb.scala.unified
+
+object UnifiedCrudTest extends UnifiedTest {
+  val directory = "crud"
+}

driver-scala/src/integrationTest/scala/org/mongodb/scala/unified/UnifiedTest.scala

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 org.mongodb.scala.unified
+
+import com.mongodb.client.gridfs.{ GridFSBucket => JGridFSBucket }
+import com.mongodb.client.unified.UnifiedTest.Language
+import com.mongodb.client.unified.{ UnifiedTest, UnifiedTest => JUnifiedTest }
+import com.mongodb.client.vault.{ ClientEncryption => JClientEncryption }
+import com.mongodb.client.{ MongoClient => JMongoClient, MongoDatabase => JMongoDatabase }
+import com.mongodb.reactivestreams.client.internal.vault.ClientEncryptionImpl
+import com.mongodb.{ ClientEncryptionSettings => JClientEncryptionSettings, MongoClientSettings }
+import org.junit.jupiter.api.TestInstance
+import org.junit.jupiter.api.TestInstance.Lifecycle
+import org.junit.jupiter.params.provider.Arguments
+import org.mongodb.scala.MongoClient
+import org.mongodb.scala.MongoClient.DEFAULT_CODEC_REGISTRY
+import org.mongodb.scala.syncadapter.{ SyncClientEncryption, SyncMongoClient }
+import org.mongodb.scala.vault.ClientEncryption
+
+import java.util
+
+@TestInstance(Lifecycle.PER_CLASS)
+abstract class UnifiedTest extends JUnifiedTest {
+
+  val directory: String
+
+  def data(): util.Collection[Arguments] = JUnifiedTest.getTestData(directory, true, Language.SCALA)
+
+  override def createMongoClient(settings: MongoClientSettings): JMongoClient =
+    SyncMongoClient(MongoClient(MongoClientSettings.builder(settings).codecRegistry(DEFAULT_CODEC_REGISTRY).build()))
+
+  override def createGridFSBucket(database: JMongoDatabase): JGridFSBucket =
+    throw new NotImplementedError("Not implemented")
+
+  override def createClientEncryption(
+      keyVaultClient: JMongoClient,
+      clientEncryptionSettings: JClientEncryptionSettings
+  ): JClientEncryption = {
+    keyVaultClient match {
+      case client: SyncMongoClient =>
+        SyncClientEncryption(ClientEncryption(new ClientEncryptionImpl(
+          client.wrapped.wrapped,
+          clientEncryptionSettings
+        )))
+      case _ => throw new IllegalArgumentException(s"Invalid keyVaultClient type: ${keyVaultClient.getClass}")
+    }
+  }
+
+  override protected def isReactive: Boolean = true
+
+  override protected def getLanguage: Language = Language.SCALA
+}

driver-scala/src/main/scala/org/mongodb/scala/MongoClient.scala

@@ -111,7 +111,7 @@ object MongoClient {
  * @param wrapped the underlying java MongoClient
  * @since 1.0
  */
-case class MongoClient(private val wrapped: JMongoClient) extends MongoCluster(wrapped) with Closeable {
+case class MongoClient(protected[scala] val wrapped: JMongoClient) extends MongoCluster(wrapped) with Closeable {
 
   /**
    * Close the client, which will close all underlying cached resources, including, for example,