Skip to content

chore(generator): move _use_client_cert_effective to _compat.py.j2 #17963

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hebaalazzeh wants to merge 7 commits into
base: main
Choose a base branch
from
Open

chore(generator): move _use_client_cert_effective to _compat.py.j2 #17963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4ca8ee4
chore(generator): remove _use_client_cert_effective from gapic-generator
hebaalazzeh Jul 31, 2026
74ec086
update goldens
hebaalazzeh Jul 31, 2026
0dcc129
update goldens
hebaalazzeh Jul 31, 2026
285b8f5
update goldens & address feedback
hebaalazzeh Jul 31, 2026
e64984d
Merge branch 'main' into remove-use-client-cert-effective
hebaalazzeh Jul 31, 2026
fdb2ef8
update goldens
hebaalazzeh Jul 31, 2026
3279a6f
update goldens
hebaalazzeh Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 21 additions & 7 deletions packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ falling back to the local implementation if not present. #}
{# TODO(https://github.com/googleapis/google-cloud-python/issues/17883):
Backfill compatibility functions being removed from the client layer. #}

import os
import json
{% if has_auto_populated_fields %}
import uuid
Expand All @@ -23,16 +24,16 @@ from typing import Any, Dict, List, Optional, Tuple
from typing import Union
{% endif %}

from google.api_core import path_template
from google.protobuf import json_format
from urllib.parse import urlparse, urlunparse

from google.auth.exceptions import MutualTLSChannelError
from google.api_core.universe import EmptyUniverseError

{% if has_auto_populated_fields %}
import google.protobuf.message

{% endif %}
from google.api_core import path_template
from google.api_core.universe import EmptyUniverseError
from google.auth.exceptions import MutualTLSChannelError
from google.auth.transport import mtls
from google.protobuf import json_format
from urllib.parse import urlparse, urlunparse

DEFAULT_UNIVERSE = "googleapis.com"

Expand Down Expand Up @@ -151,6 +152,19 @@ def get_universe_domain(
raise EmptyUniverseError()
return resolved

def should_use_client_cert() -> Optional[bool]:
"""Returns whether client certificate should be used for mTLS."""
if hasattr(mtls, "should_use_client_cert"):
return mtls.should_use_client_cert()
else:

@daniel-sanche daniel-sanche Jul 31, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thought the pattern you were using for this file was to try importing from the helper library, and provide the backup implementation if there's an ImportError. Do you plan to use hasattr for everything instead now?

use_client_cert = os.getenv(
"GOOGLE_API_USE_CLIENT_CERTIFICATE"
) or os.getenv("CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE")

if use_client_cert:
return use_client_cert.lower() == "true"
return None

{% if has_auto_populated_fields %}

def setup_request_id(
Expand Down
32 changes: 3 additions & 29 deletions ...c-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
from {{package_path}}._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert
{% if has_auto_populated_fields %}
from {{package_path}}._compat import setup_request_id
{% endif %}
Expand Down Expand Up @@ -156,32 +156,6 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):

_DEFAULT_UNIVERSE = "googleapis.com"

@staticmethod
def _use_client_cert_effective():
"""Returns whether client certificate should be used for mTLS if the
google-auth version supports should_use_client_cert automatic mTLS enablement.

Alternatively, read from the GOOGLE_API_USE_CLIENT_CERTIFICATE env var.

Returns:
bool: whether client certificate should be used for mTLS
Raises:
ValueError: (If using a version of google-auth without should_use_client_cert and
GOOGLE_API_USE_CLIENT_CERTIFICATE is set to an unexpected value.)
"""
# check if google-auth version supports should_use_client_cert for automatic mTLS enablement
if hasattr(mtls, "should_use_client_cert"): # pragma: NO COVER
return mtls.should_use_client_cert()
else: # pragma: NO COVER
# if unsupported, fallback to reading from env var
use_client_cert_str = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
if use_client_cert_str not in ("true", "false"):
raise ValueError(
"Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
" either `true` or `false`"
)
return use_client_cert_str == "true"

@classmethod
def from_service_account_info(cls, info: dict, *args, **kwargs):
"""Creates an instance of this client using the provided credentials
Expand Down Expand Up @@ -296,7 +270,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
DeprecationWarning)
if client_options is None:
client_options = client_options_lib.ClientOptions()
use_client_cert = {{ service.client_name }}._use_client_cert_effective()
use_client_cert = should_use_client_cert()
use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
if use_mtls_endpoint not in ("auto", "never", "always"):
raise MutualTLSChannelError("Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`")
Expand Down Expand Up @@ -333,7 +307,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
google.auth.exceptions.MutualTLSChannelError: If GOOGLE_API_USE_MTLS_ENDPOINT
is not any of ["auto", "never", "always"].
"""
use_client_cert = {{ service.client_name }}._use_client_cert_effective()
use_client_cert = should_use_client_cert()
use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
if use_mtls_endpoint not in ("auto", "never", "always"):
Expand Down
84 changes: 0 additions & 84 deletions .../gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,90 +212,6 @@ def test__read_environment_variables():
assert {{ service.client_name }}._read_environment_variables() == (False, "auto", "foo.com")


def test_use_client_cert_effective():
# Test case 1: Test when `should_use_client_cert` returns True.
# We mock the `should_use_client_cert` function to simulate a scenario where
# the google-auth library supports automatic mTLS and determines that a
# client certificate should be used.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=True):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 2: Test when `should_use_client_cert` returns False.
# We mock the `should_use_client_cert` function to simulate a scenario where
# the google-auth library supports automatic mTLS and determines that a
# client certificate should NOT be used.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=False):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 3: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "true".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 4: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "false".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 5: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "True".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "True"}):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 6: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "False".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "False"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 7: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "TRUE".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "TRUE"}):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 8: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "FALSE".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "FALSE"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 9: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is not set.
# In this case, the method should return False, which is the default value.
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, clear=True):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 10: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to an invalid value.
# The method should raise a ValueError as the environment variable must be either
# "true" or "false".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "unsupported"}):
with pytest.raises(ValueError):
{{ service.client_name }}._use_client_cert_effective()

# Test case 11: Test when `should_use_client_cert` is available and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to an invalid value.
# The method should return False as the environment variable is set to an invalid value.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "unsupported"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 12: Test when `should_use_client_cert` is available and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is unset. Also,
# the GOOGLE_API_CONFIG environment variable is unset.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": ""}):
with mock.patch.dict(os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": ""}):
assert {{ service.client_name }}._use_client_cert_effective() is False

def test__get_client_cert_source():
mock_provided_cert_source = mock.Mock()
mock_default_cert_source = mock.Mock()
Expand Down
31 changes: 28 additions & 3 deletions ...es/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_compat.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,21 +12,23 @@ Backfill compatibility functions tests being removed from the client layer. #}

import json
import pytest
import os
{% if has_auto_populated_fields %}
import re
{% endif %}

from unittest import mock

from google.protobuf import descriptor_pb2
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}} import _compat as universe
{% if has_auto_populated_fields %}
from {{package_path}}._compat import setup_request_id
{% endif %}
from {{package_path}}._compat import transcode_request
from {{package_path}} import _compat as universe

