chore: Correct lint and imports, plus add testcases for exceptions check

Signed-off-by: Radhika Agrawal <agrawalradhika@google.com>

Author: agrawalradhika-cell

google/auth/aio/transport/mtls.py

@@ -13,23 +13,18 @@
 # limitations under the License.
 
 """
-Helper functions for mTLS in asyncio.
+Helper functions for mTLS in async.
 """
 
-import asyncio
-import contextlib
 import logging
-import os
-from os import environ, getenv, path
-import ssl
-import tempfile
-from typing import Optional
+from os import getenv, path
 
-from google.auth import exceptions
+import google.auth.transport._mtls_helper
 
 CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json"
 _LOGGER = logging.getLogger(__name__)
 
+
 def _check_config_path(config_path):
     """Checks for config file path. If it exists, returns the absolute path with user expansion;
     otherwise returns None.
@@ -53,16 +48,10 @@ def has_default_client_cert_source():
     Returns:
         bool: indicating if the default client cert source exists.
     """
-    if (
-	_check_config_path(CERTIFICATE_CONFIGURATION_DEFAULT_PATH)
-        is not None
-    ):
+    if _check_config_path(CERTIFICATE_CONFIGURATION_DEFAULT_PATH) is not None:
         return True
     cert_config_path = getenv("GOOGLE_API_CERTIFICATE_CONFIG")
-    if (
-        cert_config_path
-        and _check_config_path(cert_config_path) is not None
-    ):
+    if cert_config_path and _check_config_path(cert_config_path) is not None:
         return True
     return False
 
