refactor(api-core): remove deprecated fallback client cert helpers and environment parsing

Author: hebaalazzeh

packages/google-api-core/google/api_core/gapic_v1/client_utils.py

@@ -127,88 +127,3 @@ def get_universe_domain(
         raise ValueError("Universe Domain cannot be an empty string.")
     return universe_domain
 
-
-def use_client_cert_effective() -> bool:
-    """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"
-
-
-def get_client_cert_source(
-    provided_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
-    use_cert_flag: bool,
-) -> Optional[Callable[[], Tuple[bytes, bytes]]]:
-    """Return the client cert source to be used by the client.
-
-    Args:
-        provided_cert_source (Callable[[], Tuple[bytes, bytes]]): The client certificate source provided.
-        use_cert_flag (bool): A flag indicating whether to use the
-            client certificate.
-
-    Returns:
-        Callable[[], Tuple[bytes, bytes]] or None: The client cert source to be used by the client.
-    """
-    if use_cert_flag:
-        if provided_cert_source:
-            return provided_cert_source
-        elif (
-            hasattr(mtls, "has_default_client_cert_source")
-            and mtls.has_default_client_cert_source()
-        ):
-            return mtls.default_client_cert_source()
-        else:
-            raise ValueError(
-                "Client certificate is required for mTLS, but no client certificate source was provided or found."
-            )
-    return None
-
-
-def read_environment_variables() -> Tuple[bool, str, Optional[str]]:
-    """Returns the environment variables used by the client.
-
-    Returns:
-        Tuple[bool, str, Optional[str]]: returns the
-        GOOGLE_API_USE_CLIENT_CERTIFICATE, GOOGLE_API_USE_MTLS_ENDPOINT,
-        and GOOGLE_CLOUD_UNIVERSE_DOMAIN environment variables.
-
-    Raises:
-        ValueError: If GOOGLE_API_USE_CLIENT_CERTIFICATE is not
-            any of ["true", "false"].
-        google.auth.exceptions.MutualTLSChannelError: If
-            GOOGLE_API_USE_MTLS_ENDPOINT is not any of
-            ["auto", "never", "always"].
-    """
-    use_client_cert = use_client_cert_effective()
-    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"):
-        raise MutualTLSChannelError(
-            "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` "
-            "must be `never`, `auto` or `always`"
-        )
-    return use_client_cert, use_mtls_endpoint, universe_domain_env

packages/google-api-core/tests/unit/gapic/test_client_utils.py

@@ -22,11 +22,8 @@
 
 from google.api_core.gapic_v1.client_utils import (
     get_api_endpoint,
-    get_client_cert_source,
     get_default_mtls_endpoint,
     get_universe_domain,
-    read_environment_variables,
-    use_client_cert_effective,
 )
 
 
@@ -205,147 +202,4 @@ def test__get_universe_domain():
     assert str(excinfo.value) == "Universe Domain cannot be an empty string."
 
 
-@pytest.mark.parametrize(
-    "has_method, method_val, env_val, expected",
-    [
-        # should_use_client_cert is available cases
-        (True, True, None, "true"),
-        (True, False, None, "false"),
-        (True, False, "unsupported", "false"),
-        # should_use_client_cert is unavailable cases
-        (False, None, "true", "true"),
-        (False, None, "false", "false"),
-        (False, None, "True", "true"),
-        (False, None, "False", "false"),
-        (False, None, "TRUE", "true"),
-        (False, None, "FALSE", "false"),
-        (False, None, None, "false"),
-        (False, None, "unsupported", "value_error"),
-    ],
-    ids=[
-        "google_auth_true",
-        "google_auth_false",
-        "google_auth_false_env_unsupported",
-        "fallback_env_true_lowercase",
-        "fallback_env_false_lowercase",
-        "fallback_env_true_titlecase",
-        "fallback_env_false_titlecase",
-        "fallback_env_true_uppercase",
-        "fallback_env_false_uppercase",
-        "fallback_env_unset",
-        "fallback_env_unsupported",
-    ],
-)
-def test_use_client_cert_effective(has_method, method_val, env_val, expected):
-    # Mock hasattr to control whether should_use_client_cert exists
-    original_hasattr = hasattr
-
-    def custom_hasattr(obj, name):
-        if obj is mtls and name == "should_use_client_cert":
-            return has_method
-        return original_hasattr(obj, name)
-
-    with mock.patch("google.api_core.gapic_v1.client_utils.hasattr", custom_hasattr):
-        with mock.patch(
-            "google.auth.transport.mtls.should_use_client_cert",
-            create=True,
-            return_value=method_val,
-        ):
-            env = {}
-            if env_val is not None:
-                env["GOOGLE_API_USE_CLIENT_CERTIFICATE"] = env_val
-            with mock.patch.dict(os.environ, env, clear=True):
-                if expected == "value_error":
-                    with pytest.raises(
-                        ValueError, match="must be either `true` or `false`"
-                    ):
-                        use_client_cert_effective()
-                else:
-                    assert use_client_cert_effective() is (expected == "true")
-
-
-@pytest.mark.parametrize(
-    "provided, use_cert, has_default_avail, default_val, expected",
-    [
-        (None, False, True, b"default", None),
-        (b"provided", False, True, b"default", None),
-        (b"provided", True, True, b"default", b"provided"),
-        (None, True, True, b"default", b"default"),
-        (None, True, False, b"default", "value_error"),
-    ],
-    ids=[
-        "use_cert_false_no_provided",
-        "use_cert_false_with_provided",
-        "use_cert_true_with_provided",
-        "use_cert_true_no_provided_default_avail",
-        "use_cert_true_no_provided_default_unavail",
-    ],
-)
-def test_get_client_cert_source(
-    provided, use_cert, has_default_avail, default_val, expected
-):
-    original_hasattr = hasattr
-
-    def custom_hasattr(obj, name):
-        if obj is mtls and name == "has_default_client_cert_source":
-            return has_default_avail
-        return original_hasattr(obj, name)
-
-    with mock.patch("google.api_core.gapic_v1.client_utils.hasattr", custom_hasattr):
-        with mock.patch(
-            "google.auth.transport.mtls.has_default_client_cert_source",
-            create=True,
-            return_value=has_default_avail,
-        ):
-            with mock.patch(
-                "google.auth.transport.mtls.default_client_cert_source",
-                create=True,
-                return_value=default_val,
-            ):
-                if expected == "value_error":
-                    with pytest.raises(
-                        ValueError, match="Client certificate is required for mTLS"
-                    ):
-                        get_client_cert_source(provided, use_cert)
-                else:
-                    assert get_client_cert_source(provided, use_cert) == expected
 
-
-@pytest.mark.parametrize(
-    "env, mock_cert_val, expected",
-    [
-        ({}, False, (False, "auto", None)),
-        ({}, True, (True, "auto", None)),
-        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}, False, (False, "never", None)),
-        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}, False, (False, "always", None)),
-        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}, False, (False, "auto", None)),
-        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "invalid"}, False, "mutual_tls_error"),
-        (
-            {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": "foo.com"},
-            False,
-            (False, "auto", "foo.com"),
-        ),
-    ],
-    ids=[
-        "default_env",
-        "client_cert_true",
-        "mtls_never",
-        "mtls_always",
-        "mtls_auto",
-        "mtls_invalid",
-        "universe_domain",
-    ],
-)
-def test_read_environment_variables(env, mock_cert_val, expected):
-    with mock.patch(
-        "google.api_core.gapic_v1.client_utils.use_client_cert_effective",
-        return_value=mock_cert_val,
-    ):
-        with mock.patch.dict(os.environ, env, clear=True):
-            if expected == "mutual_tls_error":
-                with pytest.raises(
-                    MutualTLSChannelError, match="must be `never`, `auto` or `always`"
-                ):
-                    read_environment_variables()
-            else:
-                assert read_environment_variables() == expected