PYTHON-4928 Convert CSFLE spec tests to unified test format

.evergreen/remove-unimplemented-tests.sh

@@ -6,6 +6,7 @@ rm $PYMONGO/test/connection_monitoring/wait-queue-fairness.json  # PYTHON-1873
 rm $PYMONGO/test/discovery_and_monitoring/unified/pool-clear-application-error.json  # PYTHON-4918
 rm $PYMONGO/test/discovery_and_monitoring/unified/pool-clear-checkout-error.json  # PYTHON-4918
 rm $PYMONGO/test/discovery_and_monitoring/unified/pool-clear-min-pool-size-error.json  # PYTHON-4918
+rm $PYMONGO/test/client-side-encryption/spec/unified/client-bulkWrite-qe.json  # PYTHON-4929
 
 # Python doesn't implement DRIVERS-3064
 rm $PYMONGO/test/collection_management/listCollections-rawdata.json

test/asynchronous/test_encryption.py

@@ -747,8 +747,7 @@ async def run_scenario(self):
 if _HAVE_PYMONGOCRYPT:
     globals().update(
         generate_test_classes(
-            os.path.join(SPEC_PATH, "unified"),
-            module=__name__,
+            os.path.join(SPEC_PATH, "unified"), module=__name__, expected_failures=["mapReduce .*"]
         )
     )
 

test/asynchronous/unified_format.py

@@ -255,6 +255,10 @@ def _handle_placeholders(self, spec: dict, current: dict, path: str) -> Any:
                 raise ValueError(f"Could not find a placeholder value for {path}")
             return PLACEHOLDER_MAP[path]
 
+        # Distinguish between temp and non-temp aws credentials.
+        if path.endswith("/kmsProviders/aws") and "sessionToken" in current:
+            path = path.replace("aws", "aws_temp")
+
         for key in list(current):
             value = current[key]
             if isinstance(value, dict):
@@ -275,10 +279,8 @@ async def _create_entity(self, entity_spec, uri=None):
             if "autoEncryptOpts" in spec:
                 auto_encrypt_opts = spec["autoEncryptOpts"].copy()
                 auto_encrypt_kwargs: dict = dict(kms_tls_options=DEFAULT_KMS_TLS)
-                kms_providers = ALL_KMS_PROVIDERS.copy()
+                kms_providers = auto_encrypt_opts.pop("kmsProviders", ALL_KMS_PROVIDERS.copy())
                 key_vault_namespace = auto_encrypt_opts.pop("keyVaultNamespace")
-                for provider_name, provider_value in auto_encrypt_opts.pop("kmsProviders").items():
-                    kms_providers[provider_name].update(provider_value)
                 extra_opts = auto_encrypt_opts.pop("extraOptions", {})
                 for key, value in extra_opts.items():
                     auto_encrypt_kwargs[camel_to_snake(key)] = value
@@ -552,22 +554,25 @@ async def asyncSetUp(self):
 
     def maybe_skip_test(self, spec):
         # add any special-casing for skipping tests here
-        if "Client side error in command starting transaction" in spec["description"]:
+        class_name = self.__class__.__name__.lower()
+        description = spec["description"].lower()
+
+        if "client side error in command starting transaction" in description:
             self.skipTest("Implement PYTHON-1894")
-        if "timeoutMS applied to entire download" in spec["description"]:
+        if "type=symbol" in description:
+            self.skipTest("PyMongo does not support the symbol type")
+        if "timeoutms applied to entire download" in description:
             self.skipTest("PyMongo's open_download_stream does not cap the stream's lifetime")
         if any(
-            x in spec["description"]
+            x in description
             for x in [
-                "First insertOne is never committed",
-                "Second updateOne is never committed",
-                "Third updateOne is never committed",
+                "first insertone is never committed",
+                "second updateone is never committed",
+                "third updateone is never committed",
             ]
         ):
             self.skipTest("Implement PYTHON-4597")
 
