feat(api-core): add get_universe_domain helper to universe.py (#17799)

## Description

This PR consolidates and centralizes all universe-related configuration
helpers and routing logic into `google/api_core/universe.py`. This
avoids polluting `gapic_v1` with duplicate routing logic and keeps
universe/endpoint utilities unified.

### Changes:
1. **get_universe_domain**:
- Added a more general version of `get_universe_domain` that accepts
`*potential_universes` in order of preference.
- Made `default_universe` a required keyword-only argument to align with
other client helpers.
2. **determine_domain**:
- Refactored `determine_domain` to wrap `get_universe_domain` to
preserve backward compatibility.
3. **Endpoint & mTLS Helpers**:
- Moved `get_api_endpoint` and `get_default_mtls_endpoint` from the
`gapic_v1.client_utils` namespace into `google/api_core/universe.py`.
4. **Unit Tests**:
   - Migrated all related test cases to `tests/unit/test_universe.py`.

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: hebaalazzeh

packages/google-api-core/google/api_core/universe.py

@@ -15,6 +15,9 @@
 """Helpers for universe domain."""
 
 from typing import Any, Optional
+from urllib.parse import urlparse, urlunparse
+
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
 
 DEFAULT_UNIVERSE = "googleapis.com"
 
@@ -36,6 +39,32 @@ def __init__(self, client_universe, credentials_universe):
         super().__init__(message)
 
 
+def get_universe_domain(
+    *potential_universes: Optional[str],
+    default_universe: str,
+) -> str:
+    """Return the universe domain used by the client.
+
+    Args:
+        *potential_universes (Optional[str]): Potential universe domains in order of preference.
+        default_universe (str): The default universe domain.
+
+    Returns:
+        str: The universe domain to be used by the client.
+
+    Raises:
+        EmptyUniverseError: If the resolved universe domain is an empty string.
+    """
+    resolved = next(
+        (x.strip() for x in potential_universes if x is not None),
+        default_universe,
+    )
+
+    if not resolved:
+        raise EmptyUniverseError()
+    return resolved
+
+
 def determine_domain(
     client_universe_domain: Optional[str], universe_domain_env: Optional[str]
 ) -> str:
@@ -52,14 +81,11 @@ def determine_domain(
     Raises:
         ValueError: If the universe domain is an empty string.
     """
-    universe_domain = DEFAULT_UNIVERSE
-    if client_universe_domain is not None:
-        universe_domain = client_universe_domain
-    elif universe_domain_env is not None:
-        universe_domain = universe_domain_env
-    if len(universe_domain.strip()) == 0:
-        raise EmptyUniverseError
-    return universe_domain
+    return get_universe_domain(
+        client_universe_domain,
+        universe_domain_env,
+        default_universe=DEFAULT_UNIVERSE,
+    )
 
 
 def compare_domains(client_universe: str, credentials: Any) -> bool:
@@ -80,3 +106,94 @@ def compare_domains(client_universe: str, credentials: Any) -> bool:
     if client_universe != credentials_universe:
         raise UniverseMismatchError(client_universe, credentials_universe)
     return True
+
+
+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.
+    Other URLs (including those that do not match these domain suffixes or
+    already contain '.mtls.') are passed through as-is.
+
+    Args:
+        api_endpoint (Optional[str]): the api endpoint to convert.
+
+    Returns:
+        Optional[str]: converted mTLS api endpoint.
+    """
+    if not api_endpoint or ".mtls." in api_endpoint.lower():
+        return api_endpoint
+
+    has_scheme = "://" in api_endpoint
+    if not has_scheme:
+        parsed = urlparse("//" + api_endpoint)
+    else:
+        parsed = urlparse(api_endpoint)
+
+    host = parsed.hostname
+    if not host:
+        return api_endpoint
+
+    port = f":{parsed.port}" if parsed.port else ""
+
+    lowered_host = host.lower()
+    suffix_sandbox = ".sandbox.googleapis.com"
+    suffix_google = ".googleapis.com"
+    if lowered_host.endswith(suffix_sandbox):
+        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+    elif lowered_host.endswith(suffix_google):
+        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+    else:
+        return api_endpoint
+
+    netloc = new_host + port
+    new_parsed = parsed._replace(netloc=netloc)
+
+    if not has_scheme:
+        return urlunparse(new_parsed)[2:]
+    else:
+        return urlunparse(new_parsed)
+
+
+def get_api_endpoint(
+    api_override: Optional[str],
+    universe_domain: str,
+    default_universe: str,
+    default_mtls_endpoint: Optional[str],
+    default_endpoint_template: str,
+    use_mtls: bool,
+) -> str:
+    """Return the API endpoint used by the client.
+
+    Args:
+        api_override (Optional[str]): The API endpoint override. If specified,
+            this is always returned.
+        universe_domain (str): The universe domain used by the client.
+        default_universe (str): The default universe domain.
+        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+        default_endpoint_template (str): The default endpoint template containing
+            a placeholder `{UNIVERSE_DOMAIN}`.
+        use_mtls (bool): Whether to use the mTLS endpoint.
+
+    Returns:
+        str: The API endpoint to be used by the client.
+
+    Raises:
+        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+            not supported in the configured universe domain.
+        ValueError: If mTLS is requested but no mTLS endpoint is available.
+    """
+    if api_override is not None:
+        return api_override
+
+    if use_mtls:
+        if universe_domain.lower() != default_universe.lower():
+            raise MutualTLSChannelError(
+                f"mTLS is not supported in any universe other than {default_universe}."
+            )
+        if not default_mtls_endpoint:
+            raise ValueError("mTLS endpoint is not available.")
+        return default_mtls_endpoint
+    else:
+        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)

packages/google-api-core/tests/unit/test_universe.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import pytest
+from google.auth.exceptions import MutualTLSChannelError
 
 from google.api_core import universe
 
@@ -62,3 +63,199 @@ def test_compare_domains():
         universe.compare_domains(fake_domain, _Fake_Credentials(another_fake_domain))
     assert str(excinfo.value).find(fake_domain) >= 0
     assert str(excinfo.value).find(another_fake_domain) >= 0
+
+
+def test_get_universe_domain():
+    # When universe_domain is provided
+    assert (
+        universe.get_universe_domain("foo.com", default_universe="default.com")
+        == "foo.com"
+    )
+    assert (
+        universe.get_universe_domain("  foo.com  ", default_universe="default.com")
+        == "foo.com"
+    )
+
+    # When universe_domain is None, falls back to default_universe
+    assert (
+        universe.get_universe_domain(None, default_universe="default.com")
+        == "default.com"
+    )
+
+    # When multiple potential universes are provided, resolves in order of preference
+    assert (
+        universe.get_universe_domain(
+            "foo.com", "bar.com", default_universe="default.com"
+        )
+        == "foo.com"
+    )
+    assert (
+        universe.get_universe_domain(None, "bar.com", default_universe="default.com")
+        == "bar.com"
+    )
+    assert (
+        universe.get_universe_domain(None, None, default_universe="default.com")
+        == "default.com"
+    )
+
+    # EmptyUniverseError raised when resolved value is empty string
+    with pytest.raises(universe.EmptyUniverseError) as excinfo:
+        universe.get_universe_domain("", default_universe="default.com")
+    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
+
+    with pytest.raises(universe.EmptyUniverseError) as excinfo:
+        universe.get_universe_domain("   ", default_universe="default.com")
+    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
+
+    with pytest.raises(universe.EmptyUniverseError) as excinfo:
+        universe.get_universe_domain(None, "", default_universe="default.com")
+    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
+
+
+def test_get_default_mtls_endpoint():
+    # Test valid API endpoints
+    assert (
+        universe.get_default_mtls_endpoint("foo.googleapis.com")
+        == "foo.mtls.googleapis.com"
+    )
+    assert (
+        universe.get_default_mtls_endpoint("foo.sandbox.googleapis.com")
+        == "foo.mtls.sandbox.googleapis.com"
+    )
+    # Test case-insensitivity
+    assert (
+        universe.get_default_mtls_endpoint("foo.GoogleAPIs.com")
+        == "foo.mtls.googleapis.com"
+    )
+    assert (
+        universe.get_default_mtls_endpoint("foo.Sandbox.GoogleAPIs.com")
+        == "foo.mtls.sandbox.googleapis.com"
+    )
+
+    # Test valid API endpoints with schemes
+    assert (
+        universe.get_default_mtls_endpoint("https://foo.googleapis.com")
+        == "https://foo.mtls.googleapis.com"
+    )
+    assert (
+        universe.get_default_mtls_endpoint("http://foo.googleapis.com:8080/v1")
+        == "http://foo.mtls.googleapis.com:8080/v1"
+    )
+
+    # Test valid API endpoints with ports
+    assert (
+        universe.get_default_mtls_endpoint("foo.googleapis.com:443")
+        == "foo.mtls.googleapis.com:443"
+    )
+    assert (
+        universe.get_default_mtls_endpoint("foo.sandbox.googleapis.com:443")
+        == "foo.mtls.sandbox.googleapis.com:443"
+    )
+    # Test case-insensitivity with ports
+    assert (
+        universe.get_default_mtls_endpoint("foo.GoogleAPIs.com:443")
+        == "foo.mtls.googleapis.com:443"
+    )
+    assert (
+        universe.get_default_mtls_endpoint("foo.Sandbox.GoogleAPIs.com:443")
+        == "foo.mtls.sandbox.googleapis.com:443"
+    )
+
+    # Test endpoints that shouldn't be converted
+    assert (
+        universe.get_default_mtls_endpoint("foo.mtls.googleapis.com")
+        == "foo.mtls.googleapis.com"
+    )
+    assert universe.get_default_mtls_endpoint("foo.com") == "foo.com"
+    assert universe.get_default_mtls_endpoint("foo.com:8080") == "foo.com:8080"
+
+    # Test empty/None endpoints
+    assert universe.get_default_mtls_endpoint("") == ""
+    assert universe.get_default_mtls_endpoint(None) is None
+
+    # Test endpoints without host
+    assert universe.get_default_mtls_endpoint("http://") == "http://"
+    assert universe.get_default_mtls_endpoint("https://") == "https://"
+
+
+@pytest.mark.parametrize(
+    "api_override,universe_domain,default_universe,default_mtls_endpoint,default_endpoint_template,use_mtls,expected",
+    [
+        (
+            "foo.com",
+            "googleapis.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            "foo.com",
+        ),
+        (
+            None,
+            "googleapis.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            "foo.mtls.googleapis.com",
+        ),
+        (
+            None,
+            "googleapis.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            False,
+            "foo.googleapis.com",
+        ),
+        (
+            None,
+            "bar.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            MutualTLSChannelError,
+        ),
+        (
+            None,
+            "googleapis.com",
+            "googleapis.com",
+            None,
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            ValueError,
+        ),
+    ],
+)
+def test_get_api_endpoint(
+    api_override,
+    universe_domain,
+    default_universe,
+    default_mtls_endpoint,
+    default_endpoint_template,
+    use_mtls,
+    expected,
+):
+    if isinstance(expected, type) and issubclass(expected, Exception):
+        with pytest.raises(expected):
+            universe.get_api_endpoint(
+                api_override,
+                universe_domain,
+                default_universe,
+                default_mtls_endpoint,
+                default_endpoint_template,
+                use_mtls,
+            )
+    else:
+        assert (
+            universe.get_api_endpoint(
+                api_override,
+                universe_domain,
+                default_universe,
+                default_mtls_endpoint,
+                default_endpoint_template,
+                use_mtls,
+            )
+            == expected
+        )