from google.protobuf import descriptor_pb2
from google.auth.exceptions import MutualTLSChannelError


Expand Down Expand Up @@ -226,6 +228,29 @@ def test_get_api_endpoint(
)


def test_should_use_client_cert_with_should_use_client_cert():
mock_mtls = mock.Mock()
mock_mtls.should_use_client_cert.return_value = True
with mock.patch(f"{universe.__name__}.mtls", mock_mtls):
assert universe.should_use_client_cert() is True

mock_mtls.should_use_client_cert.return_value = False
with mock.patch(f"{universe.__name__}.mtls", mock_mtls):
assert universe.should_use_client_cert() is False


def test_should_use_client_cert_fallback_env():
mock_mtls = mock.Mock(spec=[])
with mock.patch(f"{universe.__name__}.mtls", mock_mtls):
with mock.patch.dict(os.environ, {"CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "true"}, clear=True):
assert universe.should_use_client_cert() is True

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}, clear=True):
assert universe.should_use_client_cert() is False

with mock.patch.dict(os.environ, {}, clear=True):
assert universe.should_use_client_cert() is None

{% if has_auto_populated_fields %}

class MockRequest:
Expand Down
21 changes: 17 additions & 4 deletions packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,18 +15,18 @@
#
"""A compatibility module for older versions of google-api-core."""

