chore(generator): move get_client_cert_source to compat layer

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2

@@ -34,12 +34,15 @@ from google.auth.exceptions import MutualTLSChannelError
 from google.protobuf import json_format
 from urllib.parse import urlparse, urlunparse
 
+from google.auth.transport import mtls  # type: ignore
+
 try:
     # note: `#type: ignore` is added because the return type for `should_use_client_cert`
     # is different than that of the fallback implementation below. This will be removed once 
     # we bump the minimum supported version of google-auth.
     from google.auth.transport.mtls import should_use_client_cert  # type: ignore
 except ImportError:  # pragma: NO COVER
+    {# TODO: Remove client cert compatibility fallbacks when the minimum supported version of google-auth is >= 2.43.0 (currently 2.14.1+). #}
     def should_use_client_cert():
         """Returns whether client certificate should be used for mTLS."""
         use_client_cert = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
@@ -78,6 +81,25 @@ def read_environment_variables():
     return use_client_cert, use_mtls_endpoint, universe_domain_env
 
 
+def get_client_cert_source(provided_cert_source, use_cert_flag):
+    """Return the client cert source to be used by the client.
+
+    Args:
+        provided_cert_source (bytes): The client certificate source provided.
+        use_cert_flag (bool): A flag indicating whether to use the client certificate.
+
+    Returns:
+        bytes or None: The client cert source to be used by the client.
+    """
+    client_cert_source = None
+    if use_cert_flag:
+        if provided_cert_source:
+            client_cert_source = provided_cert_source
+        elif mtls.has_default_client_cert_source():
+            client_cert_source = mtls.default_client_cert_source()
+    return client_cert_source
+
+
 DEFAULT_UNIVERSE = "googleapis.com"
 
 

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2

@@ -30,7 +30,7 @@ from google.api_core import exceptions as core_exceptions
 from google.api_core import extended_operation
 {% endif %}
 from google.api_core import gapic_v1
-from {{package_path}}._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables
+from {{package_path}}._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables, get_client_cert_source
 {% if has_auto_populated_fields %}
 from {{package_path}}._compat import setup_request_id
 {% endif %}
@@ -295,24 +295,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
 
 
-    @staticmethod
-    def _get_client_cert_source(provided_cert_source, use_cert_flag):
-        """Return the client cert source to be used by the client.
-
-        Args:
-            provided_cert_source (bytes): The client certificate source provided.
-            use_cert_flag (bool): A flag indicating whether to use the client certificate.
 
-        Returns:
-            bytes or None: The client cert source to be used by the client.
-        """
-        client_cert_source = None
-        if use_cert_flag:
-            if provided_cert_source:
-                client_cert_source = provided_cert_source
-            elif mtls.has_default_client_cert_source():
-                client_cert_source = mtls.default_client_cert_source()
-        return client_cert_source
 
 
     def _validate_universe_domain(self):
@@ -441,7 +424,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables()
-        self._client_cert_source = {{ service.client_name }}._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
+        self._client_cert_source = get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
         self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 

packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2

@@ -168,18 +168,6 @@ def set_event_loop():
             asyncio.set_event_loop(None)
 
 
-def test__get_client_cert_source():
-    mock_provided_cert_source = mock.Mock()
-    mock_default_cert_source = mock.Mock()
-
-    assert {{ service.client_name }}._get_client_cert_source(None, False) is None
-    assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, False) is None
-    assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source
-
-    with mock.patch('google.auth.transport.mtls.has_default_client_cert_source', return_value=True):
-        with mock.patch('google.auth.transport.mtls.default_client_cert_source', return_value=mock_default_cert_source):
-            assert {{ service.client_name }}._get_client_cert_source(None, True) is mock_default_cert_source
-            assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
 
 
 

packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_compat.py.j2

@@ -23,7 +23,7 @@ import google.auth.transport.mtls
 
 {% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
 from {{package_path}}._compat import transcode_request
-from {{package_path}}._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables
+from {{package_path}}._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables, get_client_cert_source
 {% if has_auto_populated_fields %}
 from {{package_path}}._compat import setup_request_id
 {% endif %}
@@ -531,4 +531,21 @@ def test_read_environment_variables():
         with pytest.raises(MutualTLSChannelError):
             read_environment_variables()
 
+
+def test_get_client_cert_source():
+    mock_provided_cert_source = mock.Mock()
+    mock_default_cert_source = mock.Mock()
+
+    assert get_client_cert_source(None, False) is None
+    assert get_client_cert_source(mock_provided_cert_source, False) is None
+    assert get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source
+
+    with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=True):
+        with mock.patch("google.auth.transport.mtls.default_client_cert_source", return_value=mock_default_cert_source):
+            assert get_client_cert_source(None, True) is mock_default_cert_source
+            assert get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
+
+    with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=False):
+        assert get_client_cert_source(None, True) is None
+
 {% endblock %}

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py

