PYTHON-1884 Implement auto encryption spec tests

Skip test for symbol type which pymongo converts to string.
Fix {} comparison with RawBSONDocument in command events.
Add support for $$type assertions.
Nicer message in check_events.
Support errorContains with empty string.
Move custom data files to custom/.

Author: ShaneHarvey

.gitignore

@@ -13,3 +13,4 @@ pymongo.egg-info/
 *.so
 *.egg
 .tox
+mongocryptd.pid

pymongo/encryption.py

@@ -30,7 +30,8 @@
                            _inflate_bson)
 from bson.son import SON
 
-from pymongo.errors import ServerSelectionTimeoutError
+from pymongo.errors import (EncryptionError,
+                            ServerSelectionTimeoutError)
 from pymongo.mongo_client import MongoClient
 from pymongo.pool import _configured_socket, PoolOptions
 from pymongo.ssl_support import get_ssl_context
@@ -210,8 +211,11 @@ def encrypt(self, database, cmd, check_keys, codec_options):
         """
         # Workaround for $clusterTime which is incompatible with check_keys.
         cluster_time = check_keys and cmd.pop('$clusterTime', None)
-        encrypted_cmd = self._auto_encrypter.encrypt(
-            database, _dict_to_bson(cmd, check_keys, codec_options))
+        encoded_cmd = _dict_to_bson(cmd, check_keys, codec_options)
+        try:
+            encrypted_cmd = self._auto_encrypter.encrypt(database, encoded_cmd)
+        except MongoCryptError as exc:
+            raise EncryptionError(exc)
         # TODO: PYTHON-1922 avoid decoding the encrypted_cmd.
         encrypt_cmd = _inflate_bson(encrypted_cmd, DEFAULT_RAW_BSON_OPTIONS)
         if cluster_time:
@@ -227,7 +231,10 @@ def decrypt(self, response):
         :Returns:
           The decrypted command response.
         """
-        return self._auto_encrypter.decrypt(response)
+        try:
+            return self._auto_encrypter.decrypt(response)
+        except MongoCryptError as exc:
+            raise EncryptionError(exc)
 
     def close(self):
         """Cleanup resources."""

pymongo/errors.py

@@ -247,3 +247,10 @@ class DocumentTooLarge(InvalidDocument):
     """Raised when an encoded document is too large for the connected server.
     """
     pass
+
+
+class EncryptionError(OperationFailure):
+    """Raised when encryption or decryption fails.
+
+    .. versionadded:: 3.9
+    """

test/client-side-encryption/key-document-local.json renamed to test/client-side-encryption/custom/key-document-local.json

@@ -33,7 +33,8 @@
 from pymongo.write_concern import WriteConcern
 
 from test import unittest, IntegrationTest, PyMongoTestCase, client_context
-from test.utils import wait_until
+from test.utils import TestCreator, camel_to_snake_args, wait_until
+from test.utils_spec_runner import SpecRunner
 
 
 if _HAVE_PYMONGOCRYPT:
@@ -129,8 +130,10 @@ def setUpClass(cls):
 
 
 # Location of JSON test files.
