diff --git a/packages/google-api-core/google/api_core/gapic_v1/__init__.py b/packages/google-api-core/google/api_core/gapic_v1/__init__.py
index e5b7ad352ee6..4fb3e7aeb2df 100644
--- a/packages/google-api-core/google/api_core/gapic_v1/__init__.py
+++ b/packages/google-api-core/google/api_core/gapic_v1/__init__.py
@@ -12,12 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from google.api_core.gapic_v1 import client_info
-from google.api_core.gapic_v1 import config
-from google.api_core.gapic_v1 import config_async
-from google.api_core.gapic_v1 import method
-from google.api_core.gapic_v1 import method_async
-from google.api_core.gapic_v1 import routing_header
+from google.api_core.gapic_v1 import (
+    client_info,
+    config,
+    config_async,
+    method,
+    method_async,
+    routing_header,
+)
 
 __all__ = [
     "client_info",
diff --git a/packages/google-api-core/google/api_core/gapic_v1/_client_cert.py b/packages/google-api-core/google/api_core/gapic_v1/_client_cert.py
new file mode 100644
index 000000000000..92b7ec5f5b65
--- /dev/null
+++ b/packages/google-api-core/google/api_core/gapic_v1/_client_cert.py
@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for client certificate handling and mTLS authentication."""
+
+import os
+from typing import Callable, Optional, Tuple
+
+from google.auth.transport import mtls  # type: ignore
+
+
+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.
+    """
+    client_cert_source = None
+    if use_cert_flag:
+        if provided_cert_source:
+            client_cert_source = provided_cert_source
+        elif (
+            hasattr(mtls, "has_default_client_cert_source")
+            and mtls.has_default_client_cert_source()
+        ):
+            client_cert_source = mtls.default_client_cert_source()
+    return client_cert_source
diff --git a/packages/google-api-core/google/api_core/gapic_v1/_config_helpers.py b/packages/google-api-core/google/api_core/gapic_v1/_config_helpers.py
new file mode 100644
index 000000000000..85be1daf1fda
--- /dev/null
+++ b/packages/google-api-core/google/api_core/gapic_v1/_config_helpers.py
@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for parsing environment variables."""
+
+import os
+from typing import Optional, Tuple
+
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+
+from google.api_core.gapic_v1._client_cert import _use_client_cert_effective
+
+
+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
diff --git a/packages/google-api-core/google/api_core/gapic_v1/_method_helpers.py b/packages/google-api-core/google/api_core/gapic_v1/_method_helpers.py
new file mode 100644
index 000000000000..3f8fd91874db
--- /dev/null
+++ b/packages/google-api-core/google/api_core/gapic_v1/_method_helpers.py
@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for method requests."""
+
+import uuid
+from typing import Any
+
+
+def _setup_request_id(request: Any, field_name: str, is_proto3_optional: bool) -> None:
+    """Populate a UUID4 field in the request if it is not already set.
+
+    Args:
+        request (Union[google.protobuf.message.Message, dict]): The
+            request object.
+        field_name (str): The name of the field to populate.
+        is_proto3_optional (bool): Whether the field is proto3 optional.
+    """
+    if isinstance(request, dict):
+        if is_proto3_optional:
+            if field_name not in request:
+                request[field_name] = str(uuid.uuid4())
+        elif not request.get(field_name):
+            request[field_name] = str(uuid.uuid4())
+        return
+
+    if is_proto3_optional:
+        try:
+            # Pure protobuf messages
+            if not request.HasField(field_name):
+                setattr(request, field_name, str(uuid.uuid4()))
+        except (AttributeError, ValueError):
+            # Proto-plus messages or other objects
+            if not getattr(request, field_name, None):
+                setattr(request, field_name, str(uuid.uuid4()))
+    else:
+        if not getattr(request, field_name):
+            setattr(request, field_name, str(uuid.uuid4()))
diff --git a/packages/google-api-core/google/api_core/gapic_v1/_routing.py b/packages/google-api-core/google/api_core/gapic_v1/_routing.py
new file mode 100644
index 000000000000..56bc31ed83f0
--- /dev/null
+++ b/packages/google-api-core/google/api_core/gapic_v1/_routing.py
@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for routing and endpoint resolution."""
+
+import re
+from typing import Any, Optional
+
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+
+_MTLS_ENDPOINT_RE = re.compile(
+    r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?"
+    r"(?P<googledomain>\.googleapis\.com)?"
+)
+
+
+def _get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+    """Converts api endpoint to mTLS endpoint.
+
+    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+    Args:
+        api_endpoint (Optional[str]): the api endpoint to convert.
+    Returns:
+        Optional[str]: converted mTLS api endpoint.
+    """
+    if not api_endpoint:
+        return api_endpoint
+
+    m = _MTLS_ENDPOINT_RE.match(api_endpoint)
+    if m is None:
+        # Could not parse api_endpoint; return as-is.
+        return api_endpoint
+
+    name, mtls_group, sandbox, googledomain = m.groups()
+    if mtls_group or not googledomain:
+        return api_endpoint
+
+    if sandbox:
+        return api_endpoint.replace(
+            "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
+        )
+
+    return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
+
+
+def _get_api_endpoint(
+    api_override: Optional[str],
+    client_cert_source: Optional[Any],
+    universe_domain: str,
+    use_mtls_endpoint: str,
+    default_universe: str,
+    default_mtls_endpoint: Optional[str],
+    default_endpoint_template: Optional[str],
+) -> Optional[str]:
+    """Return the API endpoint used by the client."""
+    if api_override is not None:
+        return api_override
+    elif use_mtls_endpoint == "always" or (
+        use_mtls_endpoint == "auto" and client_cert_source
+    ):
+        if universe_domain.lower() != default_universe.lower():
+            raise MutualTLSChannelError(
+                f"mTLS is not supported in any universe other than {default_universe}."
+            )
+        return default_mtls_endpoint
+    else:
+        return (
+            default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+            if default_endpoint_template
+            else None
+        )
+
+
+def _get_universe_domain(
+    client_universe_domain: Optional[str],
+    universe_domain_env: Optional[str],
+    default_universe: str,
+) -> str:
+    """Return the universe domain used by the client."""
+    universe_domain = default_universe
+    if client_universe_domain is not None:
+        universe_domain = client_universe_domain.strip()
+    elif universe_domain_env is not None:
+        universe_domain = universe_domain_env.strip()
+    if not universe_domain:
+        raise ValueError("Universe Domain cannot be an empty string.")
+    return universe_domain
diff --git a/packages/google-api-core/noxfile.py b/packages/google-api-core/noxfile.py
index 0bad668a80dd..3cbebbaa84d2 100644
--- a/packages/google-api-core/noxfile.py
+++ b/packages/google-api-core/noxfile.py
@@ -350,7 +350,7 @@ def prerelease_deps(session):
 @nox.session(python=DEFAULT_PYTHON_VERSION)
 def core_deps_from_source(session):
     """Run the test suite installing dependencies from source."""
-    default(session, prerelease=True)
+    default(session, install_deps_from_source=True)
 
 
 @nox.session(python=DEFAULT_PYTHON_VERSION)
diff --git a/packages/google-api-core/tests/conftest.py b/packages/google-api-core/tests/conftest.py
new file mode 100644
index 000000000000..1872207be107
--- /dev/null
+++ b/packages/google-api-core/tests/conftest.py
@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+import pytest
+
+
+@pytest.fixture(scope="session", autouse=True)
+def mock_mtls_env():
+    """Autouse session-scoped fixture to isolate unit tests from workstation mTLS environments."""
+    with mock.patch.dict(
+        os.environ,
+        {
+            "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false",
+            "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false",
+        },
+    ):
+        yield
diff --git a/packages/google-api-core/tests/unit/gapic/test_client_cert.py b/packages/google-api-core/tests/unit/gapic/test_client_cert.py
new file mode 100644
index 000000000000..c51689561809
--- /dev/null
+++ b/packages/google-api-core/tests/unit/gapic/test_client_cert.py
@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+
+import pytest
+
+# We need to skip this test module if grpc is not installed because importing
+# gapic_v1._client_cert will load gapic_v1/__init__, which unconditionally
+# imports gapic_v1.config, which imports grpc.
+try:
+    import grpc  # noqa: F401
+except ImportError:
+    pytest.skip("No GRPC", allow_module_level=True)
+
+from google.api_core.gapic_v1._client_cert import (
+    _get_client_cert_source,
+    _use_client_cert_effective,
+)
+
+
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", create=True)
+def test_use_client_cert_effective_with_google_auth(mock_method):
+    # Test when google-auth supports the method
+    mock_method.return_value = True
+    assert _use_client_cert_effective() is True
+
+    mock_method.return_value = False
+    assert _use_client_cert_effective() is False
+
+
+@mock.patch.dict(os.environ, {}, clear=True)
+def test_use_client_cert_effective_fallback():
+    # We must patch hasattr to simulate google-auth lacking the method
+    with mock.patch(
+        "google.api_core.gapic_v1._client_cert.hasattr", return_value=False
+    ):
+        # Default is false
+        assert _use_client_cert_effective() is False
+
+        env_true = {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        with mock.patch.dict(os.environ, env_true):
+            assert _use_client_cert_effective() is True
+
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}
+        ):
+            assert _use_client_cert_effective() is False
+
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "invalid"}
+        ):
+            match_str = "must be either `true` or `false`"
+            with pytest.raises(ValueError, match=match_str):
+                _use_client_cert_effective()
+
+
+@mock.patch(
+    "google.auth.transport.mtls.has_default_client_cert_source", create=True
+)  # noqa: E501
+@mock.patch(
+    "google.auth.transport.mtls.default_client_cert_source", create=True
+)  # noqa: E501
+def test_get_client_cert_source(mock_default, mock_has_default):
+    mock_default.return_value = b"default_cert"
+    mock_has_default.return_value = True
+
+    # When use_cert_flag is False, return None
+    assert _get_client_cert_source(b"provided", False) is None
+
+    # When provided_cert_source is given, return provided
+    assert _get_client_cert_source(b"provided", True) == b"provided"  # noqa: E501
+
+    # When no provided cert but default is available
+    assert _get_client_cert_source(None, True) == b"default_cert"
diff --git a/packages/google-api-core/tests/unit/gapic/test_config_helpers.py b/packages/google-api-core/tests/unit/gapic/test_config_helpers.py
new file mode 100644
index 000000000000..312ad9795756
--- /dev/null
+++ b/packages/google-api-core/tests/unit/gapic/test_config_helpers.py
@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+
+import pytest
+
+# We need to skip this test module if grpc is not installed because importing
+# gapic_v1._config_helpers will load gapic_v1/__init__, which unconditionally
+# imports gapic_v1.config, which imports grpc.
+try:
+    import grpc  # noqa: F401
+except ImportError:
+    pytest.skip("No GRPC", allow_module_level=True)
+
+from google.auth.exceptions import MutualTLSChannelError
+
+from google.api_core.gapic_v1._config_helpers import _read_environment_variables
+
+
+@mock.patch(
+    "google.api_core.gapic_v1._config_helpers._use_client_cert_effective"
+)  # noqa: E501
+@mock.patch.dict(os.environ, clear=True)
+def test_read_environment_variables(mock_effective):
+    mock_effective.return_value = True
+    os.environ["GOOGLE_API_USE_MTLS_ENDPOINT"] = "always"
+    os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] = "custom.com"
+
+    cert, mtls, domain = _read_environment_variables()
+    assert cert is True
+    assert mtls == "always"
+    assert domain == "custom.com"
+
+
+@mock.patch.dict(os.environ, clear=True)
+def test_read_environment_variables_invalid_mtls():
+    os.environ["GOOGLE_API_USE_MTLS_ENDPOINT"] = "invalid"
+    with pytest.raises(
+        MutualTLSChannelError, match="must be `never`, `auto` or `always`"
+    ):
+        _read_environment_variables()
diff --git a/packages/google-api-core/tests/unit/gapic/test_method_helpers.py b/packages/google-api-core/tests/unit/gapic/test_method_helpers.py
new file mode 100644
index 000000000000..970016574f69
--- /dev/null
+++ b/packages/google-api-core/tests/unit/gapic/test_method_helpers.py
@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import pytest
+
+# We need to skip this test module if grpc is not installed because importing
+# gapic_v1._method_helpers will load gapic_v1/__init__, which unconditionally
+# imports gapic_v1.config, which imports grpc.
+try:
+    import grpc  # noqa: F401
+except ImportError:
+    pytest.skip("No GRPC", allow_module_level=True)
+
+
+from google.api_core.gapic_v1._method_helpers import _setup_request_id
+
+
+def test_setup_request_id():
+    import uuid
+
+    # test dict request
+    req = {}
+    _setup_request_id(req, "request_id", True)
+    assert "request_id" in req
+    uuid_str = req["request_id"]
+    uuid.UUID(uuid_str)  # verify it is a valid UUID
+
+    # test dict request when already set
+    req = {"request_id": "existing"}
+    _setup_request_id(req, "request_id", True)
+    assert req["request_id"] == "existing"
+
+    class DummyRequest:
+        def __init__(self):
+            self.request_id = ""
+
+        def HasField(self, field_name):
+            if not hasattr(self, field_name):
+                raise ValueError()
+            return bool(getattr(self, field_name))
+
+    # test object request proto3 optional true
+    req_obj = DummyRequest()
+    _setup_request_id(req_obj, "request_id", True)
+    assert req_obj.request_id != ""
+    uuid.UUID(req_obj.request_id)
+
+    # test object request proto3 optional false
+    req_obj2 = DummyRequest()
+    _setup_request_id(req_obj2, "request_id", False)
+    assert req_obj2.request_id != ""
+    uuid.UUID(req_obj2.request_id)
+
+    class CustomRequestWrapper:
+        def __init__(self):
+            self.request_id = ""
+
+    # test custom non-iterable object wrapper
+    req_obj3 = CustomRequestWrapper()
+    _setup_request_id(req_obj3, "request_id", True)
+    assert req_obj3.request_id != ""
+    uuid.UUID(req_obj3.request_id)
diff --git a/packages/google-api-core/tests/unit/gapic/test_routing.py b/packages/google-api-core/tests/unit/gapic/test_routing.py
new file mode 100644
index 000000000000..a61d5e9f918e
--- /dev/null
+++ b/packages/google-api-core/tests/unit/gapic/test_routing.py
@@ -0,0 +1,185 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+import pytest
+
+# We need to skip this test module if grpc is not installed because importing
+# gapic_v1._routing will load gapic_v1/__init__, which unconditionally
+# imports gapic_v1.config, which imports grpc.
+try:
+    import grpc  # noqa: F401
+except ImportError:
+    pytest.skip("No GRPC", allow_module_level=True)
+
+from google.auth.exceptions import MutualTLSChannelError
+
+from google.api_core.gapic_v1._routing import (
+    _get_api_endpoint,
+    _get_default_mtls_endpoint,
+    _get_universe_domain,
+)
+
+
+def test_get_default_mtls_endpoint():
+    # Test valid API endpoints
+    assert _get_default_mtls_endpoint("foo.googleapis.com") == "foo.mtls.googleapis.com"
+    assert (
+        _get_default_mtls_endpoint("foo.sandbox.googleapis.com")
+        == "foo.mtls.sandbox.googleapis.com"
+    )
+
+    # Test endpoints that shouldn't be converted
+    assert (
+        _get_default_mtls_endpoint("foo.mtls.googleapis.com")
+        == "foo.mtls.googleapis.com"
+    )
+    assert _get_default_mtls_endpoint("foo.com") == "foo.com"
+
+    # Test empty/None endpoints
+    assert _get_default_mtls_endpoint("") == ""
+    assert _get_default_mtls_endpoint(None) is None
+
+
+def test_get_api_endpoint_override():
+    # If api_override is provided, it should be returned
+    # regardless of other args
+    endpoint = _get_api_endpoint(
+        api_override="custom.endpoint.com",
+        client_cert_source=None,
+        universe_domain="googleapis.com",
+        use_mtls_endpoint="auto",
+        default_universe="googleapis.com",
+        default_mtls_endpoint="foo.mtls.googleapis.com",
+        default_endpoint_template="foo.{UNIVERSE_DOMAIN}",
+    )
+    assert endpoint == "custom.endpoint.com"
+
+
+def test_get_api_endpoint_mtls_always():
+    # use_mtls_endpoint == "always" should use the default mtls endpoint
+    endpoint = _get_api_endpoint(
+        api_override=None,
+        client_cert_source=None,
+        universe_domain="googleapis.com",
+        use_mtls_endpoint="always",
+        default_universe="googleapis.com",
+        default_mtls_endpoint="foo.mtls.googleapis.com",
+        default_endpoint_template="foo.{UNIVERSE_DOMAIN}",
+    )
+    assert endpoint == "foo.mtls.googleapis.com"
+
+
+def test_get_api_endpoint_mtls_auto_with_cert():
+    # "auto" with client_cert_source should use mtls
+    endpoint = _get_api_endpoint(
+        api_override=None,
+        client_cert_source=mock.Mock(),
+        universe_domain="googleapis.com",
+        use_mtls_endpoint="auto",
+        default_universe="googleapis.com",
+        default_mtls_endpoint="foo.mtls.googleapis.com",
+        default_endpoint_template="foo.{UNIVERSE_DOMAIN}",
+    )
+    assert endpoint == "foo.mtls.googleapis.com"
+
+
+def test_get_api_endpoint_mtls_auto_no_cert():
+    # "auto" without client_cert_source should use the default template
+    endpoint = _get_api_endpoint(
+        api_override=None,
+        client_cert_source=None,
+        universe_domain="googleapis.com",
+        use_mtls_endpoint="auto",
+        default_universe="googleapis.com",
+        default_mtls_endpoint="foo.mtls.googleapis.com",
+        default_endpoint_template="foo.{UNIVERSE_DOMAIN}",
+    )
+    assert endpoint == "foo.googleapis.com"
+
+
+def test_get_api_endpoint_mtls_universe_mismatch():
+    # mTLS is only supported in the default universe
+    with pytest.raises(MutualTLSChannelError, match="mTLS is not supported"):
+        _get_api_endpoint(
+            api_override=None,
+            client_cert_source=mock.Mock(),
+            universe_domain="custom-universe.com",
+            use_mtls_endpoint="auto",
+            default_universe="googleapis.com",
+            default_mtls_endpoint="foo.mtls.googleapis.com",
+            default_endpoint_template="foo.{UNIVERSE_DOMAIN}",
+        )
+
+
+def test_get_api_endpoint_mtls_case_insensitive():
+    # mTLS universe check should be case insensitive
+    endpoint = _get_api_endpoint(
+        api_override=None,
+        client_cert_source=mock.Mock(),
+        universe_domain="GOOGLEAPIS.COM",
+        use_mtls_endpoint="auto",
+        default_universe="googleapis.com",
+        default_mtls_endpoint="foo.mtls.googleapis.com",
+        default_endpoint_template="foo.{UNIVERSE_DOMAIN}",
+    )
+    assert endpoint == "foo.mtls.googleapis.com"
+
+
+def test_get_universe_domain():
+    # client_universe_domain takes precedence
+    assert (
+        _get_universe_domain("client.com", "env.com", "default.com")  # noqa: E501
+        == "client.com"
+    )
+
+    # env takes precedence over default
+    assert (
+        _get_universe_domain(None, "env.com", "default.com") == "env.com"  # noqa: E501
+    )
+
+    # fallback to default
+    assert (
+        _get_universe_domain(None, None, "default.com") == "default.com"
+    )  # noqa: E501
+
+
+def test_get_universe_domain_strip():
+    # check that whitespace is stripped
+    assert (
+        _get_universe_domain("  client.com  ", "env.com", "default.com") == "client.com"
+    )
+    assert _get_universe_domain(None, "  env.com  ", "default.com") == "env.com"
+
+
+def test_get_universe_domain_empty():
+    with pytest.raises(ValueError, match="cannot be an empty string"):
+        _get_universe_domain("", None, "default.com")
+    with pytest.raises(ValueError, match="cannot be an empty string"):
+        _get_universe_domain("   ", None, "default.com")
+
+
+def test_get_api_endpoint_none_template():
+    endpoint = _get_api_endpoint(
+        api_override=None,
+        client_cert_source=None,
+        universe_domain="googleapis.com",
+        use_mtls_endpoint="never",
+        default_universe="googleapis.com",
+        default_mtls_endpoint=None,
+        default_endpoint_template=None,
+    )
+    assert endpoint is None
diff --git a/packages/google-api-core/tests/unit/test_bidi.py b/packages/google-api-core/tests/unit/test_bidi.py
index 4a8eb74fac94..0f4810cb0a12 100644
--- a/packages/google-api-core/tests/unit/test_bidi.py
+++ b/packages/google-api-core/tests/unit/test_bidi.py
@@ -31,8 +31,7 @@
 except ImportError:  # pragma: NO COVER
     pytest.skip("No GRPC", allow_module_level=True)
 
-from google.api_core import bidi
-from google.api_core import exceptions
+from google.api_core import bidi, exceptions
 
 
 class Test_RequestQueueGenerator(object):
@@ -195,7 +194,7 @@ def test_delays_entry_attempts_above_threshold(self):
         # (NOTE: not using assert all(...), b/c the coverage check would complain)
         for i, entry in enumerate(entries):
             if i != 3:
-                assert entry["reported_wait"] == 0.0
+                assert entry["reported_wait"] < 0.01
 
         # The delayed entry is expected to have been delayed for a significant
         # chunk of the full second, and the actual and reported delay times