fix use of placeholders

Author: blink1073

test/asynchronous/helpers.py

@@ -120,6 +120,11 @@
     "privateKey": os.environ.get("FLE_GCP_PRIVATEKEY", ""),
 }
 KMIP_CREDS = {"endpoint": os.environ.get("FLE_KMIP_ENDPOINT", "localhost:5698")}
+AWS_TEMP_CREDS = {
+    "accessKeyId": os.environ.get("CSFLE_AWS_TEMP_ACCESS_KEY_ID", ""),
+    "secretAccessKey": os.environ.get("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY", ""),
+    "sessionToken": os.environ.get("CSFLE_AWS_TEMP_SESSION_TOKEN", ""),
+}
 
 ALL_KMS_PROVIDERS = dict(
     aws=AWS_CREDS,

test/asynchronous/test_encryption.py

@@ -57,6 +57,7 @@
 from test.asynchronous.helpers import (
     ALL_KMS_PROVIDERS,
     AWS_CREDS,
+    AWS_TEMP_CREDS,
     AZURE_CREDS,
     CA_PEM,
     CLIENT_PEM,
@@ -618,12 +619,6 @@ async def test_with_statement(self):
 
 
 # Spec tests
-AWS_TEMP_CREDS = {
-    "accessKeyId": os.environ.get("CSFLE_AWS_TEMP_ACCESS_KEY_ID", ""),
-    "secretAccessKey": os.environ.get("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY", ""),
-    "sessionToken": os.environ.get("CSFLE_AWS_TEMP_SESSION_TOKEN", ""),
-}
-
 AWS_TEMP_NO_SESSION_CREDS = {
     "accessKeyId": os.environ.get("CSFLE_AWS_TEMP_ACCESS_KEY_ID", ""),
     "secretAccessKey": os.environ.get("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY", ""),

test/helpers.py

@@ -120,6 +120,11 @@
     "privateKey": os.environ.get("FLE_GCP_PRIVATEKEY", ""),
 }
 KMIP_CREDS = {"endpoint": os.environ.get("FLE_KMIP_ENDPOINT", "localhost:5698")}
+AWS_TEMP_CREDS = {
+    "accessKeyId": os.environ.get("CSFLE_AWS_TEMP_ACCESS_KEY_ID", ""),
+    "secretAccessKey": os.environ.get("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY", ""),
+    "sessionToken": os.environ.get("CSFLE_AWS_TEMP_SESSION_TOKEN", ""),
+}
 
 ALL_KMS_PROVIDERS = dict(
     aws=AWS_CREDS,

test/test_encryption.py

@@ -57,6 +57,7 @@
 from test.helpers import (
     ALL_KMS_PROVIDERS,
     AWS_CREDS,
+    AWS_TEMP_CREDS,
     AZURE_CREDS,
     CA_PEM,
     CLIENT_PEM,
@@ -616,12 +617,6 @@ def test_with_statement(self):
 
 
 # Spec tests
-AWS_TEMP_CREDS = {
-    "accessKeyId": os.environ.get("CSFLE_AWS_TEMP_ACCESS_KEY_ID", ""),
-    "secretAccessKey": os.environ.get("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY", ""),
-    "sessionToken": os.environ.get("CSFLE_AWS_TEMP_SESSION_TOKEN", ""),
-}
-
 AWS_TEMP_NO_SESSION_CREDS = {
     "accessKeyId": os.environ.get("CSFLE_AWS_TEMP_ACCESS_KEY_ID", ""),
     "secretAccessKey": os.environ.get("CSFLE_AWS_TEMP_SECRET_ACCESS_KEY", ""),

test/unified_format_shared.py

@@ -28,6 +28,7 @@
 from test.helpers import (
     AWS_CREDS,
     AWS_CREDS_2,
+    AWS_TEMP_CREDS,
     AZURE_CREDS,
     CA_PEM,
     CLIENT_PEM,
@@ -118,10 +119,22 @@
     ("kmip", KMIP_CREDS),
     ("kmip:name1", KMIP_CREDS),
 ]:
+    # Use the temp aws creds for autoEncryptOpts.
+    if provider_name == "aws":
+        for key, value in AWS_TEMP_CREDS.items():
+            placeholder = f"/autoEncryptOpts/kmsProviders/{provider_name}/{key}"
+            PLACEHOLDER_MAP[placeholder] = value
+
     for key, value in provider_data.items():
         placeholder = f"/clientEncryptionOpts/kmsProviders/{provider_name}/{key}"
         PLACEHOLDER_MAP[placeholder] = value
 
+        if provider_name == "aws":
+            continue
+
+        placeholder = f"/autoEncryptOpts/kmsProviders/{provider_name}/{key}"
+        PLACEHOLDER_MAP[placeholder] = value
+
 OIDC_ENV = os.environ.get("OIDC_ENV", "test")
 if OIDC_ENV == "test":
     PLACEHOLDER_MAP["/uriOptions/authMechanismProperties"] = {"ENVIRONMENT": "test"}