-TEST_PATH = os.path.join(
+BASE = os.path.join(
     os.path.dirname(os.path.realpath(__file__)), 'client-side-encryption')
+CUSTOM_PATH = os.path.join(BASE, 'custom')
+SPEC_PATH = os.path.join(BASE, 'spec')
 
 OPTS = CodecOptions(uuid_representation=STANDARD)
 
@@ -139,7 +142,7 @@ def setUpClass(cls):
 
 
 def read(filename):
-    with open(os.path.join(TEST_PATH, filename)) as fp:
+    with open(os.path.join(CUSTOM_PATH, filename)) as fp:
         return fp.read()
 
 
@@ -231,5 +234,104 @@ def test_auto_encrypt_local_schema_map(self):
         self._test_auto_encrypt(opts)
 
 
+# Spec tests
+
+AWS_CREDS = {
+    'accessKeyId': os.environ.get('FLE_AWS_KEY', ''),
+    'secretAccessKey': os.environ.get('FLE_AWS_SECRET', '')
+}
+
+
+class TestSpec(SpecRunner):
+
+    @classmethod
+    @unittest.skipUnless(_HAVE_PYMONGOCRYPT, 'pymongocrypt is not installed')
+    def setUpClass(cls):
+        super(TestSpec, cls).setUpClass()
+
+    def parse_auto_encrypt_opts(self, opts):
+        """Parse clientOptions.autoEncryptOpts."""
+        opts = camel_to_snake_args(opts)
+        kms_providers = opts['kms_providers']
+        if 'aws' in kms_providers:
+            kms_providers['aws'] = AWS_CREDS
+            if not any(AWS_CREDS.values()):
+                self.skipTest('AWS environment credentials are not set')
+        if 'key_vault_namespace' not in opts:
+            opts['key_vault_namespace'] = 'admin.datakeys'
+        opts = dict(opts)
+        return AutoEncryptionOpts(**opts)
+
+    def parse_client_options(self, opts):
+        """Override clientOptions parsing to support autoEncryptOpts."""
+        encrypt_opts = opts.pop('autoEncryptOpts')
+        if encrypt_opts:
+            opts['auto_encryption_opts'] = self.parse_auto_encrypt_opts(
+                encrypt_opts)
+
+        return super(TestSpec, self).parse_client_options(opts)
+
+    def get_object_name(self, op):
+        """Default object is collection."""
+        return op.get('object', 'collection')
+
+    def maybe_skip_scenario(self, test):
+        super(TestSpec, self).maybe_skip_scenario(test)
+        if 'type=symbol' in test['description'].lower():
+            raise unittest.SkipTest(
+                'PyMongo does not support the symbol type')
+
+    def setup_scenario(self, scenario_def):
+        """Override a test's setup."""
+        key_vault_data = scenario_def['key_vault_data']
+        if key_vault_data:
+            coll = client_context.client.get_database(
+                'admin',
+                write_concern=WriteConcern(w='majority'),
+                codec_options=OPTS)['datakeys']
+            coll.drop()
+            coll.insert_many(key_vault_data)
+
+        db_name = self.get_scenario_db_name(scenario_def)
+        coll_name = self.get_scenario_coll_name(scenario_def)
+        db = client_context.client.get_database(
+            db_name, write_concern=WriteConcern(w='majority'),
+            codec_options=OPTS)
+        coll = db[coll_name]
+        coll.drop()
+        json_schema = scenario_def['json_schema']
+        if json_schema:
+            db.create_collection(
+                coll_name,
+                validator={'$jsonSchema': json_schema}, codec_options=OPTS)
+        else:
+            db.create_collection(coll_name)
+
+        if scenario_def['data']:
+            # Load data.
+            coll.insert_many(scenario_def['data'])
+
+    def allowable_errors(self, op):
+        """Override expected error classes."""
+        errors = super(TestSpec, self).allowable_errors(op)
+        # An updateOne test expects encryption to error when no $ operator
+        # appears but pymongo raises a client side ValueError in this case.
+        if op['name'] == 'updateOne':
+            errors += (ValueError,)
+        return errors
+
+
+def create_test(scenario_def, test, name):
+    @client_context.require_test_commands
+    def run_scenario(self):
+        self.run_scenario(scenario_def, test)
+
+    return run_scenario
+
+
+test_creator = TestCreator(create_test, TestSpec, SPEC_PATH)
+test_creator.create_tests()
+
+
 if __name__ == "__main__":
     unittest.main()

test/client-side-encryption/schema.json renamed to test/client-side-encryption/custom/schema.json

@@ -20,6 +20,7 @@
 sys.path[0:0] = [""]
 
 from pymongo.mongo_client import MongoClient
+from pymongo.write_concern import WriteConcern
 
 from test import unittest, client_context, PyMongoTestCase
 from test.utils import TestCreator
@@ -67,6 +68,28 @@ def maybe_skip_scenario(self, test):
                 raise unittest.SkipTest(
                     'PyMongo does not support %s' % (name,))
 
+    def get_scenario_coll_name(self, scenario_def):
+        """Override a test's collection name to support GridFS tests."""
+        if 'bucket_name' in scenario_def:
+            return scenario_def['bucket_name']
+        return super(TestSpec, self).get_scenario_coll_name(scenario_def)
+
+    def setup_scenario(self, scenario_def):
+        """Override a test's setup to support GridFS tests."""
+        if 'bucket_name' in scenario_def:
+            db_name = self.get_scenario_db_name(scenario_def)
+            db = client_context.client.get_database(
+                db_name, write_concern=WriteConcern(w='majority'))
+            # Create a bucket for the retryable reads GridFS tests.
+            client_context.client.drop_database(db_name)
+            if scenario_def['data']:
+                data = scenario_def['data']
+                # Load data.
+                db['fs.chunks'].insert_many(data['fs.chunks'])
+                db['fs.files'].insert_many(data['fs.files'])
+        else:
+            super(TestSpec, self).setup_scenario(scenario_def)
+
 
 def create_test(scenario_def, test, name):
     @client_context.require_test_commands

test/test_encryption.py

@@ -321,8 +321,12 @@ def create_tests(self):
 
             for filename in filenames:
                 with open(os.path.join(dirpath, filename)) as scenario_stream:
+                    # Use tz_aware=False to match how CodecOptions decodes
+                    # dates.
+                    opts = json_util.JSONOptions(tz_aware=False)
                     scenario_def = ScenarioDict(
-                        json_util.loads(scenario_stream.read()))
+                        json_util.loads(scenario_stream.read(),
+                                        json_options=opts))
 
                 test_type = os.path.splitext(filename)[0]