import os
import json

from typing import Any, Dict, List, Optional, Tuple

from google.api_core import path_template
from google.api_core.universe import EmptyUniverseError
from google.auth.exceptions import MutualTLSChannelError
from google.auth.transport import mtls
from google.protobuf import json_format
from urllib.parse import urlparse, urlunparse

from google.auth.exceptions import MutualTLSChannelError
from google.api_core.universe import EmptyUniverseError


DEFAULT_UNIVERSE = "googleapis.com"


Expand Down Expand Up @@ -144,6 +144,19 @@ def get_universe_domain(
raise EmptyUniverseError()
return resolved

def should_use_client_cert() -> Optional[bool]:
"""Returns whether client certificate should be used for mTLS."""
if hasattr(mtls, "should_use_client_cert"):
return mtls.should_use_client_cert()
else:
use_client_cert = os.getenv(
"GOOGLE_API_USE_CLIENT_CERTIFICATE"
) or os.getenv("CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE")

if use_client_cert:
return use_client_cert.lower() == "true"
return None


def transcode_request(
http_options: List[Dict[str, str]],
Expand Down
32 changes: 3 additions & 29 deletions ...or/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
from google.cloud.asset_v1._compat import get_universe_domain, get_api_endpoint, get_default_mtls_endpoint, should_use_client_cert
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
Expand Down Expand Up @@ -108,32 +108,6 @@ class AssetServiceClient(metaclass=AssetServiceClientMeta):
_DEFAULT_ENDPOINT_TEMPLATE = "cloudasset.{UNIVERSE_DOMAIN}"
_DEFAULT_UNIVERSE = "googleapis.com"

@staticmethod
def _use_client_cert_effective():
"""Returns whether client certificate should be used for mTLS if the
google-auth version supports should_use_client_cert automatic mTLS enablement.

Alternatively, read from the GOOGLE_API_USE_CLIENT_CERTIFICATE env var.

Returns:
bool: whether client certificate should be used for mTLS
Raises:
ValueError: (If using a version of google-auth without should_use_client_cert and
GOOGLE_API_USE_CLIENT_CERTIFICATE is set to an unexpected value.)
"""
# check if google-auth version supports should_use_client_cert for automatic mTLS enablement
if hasattr(mtls, "should_use_client_cert"): # pragma: NO COVER
return mtls.should_use_client_cert()
else: # pragma: NO COVER
# if unsupported, fallback to reading from env var
use_client_cert_str = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
if use_client_cert_str not in ("true", "false"):
raise ValueError(
"Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
" either `true` or `false`"
)
return use_client_cert_str == "true"

@classmethod
def from_service_account_info(cls, info: dict, *args, **kwargs):
"""Creates an instance of this client using the provided credentials
Expand Down Expand Up @@ -351,7 +325,7 @@ def get_mtls_endpoint_and_cert_source(cls, client_options: Optional[client_optio
DeprecationWarning)
if client_options is None:
client_options = client_options_lib.ClientOptions()
use_client_cert = AssetServiceClient._use_client_cert_effective()
use_client_cert = should_use_client_cert()
use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
if use_mtls_endpoint not in ("auto", "never", "always"):
raise MutualTLSChannelError("Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`")
Expand Down Expand Up @@ -388,7 +362,7 @@ def _read_environment_variables():
google.auth.exceptions.MutualTLSChannelError: If GOOGLE_API_USE_MTLS_ENDPOINT
is not any of ["auto", "never", "always"].
"""
use_client_cert = AssetServiceClient._use_client_cert_effective()
use_client_cert = should_use_client_cert()
use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
if use_mtls_endpoint not in ("auto", "never", "always"):
Expand Down
Loading
Loading