-        class_name = self.__class__.__name__.lower()
-        description = spec["description"].lower()
         if "csot" in class_name:
             # Skip tests that are too slow to run on a given platform.
             slow_macos = [
@@ -785,6 +790,38 @@ async def _databaseOperation_createCommandCursor(self, target, **kwargs):
 
         return cursor
 
+    async def _collectionOperation_assertIndexExists(self, target, **kwargs):
+        collection = self.client[kwargs["database_name"]][kwargs["collection_name"]]
+        index_names = [idx["name"] async for idx in await collection.list_indexes()]
+        self.assertIn(kwargs["index_name"], index_names)
+
+    async def _collectionOperation_assertIndexNotExists(self, target, **kwargs):
+        collection = self.client[kwargs["database_name"]][kwargs["collection_name"]]
+        async for index in await collection.list_indexes():
+            self.assertNotEqual(kwargs["indexName"], index["name"])
+
+    async def _collectionOperation_assertCollectionExists(self, target, **kwargs):
+        database_name = kwargs["database_name"]
+        collection_name = kwargs["collection_name"]
+        collection_name_list = await self.client.get_database(database_name).list_collection_names()
+        self.assertIn(collection_name, collection_name_list)
+
+    async def _databaseOperation_assertIndexExists(self, target, **kwargs):
+        collection = self.client[kwargs["database_name"]][kwargs["collection_name"]]
+        index_names = [idx["name"] async for idx in await collection.list_indexes()]
+        self.assertIn(kwargs["index_name"], index_names)
+
+    async def _databaseOperation_assertIndexNotExists(self, target, **kwargs):
+        collection = self.client[kwargs["database_name"]][kwargs["collection_name"]]
+        async for index in await collection.list_indexes():
+            self.assertNotEqual(kwargs["indexName"], index["name"])
+
+    async def _databaseOperation_assertCollectionExists(self, target, **kwargs):
+        database_name = kwargs["database_name"]
+        collection_name = kwargs["collection_name"]
+        collection_name_list = await self.client.get_database(database_name).list_collection_names()
+        self.assertIn(collection_name, collection_name_list)
+
     async def kill_all_sessions(self):
         if getattr(self, "client", None) is None:
             return

test/client-side-encryption/etc/data/encryptedFields-prefix-suffix.json

@@ -0,0 +1,38 @@
+{
+  "fields": [
+    {
+            "keyId": {
+                "$binary": {
+                    "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                    "subType": "04"
+                }
+            },
+            "path": "encryptedText",
+            "bsonType": "string",
+            "queries": [
+                {
+                    "queryType": "prefixPreview",
+                    "strMinQueryLength": {
+                        "$numberInt": "2"
+                    },
+                    "strMaxQueryLength": {
+                        "$numberInt": "10"
+                    },
+                    "caseSensitive": true,
+                    "diacriticSensitive": true
+                },
+                {
+                    "queryType": "suffixPreview",
+                    "strMinQueryLength": {
+                        "$numberInt": "2"
+                    },
+                    "strMaxQueryLength": {
+                        "$numberInt": "10"
+                    },
+                    "caseSensitive": true,
+                    "diacriticSensitive": true
+                }
+            ]
+        }
+  ]
+}

test/client-side-encryption/etc/data/encryptedFields-substring.json

@@ -0,0 +1,30 @@
+{
+  "fields": [
+    {
+            "keyId": {
+                "$binary": {
+                    "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                    "subType": "04"
+                }
+            },
+            "path": "encryptedText",
+            "bsonType": "string",
+            "queries": [
+                {
+                    "queryType": "substringPreview",
+                    "strMaxLength": {
+                        "$numberInt": "10"
+                    },
+                    "strMinQueryLength": {
+                        "$numberInt": "2"
+                    },
+                    "strMaxQueryLength": {
+                        "$numberInt": "10"
+                    },
+                    "caseSensitive": true,
+                    "diacriticSensitive": true
+                }
+            ]
+        }
+  ]
+}

test/client-side-encryption/limits/limits-encryptedFields.json

@@ -0,0 +1,14 @@
+{
+    "fields": [
+        {
+            "keyId": {
+                "$binary": {
+                    "base64": "LOCALAAAAAAAAAAAAAAAAA==",
+                    "subType": "04"
+                }
+            },
+            "path": "foo",
+            "bsonType": "string"
+        }
+    ]
+}

test/client-side-encryption/limits/limits-qe-doc.json

@@ -0,0 +1,3 @@
+{
+    "foo": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+}