@@ -95,7 +84,9 @@ def get_client_ssl_credentials(
     """
 
     # Attempt to retrieve X.509 Workload cert and key.
-    cert, key =  google.auth.transport._mtls_helper._get_workload_cert_and_key(certificate_config_path)
+    cert, key = google.auth.transport._mtls_helper._get_workload_cert_and_key(
+        certificate_config_path
+    )
     if cert and key:
         return True, cert, key, None
 
@@ -128,4 +119,3 @@ def get_client_cert_and_key(client_cert_callback=None):
 
     has_cert, cert, key, _ = get_client_ssl_credentials(generate_encrypted_key=False)
     return has_cert, cert, key
-

tests/transport/test_aio_mtls_helper.py

@@ -1,88 +1,101 @@
-import os
-import pytest
+# Copyright 2024 Google LLC
+# Licensed under the Apache License, Version 2.0...
+
 from unittest import mock
+
+import pytest
+
 from google.auth import exceptions
-# Assuming the provided code is in a file named google/auth/transport/aio/mtls_helper.py
-from google.auth.transport.aio import mtls_helper
+from google.auth.aio.transport import mtls
 
 CERT_DATA = b"client-cert"
 KEY_DATA = b"client-key"
 
-class TestMTLSHelper:
 
-    @mock.patch("os.path.expanduser")
-    @mock.patch("os.path.exists")
+class TestMTLS:
+    @mock.patch("google.auth.aio.transport.mtls.path.expanduser")
+    @mock.patch("google.auth.aio.transport.mtls.path.exists")
     def test__check_config_path_exists(self, mock_exists, mock_expand):
         mock_expand.side_effect = lambda x: x.replace("~", "/home/user")
         mock_exists.return_value = True
-        
-        path = "/home/user/config.json"
-        result = mtls_helper._check_config_path("~/config.json")
-        
-        assert result == path
-        mock_exists.assert_called_with(path)
-
-    @mock.patch("os.path.exists", return_value=False)
+
+        input_path = "~/config.json"
+        expected_path = "/home/user/config.json"
+        result = mtls._check_config_path(input_path)
+
+        assert result == expected_path
+        mock_exists.assert_called_with(expected_path)
+
+    @mock.patch("google.auth.aio.transport.mtls.path.exists", return_value=False)
     def test__check_config_path_not_found(self, mock_exists):
-        result = mtls_helper._check_config_path("nonexistent.json")
+        result = mtls._check_config_path("nonexistent.json")
         assert result is None
 
-    @mock.patch("google.auth.transport.aio.mtls_helper._check_config_path")
-    @mock.patch("os.getenv")
-    def test_has_default_client_cert_source_default_path(self, mock_getenv, mock_check):
-        # Case 1: Default config path exists
-        mock_check.side_effect = lambda x: x if x == mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH else None
-        
-        assert mtls_helper.has_default_client_cert_source() is True
-
-    @mock.patch("google.auth.transport.aio.mtls_helper._check_config_path")
-    @mock.patch("os.getenv")
+    @mock.patch("google.auth.aio.transport.mtls._check_config_path")
+    @mock.patch("google.auth.aio.transport.mtls.getenv")
     def test_has_default_client_cert_source_env_var(self, mock_getenv, mock_check):
-        # Case 2: Default path doesn't exist, but env var path does
+        # Mocking so the default path fails but the env var path succeeds
         custom_path = "/custom/path.json"
-        mock_check.side_effect = lambda x: x if x == custom_path else None
+        mock_check.side_effect = lambda x: custom_path if x == custom_path else None
         mock_getenv.return_value = custom_path
-        
-        assert mtls_helper.has_default_client_cert_source() is True
 
-    @mock.patch("google.auth.transport.aio.mtls_helper._check_config_path", return_value=None)
-    def test_has_default_client_cert_source_none(self, mock_check):
-        assert mtls_helper.has_default_client_cert_source() is False
+        assert mtls.has_default_client_cert_source() is True
 
     @mock.patch("google.auth.transport._mtls_helper._get_workload_cert_and_key")
     def test_get_client_ssl_credentials_success(self, mock_workload):
         mock_workload.return_value = (CERT_DATA, KEY_DATA)
-        
-        success, cert, key, passphrase = mtls_helper.get_client_ssl_credentials()
-        
+
+        success, cert, key, passphrase = mtls.get_client_ssl_credentials()
+
         assert success is True
         assert cert == CERT_DATA
         assert key == KEY_DATA
         assert passphrase is None
 
-    @mock.patch("google.auth.transport._mtls_helper._get_workload_cert_and_key", return_value=(None, None))
-    def test_get_client_ssl_credentials_fail(self, mock_workload):
-        success, cert, key, passphrase = mtls_helper.get_client_ssl_credentials()
-        assert success is False
-        assert cert is None
-
     def test_get_client_cert_and_key_callback(self):
-        # Callback should take priority
+        # The callback should be tried first and return immediately
         callback = mock.Mock(return_value=(CERT_DATA, KEY_DATA))
-        
-        success, cert, key = mtls_helper.get_client_cert_and_key(callback)
-        
+
+        success, cert, key = mtls.get_client_cert_and_key(callback)
+
         assert success is True
         assert cert == CERT_DATA
         assert key == KEY_DATA
         callback.assert_called_once()
 
-    @mock.patch("google.auth.transport.aio.mtls_helper.get_client_ssl_credentials")
+    @mock.patch("google.auth.aio.transport.mtls.get_client_ssl_credentials")
     def test_get_client_cert_and_key_default(self, mock_get_ssl):
+        # If no callback, it should call get_client_ssl_credentials
         mock_get_ssl.return_value = (True, CERT_DATA, KEY_DATA, None)
-        
-        success, cert, key = mtls_helper.get_client_cert_and_key(None)
-        
+
+        success, cert, key = mtls.get_client_cert_and_key(None)
+
         assert success is True
         assert cert == CERT_DATA
         assert key == KEY_DATA
+        mock_get_ssl.assert_called_with(generate_encrypted_key=False)
+
+    @mock.patch("google.auth.transport._mtls_helper._get_workload_cert_and_key")
+    def test_get_client_ssl_credentials_error(self, mock_workload):
+        """Tests that ClientCertError is propagated correctly."""
+        # Setup the mock to raise the specific google-auth exception
+        mock_workload.side_effect = exceptions.ClientCertError(
+            "Failed to read metadata"
+        )
+
+        # Verify that calling our function raises the same exception
+        with pytest.raises(exceptions.ClientCertError, match="Failed to read metadata"):
+            mtls.get_client_ssl_credentials()
+
+    @mock.patch("google.auth.aio.transport.mtls.get_client_ssl_credentials")
+    def test_get_client_cert_and_key_exception_propagation(self, mock_get_ssl):
+        """Tests that get_client_cert_and_key propagates errors from its internal calls."""
+        mock_get_ssl.side_effect = exceptions.ClientCertError(
+            "Underlying credentials failed"
+        )
+
+        with pytest.raises(
+            exceptions.ClientCertError, match="Underlying credentials failed"
+        ):
+            # Pass None for callback so it attempts to call get_client_ssl_credentials
+            mtls.get_client_cert_and_key(client_cert_callback=None)