fix: allow user to customize context aware metadata path in _mtls_helper

google/auth/transport/_mtls_helper.py

@@ -141,13 +141,17 @@ def _run_cert_provider_command(command, expect_encrypted_key=False):
     return cert_match[0], key_match[0], None
 
 
-def get_client_ssl_credentials(generate_encrypted_key=False):
+def get_client_ssl_credentials(
+    generate_encrypted_key=False,
+    context_aware_metadata_path=CONTEXT_AWARE_METADATA_PATH,
+):
     """Returns the client side certificate, private key and passphrase.
 
     Args:
         generate_encrypted_key (bool): If set to True, encrypted private key
             and passphrase will be generated; otherwise, unencrypted private key
             will be generated and passphrase will be None.
+        context_aware_metadata_path (str): The context_aware_metadata.json file path.
 
     Returns:
         Tuple[bool, bytes, bytes, bytes]:
@@ -158,7 +162,7 @@ def get_client_ssl_credentials(generate_encrypted_key=False):
         google.auth.exceptions.ClientCertError: if problems occurs when getting
             the cert, key and passphrase.
     """
-    metadata_path = _check_dca_metadata_path(CONTEXT_AWARE_METADATA_PATH)
+    metadata_path = _check_dca_metadata_path(context_aware_metadata_path)
 
     if metadata_path:
         metadata_json = _read_dca_metadata_file(metadata_path)

tests/transport/test__mtls_helper.py

@@ -358,6 +358,39 @@ def test_missing_cert_command(
         with pytest.raises(exceptions.ClientCertError):
             _mtls_helper.get_client_ssl_credentials()
 
+    @mock.patch(
+        "google.auth.transport._mtls_helper._run_cert_provider_command", autospec=True
+    )
+    @mock.patch(
+        "google.auth.transport._mtls_helper._read_dca_metadata_file", autospec=True
+    )
+    @mock.patch(
+        "google.auth.transport._mtls_helper._check_dca_metadata_path", autospec=True
+    )
+    def test_customize_context_aware_metadata_path(
+        self,
+        mock_check_dca_metadata_path,
+        mock_read_dca_metadata_file,
+        mock_run_cert_provider_command,
+    ):
+        context_aware_metadata_path = "/path/to/metata/data"
+        mock_check_dca_metadata_path.return_value = context_aware_metadata_path
+        mock_read_dca_metadata_file.return_value = {
+            "cert_provider_command": ["command"]
+        }
+        mock_run_cert_provider_command.return_value = (b"cert", b"key", None)
+
+        has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials(
+            context_aware_metadata_path=context_aware_metadata_path
+        )
+
+        assert has_cert
+        assert cert == b"cert"
+        assert key == b"key"
+        assert passphrase is None
+        mock_check_dca_metadata_path.assert_called_with(context_aware_metadata_path)
+        mock_read_dca_metadata_file.assert_called_with(context_aware_metadata_path)
+
 
 class TestGetClientCertAndKey(object):
     def test_callback_success(self):