@@ -26,6 +26,8 @@
 from google.protobuf import json_format
 from urllib.parse import urlparse, urlunparse
 
+from google.auth.transport import mtls  # type: ignore
+
 try:
     # note: `#type: ignore` is added because the return type for `should_use_client_cert`
     # is different than that of the fallback implementation below. This will be removed once
@@ -67,6 +69,25 @@ def read_environment_variables():
     return use_client_cert, use_mtls_endpoint, universe_domain_env
 
 
+def get_client_cert_source(provided_cert_source, use_cert_flag):
+    """Return the client cert source to be used by the client.
+
+    Args:
+        provided_cert_source (bytes): The client certificate source provided.
+        use_cert_flag (bool): A flag indicating whether to use the client certificate.
+
+    Returns:
+        bytes or None: The client cert source to be used by the client.
+    """
+    client_cert_source = None
+    if use_cert_flag:
+        if provided_cert_source:
+            client_cert_source = provided_cert_source
+        elif mtls.has_default_client_cert_source():
+            client_cert_source = mtls.default_client_cert_source()
+    return client_cert_source
+
+
 DEFAULT_UNIVERSE = "googleapis.com"
 
 

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py

@@ -27,7 +27,7 @@
 from google.api_core import client_options as client_options_lib
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
-from google.cloud.asset_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables
+from google.cloud.asset_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables, get_client_cert_source
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials             # type: ignore
 from google.auth.transport import mtls                            # type: ignore
