PYTHON-5373 test client auth on cloud-dev (#2423)

.evergreen/generated_configs/variants.yml

@@ -75,6 +75,8 @@ buildvariants:
     display_name: Atlas connect RHEL8
     run_on:
       - rhel87-small
+    expansions:
+      TEST_NAME: atlas_connect
     tags: [pr]
 
   # Atlas data lake tests

.evergreen/scripts/generate_config.py

@@ -425,6 +425,7 @@ def create_atlas_connect_variants():
             get_variant_name("Atlas connect", host),
             tags=["pr"],
             host=DEFAULT_HOST,
+            expansions=dict(TEST_NAME="atlas_connect"),
         )
     ]
 

.evergreen/scripts/setup_tests.py

@@ -417,7 +417,18 @@ def handle_test_env() -> None:
             run_command(f"bash {auth_aws_dir}/setup-secrets.sh")
 
     if test_name == "atlas_connect":
-        get_secrets("drivers/atlas_connect")
+        secrets = get_secrets("drivers/atlas_connect")
+
+        # Write file with Atlas X509 client certificate:
+        decoded = base64.b64decode(secrets["ATLAS_X509_DEV_CERT_BASE64"]).decode("utf8")
+        cert_file = ROOT / ".evergreen/atlas_x509_dev_client_certificate.pem"
+        with cert_file.open("w") as file:
+            file.write(decoded)
+        write_env(
+            "ATLAS_X509_DEV_WITH_CERT",
+            secrets["ATLAS_X509_DEV"] + "&tlsCertificateKeyFile=" + str(cert_file),
+        )
+
         # We do not want the default client_context to be initialized.
         write_env("DISABLE_CONTEXT")
 

.gitignore

@@ -30,6 +30,7 @@ expansion.yml
 .evergreen/scripts/test-env.sh
 specifications/
 results.json
+.evergreen/atlas_x509_dev_client_certificate.pem
 
 # Lambda temp files
 test/lambda/.aws-sam

test/atlas/test_connection.py

@@ -42,6 +42,7 @@ URIS = {
     "ATLAS_SRV_FREE": os.environ.get("ATLAS_SRV_FREE"),
     "ATLAS_SRV_TLS11": os.environ.get("ATLAS_SRV_TLS11"),
     "ATLAS_SRV_TLS12": os.environ.get("ATLAS_SRV_TLS12"),
+    "ATLAS_X509_DEV_WITH_CERT": os.environ.get("ATLAS_X509_DEV_WITH_CERT"),
 }
 
 
@@ -91,6 +92,9 @@ class TestAtlasConnect(PyMongoTestCase):
     def test_srv_tls_12(self):
         self.connect_srv(URIS["ATLAS_SRV_TLS12"])
 
+    def test_x509_with_cert(self):
+        self.connect(URIS["ATLAS_X509_DEV_WITH_CERT"])
+
     def test_uniqueness(self):
         """Ensure that we don't accidentally duplicate the test URIs."""
         uri_to_names = defaultdict(list)