retrieve configured provider from conn settings

Author: aclark4life

tests/encryption_/test_base.py

@@ -1,5 +1,3 @@
-import os
-
 import pymongo
 from bson.binary import Binary
 from django.conf import settings
@@ -21,18 +19,3 @@ def assertEncrypted(self, model, field):
             collection = db[model._meta.db_table]
             data = collection.find_one({}, {field: 1, "_id": 0})
             self.assertIsInstance(data[field], Binary)
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        AWS_CREDS = {
-            "accessKeyId": os.environ.get("FLE_AWS_KEY", ""),
-            "secretAccessKey": os.environ.get("FLE_AWS_SECRET", ""),
-        }
-        _USE_AWS_KMS = any(AWS_CREDS.values())
-
-        if _USE_AWS_KMS:
-            self.DEFAULT_KMS_PROVIDER = "aws"
-        else:
-            # Local-only fallback
-            self.DEFAULT_KMS_PROVIDER = "local"

tests/encryption_/test_management.py

@@ -114,6 +114,9 @@ def test_show_encrypted_fields_map(self):
                 self._compare_output(expected, command_output[model_key])
 
     def test_missing_key(self):
+        connection = connections["encrypted"]
+        auto_encryption_opts = connection.connection._options.auto_encryption_opts
+        kms_providers = auto_encryption_opts._kms_providers
         test_key = "encryption__patient.patient_record.ssn"
         msg = (
             f"Encryption key {test_key} not found. Have migrated the "
@@ -125,11 +128,10 @@ def test_missing_key(self):
                 call_command("showencryptedfieldsmap", "--database", "encrypted", verbosity=0)
         finally:
             # Replace the deleted key.
-            master_key = connections["encrypted"].settings_dict["KMS_CREDENTIALS"][
-                self.DEFAULT_KMS_PROVIDER
-            ]
-            connections["encrypted"].client_encryption.create_data_key(
-                kms_provider=self.DEFAULT_KMS_PROVIDER,
+            kms_provider = next(iter(kms_providers.keys()))
+            master_key = connection.settings_dict["KMS_CREDENTIALS"][kms_provider]
+            connection.client_encryption.create_data_key(
+                kms_provider=kms_provider,
                 master_key=master_key,
                 key_alt_names=[test_key],
             )