@@ -348,25 +348,6 @@ def get_mtls_endpoint_and_cert_source(cls, client_options: Optional[client_optio
 
         return api_endpoint, client_cert_source
 
-    @staticmethod
-    def _get_client_cert_source(provided_cert_source, use_cert_flag):
-        """Return the client cert source to be used by the client.
-
-        Args:
-            provided_cert_source (bytes): The client certificate source provided.
-            use_cert_flag (bool): A flag indicating whether to use the client certificate.
-
-        Returns:
-            bytes or None: The client cert source to be used by the client.
-        """
-        client_cert_source = None
-        if use_cert_flag:
-            if provided_cert_source:
-                client_cert_source = provided_cert_source
-            elif mtls.has_default_client_cert_source():
-                client_cert_source = mtls.default_client_cert_source()
-        return client_cert_source
-
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.
 
@@ -490,7 +471,7 @@ def __init__(self, *,
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables()
-        self._client_cert_source = AssetServiceClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
+        self._client_cert_source = get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
         self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 

packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py

@@ -132,20 +132,6 @@ def set_event_loop():
             asyncio.set_event_loop(None)
 
 
-def test__get_client_cert_source():
-    mock_provided_cert_source = mock.Mock()
-    mock_default_cert_source = mock.Mock()
-
-    assert AssetServiceClient._get_client_cert_source(None, False) is None
-    assert AssetServiceClient._get_client_cert_source(mock_provided_cert_source, False) is None
-    assert AssetServiceClient._get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source
-
-    with mock.patch('google.auth.transport.mtls.has_default_client_cert_source', return_value=True):
-        with mock.patch('google.auth.transport.mtls.default_client_cert_source', return_value=mock_default_cert_source):
-            assert AssetServiceClient._get_client_cert_source(None, True) is mock_default_cert_source
-            assert AssetServiceClient._get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
-
-
 @pytest.mark.parametrize("error_code,cred_info_json,show_cred_info", [
     (401, CRED_INFO_JSON, True),
     (403, CRED_INFO_JSON, True),

packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_compat.py

@@ -24,7 +24,7 @@
 import google.auth.transport.mtls
 
 from google.cloud.asset_v1._compat import transcode_request
-from google.cloud.asset_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables
+from google.cloud.asset_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables, get_client_cert_source
 
 from google.auth.exceptions import MutualTLSChannelError
 from google.api_core.universe import EmptyUniverseError
@@ -423,3 +423,20 @@ def test_read_environment_variables():
     with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "invalid"}):
         with pytest.raises(MutualTLSChannelError):
             read_environment_variables()
+
+
+def test_get_client_cert_source():
+    mock_provided_cert_source = mock.Mock()
+    mock_default_cert_source = mock.Mock()
+
+    assert get_client_cert_source(None, False) is None
+    assert get_client_cert_source(mock_provided_cert_source, False) is None
+    assert get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source
+
+    with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=True):
+        with mock.patch("google.auth.transport.mtls.default_client_cert_source", return_value=mock_default_cert_source):
+            assert get_client_cert_source(None, True) is mock_default_cert_source
+            assert get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
+
+    with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=False):
+        assert get_client_cert_source(None, True) is None

packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py

@@ -26,6 +26,8 @@
 from google.protobuf import json_format
 from urllib.parse import urlparse, urlunparse
 
+from google.auth.transport import mtls  # type: ignore
+
 try:
     # note: `#type: ignore` is added because the return type for `should_use_client_cert`
     # is different than that of the fallback implementation below. This will be removed once
@@ -67,6 +69,25 @@ def read_environment_variables():
     return use_client_cert, use_mtls_endpoint, universe_domain_env
 
 
+def get_client_cert_source(provided_cert_source, use_cert_flag):
+    """Return the client cert source to be used by the client.
+
+    Args:
+        provided_cert_source (bytes): The client certificate source provided.
+        use_cert_flag (bool): A flag indicating whether to use the client certificate.
+
+    Returns:
+        bytes or None: The client cert source to be used by the client.
+    """
+    client_cert_source = None
+    if use_cert_flag:
+        if provided_cert_source:
+            client_cert_source = provided_cert_source
+        elif mtls.has_default_client_cert_source():
+            client_cert_source = mtls.default_client_cert_source()
+    return client_cert_source
+
+
 DEFAULT_UNIVERSE = "googleapis.com"
 
 

packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py

@@ -27,7 +27,7 @@
 from google.api_core import client_options as client_options_lib
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
-from google.iam.credentials_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables
+from google.iam.credentials_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables, get_client_cert_source
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials             # type: ignore
 from google.auth.transport import mtls                            # type: ignore
@@ -285,25 +285,6 @@ def get_mtls_endpoint_and_cert_source(cls, client_options: Optional[client_optio
 
         return api_endpoint, client_cert_source
 
-    @staticmethod
-    def _get_client_cert_source(provided_cert_source, use_cert_flag):
-        """Return the client cert source to be used by the client.
-
-        Args:
-            provided_cert_source (bytes): The client certificate source provided.
-            use_cert_flag (bool): A flag indicating whether to use the client certificate.
-
-        Returns:
-            bytes or None: The client cert source to be used by the client.
-        """
-        client_cert_source = None
-        if use_cert_flag:
-            if provided_cert_source:
-                client_cert_source = provided_cert_source
-            elif mtls.has_default_client_cert_source():
-                client_cert_source = mtls.default_client_cert_source()
-        return client_cert_source
-
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.
 
@@ -427,7 +408,7 @@ def __init__(self, *,
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables()
-        self._client_cert_source = IAMCredentialsClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
+        self._client_cert_source = get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
         self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 

packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_compat.py

@@ -24,7 +24,7 @@
 import google.auth.transport.mtls
 
 from google.iam.credentials_v1._compat import transcode_request
-from google.iam.credentials_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables
+from google.iam.credentials_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert, read_environment_variables, get_client_cert_source
 
 from google.auth.exceptions import MutualTLSChannelError
 from google.api_core.universe import EmptyUniverseError
@@ -423,3 +423,20 @@ def test_read_environment_variables():
     with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "invalid"}):
         with pytest.raises(MutualTLSChannelError):
             read_environment_variables()
+
+
+def test_get_client_cert_source():
+    mock_provided_cert_source = mock.Mock()
+    mock_default_cert_source = mock.Mock()
+
+    assert get_client_cert_source(None, False) is None
+    assert get_client_cert_source(mock_provided_cert_source, False) is None
+    assert get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source
+
+    with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=True):
+        with mock.patch("google.auth.transport.mtls.default_client_cert_source", return_value=mock_default_cert_source):
+            assert get_client_cert_source(None, True) is mock_default_cert_source
+            assert get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
+
+    with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=False):
+        assert get_client_cert_source(None, True) is None

packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py

@@ -123,20 +123,6 @@ def set_event_loop():
             asyncio.set_event_loop(None)
 
 
-def test__get_client_cert_source():
-    mock_provided_cert_source = mock.Mock()
-    mock_default_cert_source = mock.Mock()
-
-    assert IAMCredentialsClient._get_client_cert_source(None, False) is None
-    assert IAMCredentialsClient._get_client_cert_source(mock_provided_cert_source, False) is None
-    assert IAMCredentialsClient._get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source
-
-    with mock.patch('google.auth.transport.mtls.has_default_client_cert_source', return_value=True):
-        with mock.patch('google.auth.transport.mtls.default_client_cert_source', return_value=mock_default_cert_source):
-            assert IAMCredentialsClient._get_client_cert_source(None, True) is mock_default_cert_source
-            assert IAMCredentialsClient._get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
-
-
 @pytest.mark.parametrize("error_code,cred_info_json,show_cred_info", [
     (401, CRED_INFO_JSON, True),
     (403, CRED_INFO_JSON, True),