refactor(api-core): move universe routing helpers to universe.py

Author: hebaalazzeh

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

@@ -18,6 +18,14 @@
 
 from google.auth.exceptions import MutualTLSChannelError  # type: ignore
 
+from google.api_core.universe import get_universe_domain
+
+__all__ = [
+    "get_default_mtls_endpoint",
+    "get_api_endpoint",
+    "get_universe_domain",
+]
+
 
 def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
     """Converts api endpoint to mTLS endpoint.
@@ -106,31 +114,3 @@ def get_api_endpoint(
         return default_mtls_endpoint
     else:
         return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = "googleapis.com",
-) -> 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:
-        ValueError: 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 ValueError("Universe Domain cannot be an empty string.")
-    return resolved
-
-

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

@@ -36,6 +36,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 +78,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:

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

@@ -12,18 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-from unittest import mock
 import pytest
 from google.auth.exceptions import MutualTLSChannelError
-from google.auth.transport import mtls
 
 from google.api_core.gapic_v1.client_utils import (
     get_api_endpoint,
     get_default_mtls_endpoint,
-    get_universe_domain,
 )
-from google.auth.exceptions import MutualTLSChannelError
 
 
 def test_get_default_mtls_endpoint():
@@ -164,43 +159,3 @@ def test_get_api_endpoint(
             )
             == expected
         )
-
-
-def test_get_universe_domain():
-    # When universe_domain is provided
-    assert get_universe_domain("foo.com", default_universe="default.com") == "foo.com"
-    assert (
-        get_universe_domain("  foo.com  ", default_universe="default.com") == "foo.com"
-    )
-
-    # When universe_domain is None, falls back to default_universe
-    assert get_universe_domain(None, default_universe="default.com") == "default.com"
-
-    # When multiple potential universes are provided, resolves in order of preference
-    assert (
-        get_universe_domain("foo.com", "bar.com", default_universe="default.com")
-        == "foo.com"
-    )
-    assert (
-        get_universe_domain(None, "bar.com", default_universe="default.com")
-        == "bar.com"
-    )
-    assert (
-        get_universe_domain(None, None, default_universe="default.com") == "default.com"
-    )
-
-    # ValueError raised when resolved value is empty string
-    with pytest.raises(ValueError) as excinfo:
-        get_universe_domain("", default_universe="default.com")
-    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
-
-    with pytest.raises(ValueError) as excinfo:
-        get_universe_domain("   ", default_universe="default.com")
-    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
-
-    with pytest.raises(ValueError) as excinfo:
-        get_universe_domain(None, "", default_universe="default.com")
-    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
-
-
-

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

@@ -62,3 +62,50 @@ 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."