RUST-2077 Convert all CSFLE tests to unified format (#1491)

Author: isabelatkinson

src/client/csfle/options.rs

@@ -141,67 +141,6 @@ impl KmsProviders {
     pub(crate) fn credentials(&self) -> &HashMap<KmsProvider, Document> {
         &self.credentials
     }
-
-    #[cfg(test)]
-    pub(crate) fn set_test_options(&mut self) {
-        use mongocrypt::ctx::KmsProviderType;
-
-        use crate::{
-            bson::doc,
-            test::csfle::{ALL_KMS_PROVIDERS, AWS_KMS},
-        };
-
-        let all_kms_providers = ALL_KMS_PROVIDERS.clone();
-        for (provider, test_credentials, tls_options) in all_kms_providers {
-            if self.credentials.contains_key(&provider)
-                && !matches!(provider.provider_type(), KmsProviderType::Local)
-            {
-                self.set(provider, test_credentials, tls_options);
-            }
-        }
-
-        let aws_temp_provider = KmsProvider::other("awsTemporary".to_string());
-        if self.credentials.contains_key(&aws_temp_provider) {
-            let aws_credentials = doc! {
-                "accessKeyId": std::env::var("CSFLE_AWS_TEMP_ACCESS_KEY_ID").unwrap(),
-                "secretAccessKey": std::env::var("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY").unwrap(),
-                "sessionToken": std::env::var("CSFLE_AWS_TEMP_SESSION_TOKEN").unwrap()
-            };
-            self.set(KmsProvider::aws(), aws_credentials, AWS_KMS.clone().2);
-            self.clear(&aws_temp_provider);
-        }
-
-        let aws_temp_no_session_token_provider = KmsProvider::other("awsTemporaryNoSessionToken");
-        if self
-            .credentials
-            .contains_key(&aws_temp_no_session_token_provider)
-        {
-            let aws_credentials = doc! {
-                "accessKeyId": std::env::var("CSFLE_AWS_TEMP_ACCESS_KEY_ID").unwrap(),
-                "secretAccessKey": std::env::var("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY").unwrap(),
-            };
-            self.set(KmsProvider::aws(), aws_credentials, AWS_KMS.clone().2);
-            self.clear(&aws_temp_no_session_token_provider);
-        }
-    }
-
-    #[cfg(test)]
-    pub(crate) fn set(&mut self, provider: KmsProvider, creds: Document, tls: Option<TlsOptions>) {
-        self.credentials.insert(provider.clone(), creds);
-        if let Some(tls) = tls {
-            self.tls_options
-                .get_or_insert_with(KmsProvidersTlsOptions::new)
-                .insert(provider, tls);
-        }
-    }
-
-    #[cfg(test)]
-    pub(crate) fn clear(&mut self, provider: &KmsProvider) {
-        self.credentials.remove(provider);
-        if let Some(tls_opts) = &mut self.tls_options {
-            tls_opts.remove(provider);
-        }
-    }
 }
 
 impl AutoEncryptionOptions {

src/client/options.rs

@@ -1282,54 +1282,6 @@ impl ClientOptions {
         Ok(())
     }
 
-    /// Applies the options in other to these options if a value is not already present
-    #[cfg(test)]
-    pub(crate) fn merge(&mut self, other: ClientOptions) {
-        if self.hosts.is_empty() {
-            self.hosts = other.hosts;
-        }
-
-        #[cfg(any(
-            feature = "zstd-compression",
-            feature = "zlib-compression",
-            feature = "snappy-compression"
-        ))]
-        merge_options!(other, self, [compressors]);
-
-        merge_options!(
-            other,
-            self,
-            [
-                app_name,
-                cmap_event_handler,
-                command_event_handler,
-                connect_timeout,
-                credential,
-                direct_connection,
-                driver_info,
-                heartbeat_freq,
-                load_balanced,
-                local_threshold,
-                max_idle_time,
-                max_pool_size,
-                min_pool_size,
-                read_concern,
-                repl_set_name,
-                retry_reads,
-                retry_writes,
-                selection_criteria,
-                server_api,
-                server_selection_timeout,
-                socket_timeout,
-                test_options,
-                tls,
-                write_concern,
-                original_srv_info,
-                original_uri
-            ]
-        );
-    }
-
     #[cfg(test)]
     pub(crate) fn test_options_mut(&mut self) -> &mut TestOptions {
         self.test_options.get_or_insert_with(Default::default)

src/cmap/test/file.rs

@@ -8,25 +8,80 @@ use crate::{
     error::Result,
     event::cmap::CmapEvent,
     serde_util,
-    test::{util::fail_point::FailPoint, RunOn},
+    test::{
+        get_topology,
+        log_uncaptured,
+        server_version_matches,
+        util::fail_point::FailPoint,
+        Serverless,
+        Topology,
+    },
 };
 
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct TestFile {
+pub(super) struct TestFile {
     #[serde(rename = "version")]
     _version: u8, // can ignore this field as there's only one version
     #[serde(rename = "style")]
     _style: TestStyle, // we use the presence of fail_point / run_on to determine this
-    pub description: String,
-    pub(crate) pool_options: Option<ConnectionPoolOptions>,
-    pub operations: Vec<ThreadedOperation>,
-    pub error: Option<Error>,
-    pub(crate) events: Vec<CmapEvent>,
+    pub(super) description: String,
+    pub(super) pool_options: Option<ConnectionPoolOptions>,
+    pub(super) operations: Vec<ThreadedOperation>,
+    pub(super) error: Option<Error>,
+    pub(super) events: Vec<CmapEvent>,
     #[serde(default)]
-    pub ignore: Vec<String>,
-    pub fail_point: Option<FailPoint>,
-    pub(crate) run_on: Option<Vec<RunOn>>,
+    pub(super) ignore: Vec<String>,
+    pub(super) fail_point: Option<FailPoint>,
+    pub(super) run_on: Option<Vec<RunOn>>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(super) struct RunOn {
+    pub(super) min_server_version: Option<String>,
+    pub(super) max_server_version: Option<String>,
+    pub(super) topology: Option<Vec<Topology>>,
+    pub(super) serverless: Option<Serverless>,
+}
+
+impl RunOn {
+    pub(super) async fn can_run_on(&self) -> bool {
+        if let Some(ref min_version) = self.min_server_version {
+            if !server_version_matches(&format!(">= {min_version}")).await {
+                log_uncaptured(format!(
+                    "runOn mismatch: required server version >= {min_version}",
+                ));
+                return false;
+            }
+        }
+        if let Some(ref max_version) = self.max_server_version {
+            if !server_version_matches(&format!("<= {max_version}")).await {
+                log_uncaptured(format!(
+                    "runOn mismatch: required server version <= {max_version}",
+                ));
+                return false;
+            }
+        }
+        if let Some(ref topology) = self.topology {
+            let actual_topology = get_topology().await;
+            if !topology.contains(actual_topology) {
+                log_uncaptured(format!(
+                    "runOn mismatch: required topology in {topology:?}, got {actual_topology:?}"
+                ));
+                return false;
+            }
+        }
+        if let Some(ref serverless) = self.serverless {
+            if !serverless.can_run() {
+                log_uncaptured(format!(
+                    "runOn mismatch: required serverless {serverless:?}"
+                ));
+                return false;
+            }
+        }
+        true
+    }
 }
 
 #[derive(Debug, Deserialize)]

src/db/options.rs

@@ -84,10 +84,10 @@ pub struct CreateCollectionOptions {
     /// determine the contents of the view.
     pub pipeline: Option<Vec<Document>>,
 
-    /// The default collation for the collection or view.   
+    /// The default collation for the collection or view.
     pub collation: Option<Collation>,
 
-    /// The write concern for the operation.   
+    /// The write concern for the operation.
     #[serde(skip_serializing_if = "write_concern_is_empty")]
     pub write_concern: Option<WriteConcern>,
 
@@ -155,6 +155,7 @@ pub enum ValidationAction {
 
 /// Specifies options for a clustered collection.  Some fields have required values; the `Default`
 /// impl uses those values.
+#[skip_serializing_none]
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
@@ -169,7 +170,6 @@ pub struct ClusteredIndex {
     pub name: Option<String>,
 
     /// Optional; currently must be `2` if provided.
-    #[serde(skip_serializing_if = "Option::is_none")]
     pub v: Option<i32>,
 }
 

src/test.rs

@@ -33,7 +33,7 @@ pub(crate) mod util;
 #[cfg(feature = "in-use-encryption")]
 pub(crate) use self::csfle_skip_local as csfle;
 pub(crate) use self::{
-    spec::{run_spec_test, RunOn, Serverless, Topology},
+    spec::{run_spec_test, Serverless, Topology},
     util::{
         assert_matches,
         eq_matches,
@@ -205,7 +205,7 @@ pub(crate) async fn get_max_message_size_bytes() -> usize {
         .unwrap()
 }
 
-async fn get_topology() -> &'static Topology {
+pub(crate) async fn get_topology() -> &'static Topology {
     static TOPOLOGY: OnceCell<Topology> = OnceCell::const_new();
     TOPOLOGY
         .get_or_init(|| async {

src/test/csfle/spec.rs

@@ -1,36 +1,21 @@
-use crate::test::spec::{unified_runner::run_unified_tests, v2_runner::run_v2_tests};
+use crate::test::spec::unified_runner::run_unified_tests;
 
 #[tokio::test(flavor = "multi_thread")]
 async fn run_unified() {
     let mut skipped_tests = vec![];
     if cfg!(not(feature = "openssl-tls")) {
-        skipped_tests.push("create datakey with KMIP KMS provider");
-        skipped_tests.push("create datakey with KMIP delegated KMS provider");
-        skipped_tests.push("create datakey with named KMIP KMS provider");
+        skipped_tests.extend_from_slice(&[
+            "Insert a document with auto encryption using KMIP KMS provider",
+            "create datakey with KMIP KMS provider",
+            "create datakey with KMIP delegated KMS provider",
+            "create datakey with named KMIP KMS provider",
+            "Insert a document with auto encryption using KMIP delegated KMS provider",
+        ]);
     }
 
     run_unified_tests(&["client-side-encryption", "unified"])
         .skip_tests(&skipped_tests)
-        .await;
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn run_legacy() {
-    let mut skipped_files = vec![
-        // TODO RUST-528: unskip this file
-        "timeoutMS.json",
-        // These files have been migrated to unified tests.
-        // TODO DRIVERS-3178 remove these once the files are gone.
-        "fle2v2-BypassQueryAnalysis.json",
-        "fle2v2-EncryptedFields-vs-EncryptedFieldsMap.json",
-        "localSchema.json",
-        "maxWireVersion.json",
-    ];
-    if cfg!(not(feature = "openssl-tls")) {
-        skipped_files.push("kmipKMS.json");
-    }
-
-    run_v2_tests(&["client-side-encryption", "legacy"])
-        .skip_files(&skipped_files)
+        // TODO RUST-582: unskip this test
+        .skip_files(&["timeoutMS.json"])
         .await;
 }

src/test/spec.rs

@@ -22,7 +22,6 @@ mod sessions;
 mod trace;
 mod transactions;
 pub(crate) mod unified_runner;
-pub(crate) mod v2_runner;
 mod versioned_api;
 mod write_error;
 
@@ -38,8 +37,7 @@ use serde::{de::DeserializeOwned, Deserialize};
 
 pub(crate) use self::{
     oidc_skip_ci as oidc,
-    unified_runner::{merge_uri_options, ExpectedEventType, Topology},
-    v2_runner::{operation::Operation, test_file::RunOn},
+    unified_runner::{ExpectedEventType, Topology},
 };
 use crate::bson::Bson;
 

src/test/spec/json/client-side-encryption/README.md

@@ -619,7 +619,8 @@ Using `client_encrypted` perform the following operations:
     `{ "_id": "encryption_exceeds_16mib", "unencrypted": < the string "a" repeated (16777216 - 2000) times > }` into
     `coll`.
 
-    Expect this to fail since encryption results in a document exceeding the `maxBsonObjectSize` limit.
+    Expect this to fail indicating the document exceeded the `maxBsonObjectSize` limit. If the write is sent to the
+    server (i.e. does not fail due to a driver-side check), expect a server error with code 2 or 10334.
 
 7. If using MongoDB 8.0+, use MongoClient.bulkWrite to insert the following into `coll2`:
 
@@ -3787,6 +3788,57 @@ class AutoEncryptionOpts {
 
 Assert that an error is thrown.
 
+#### Case 4: ClientEncryption with `credentialProviders` and valid environment variables.
+
+Ensure a valid `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` are present in the environment.
+
+Create a MongoClient named `setupClient`.
+
+Create a [ClientEncryption](../client-side-encryption.md#clientencryption) object with the following options:
+
+```typescript
+class ClientEncryptionOpts {
+  keyVaultClient: <setupClient>,
+  keyVaultNamespace: "keyvault.datakeys",
+  kmsProviders: { "aws": {} },
+  credentialProviders: { "aws": <object/function that returns valid credentials from the secrets manager> }
+}
+```
+
+Use the client encryption to create a datakey using the "aws" KMS provider. This should successfully load and use the
+AWS credentials that were provided by the secrets manager for the remote provider. Assert the datakey was created and
+that the custom credential provider was called at least once.
+
+An example of this in Node.js:
+
+```typescript
+import { ClientEncryption, MongoClient } from 'mongodb';
+
+let calledCount = 0;
+const masterKey = {
+  region: '<aws region>',
+  key: '<key for arn>'
+};
+const keyVaultClient = new MongoClient(process.env.MONGODB_URI);
+const options = {
+  keyVaultNamespace: 'keyvault.datakeys',
+  kmsProviders: { aws: {} },
+  credentialProviders: {
+    aws: async () => {
+      calledCount++;
+      return {
+        accessKeyId: process.env.FLE_AWS_KEY,
+        secretAccessKey: process.env.FLE_AWS_SECRET
+      };
+    }
+  }
+};
+const clientEncryption = new ClientEncryption(keyVaultClient, options);
+const dk = await clientEncryption.createDataKey('aws', { masterKey });
+expect(dk).to.be.a(Binary);
+expect(calledCount).to.be.greaterThan(0);
+```
+
 ### 27. Text Explicit Encryption
 
 The Text Explicit Encryption tests utilize Queryable Encryption (QE) range protocol V2 and require MongoDB server 8.2.0+