Repository: googleapis/google-api-python-client (python)

Issue:
Add tests for validating credentials for the configured universe once leveraged.
```python3
def test_validate_credentials_with_no_universe(self):
            # Add test case for no configured  universe and fake credentials' universe.
            http = google_auth_httplib2.AuthorizedHttp(credentials=None, http=build_http())
            discovery = read_datafile("zoo.json")
            service = build_from_document(discovery, http=http, client_options=google.api_core.client_options.ClientOptions(universe_domain=fake_universe))

            with self.assertRaises(universe.UniverseMismatchError):
                service._validate_credentials()
                
             # Add test case for not specifying universe domain via client option.


def test_validate_credentials_with_default_universe(self):
            # Add test case for "googleapis.com" configured universe and fake credentials' universe.
            fake_universe = "foo.com"
            http = google_auth_httplib2.AuthorizedHttp(credentials=mock.Mock(universe_domain=universe.DEFAULT_UNIVERSE), http=build_http())
            discovery = read_datafile("zoo.json")
            service = build_from_document(discovery, http=http, client_options=google.api_core.client_options.ClientOptions(universe_domain=fake_universe))

            with self.assertRaises(universe.UniverseMismatchError):
                service._validate_credentials()

def test_validate_credentials_with_a_different_universe(self):
            # Add test case for fake configured universe and fake credentials' universe.
            http = google_auth_httplib2.AuthorizedHttp(credentials=mock.Mock(universe_domain=fake_universe), http=build_http())
            discovery = read_datafile("zoo.json")
            service = build_from_document(discovery, http=http, client_options=google.api_core.client_options.ClientOptions(universe_domain=fake_universe))
            assert service._validate_credentials()

def test_validate_credentials_with_already_validated_credentials(self):
           # Add test case for fake configured universe and fake credentials' universe.
            http = google_auth_httplib2.AuthorizedHttp(credentials=mock.Mock(universe_domain=fake_universe), http=build_http())
            discovery = read_datafile("zoo.json")
            service = build_from_document(discovery, http=http, client_options=google.api_core.client_options.ClientOptions(universe_domain=fake_universe))

            assert service._validate_credentials()
            assert service._credentials_validated
            assert service._validate_credentials()
            
        # TODO(omairn):
        # Once _validate_universe() is added to the API call, verify that:
        # the API call succeeds when the credentials match.
        # the API call fails when there is a mismatch.
        # the credentials are not verified twice for an already verified Resource.

        # Add test case once universe domain is supported via environment variable.

        # Once universe.determine_domain() is leveraged:
        # add test case for empty universe.
        # add test cases when universe domain is provided via both client option and env var.

        # Add test case when credentials are passed instead of http.
```



Pull request description:
feat: Allow users to configure client option `universe_domain`
Fixes #2365 🦕
Towards b/324211188 🦕


Patch:
diff --git a/googleapiclient/discovery.py b/googleapiclient/discovery.py
index eb01be65a..f7b3f0041 100644
--- a/googleapiclient/discovery.py
+++ b/googleapiclient/discovery.py
@@ -551,6 +551,13 @@ def build_from_document(
 
     # If an API Endpoint is provided on client options, use that as the base URL
     base = urllib.parse.urljoin(service["rootUrl"], service["servicePath"])
+    universe_domain = None
+    if HAS_UNIVERSE:
+        universe_domain = universe.determine_domain(
+            client_options.universe_domain, None
+        )
+        base = base.replace(universe.DEFAULT_UNIVERSE, universe_domain)
+
     audience_for_self_signed_jwt = base
     if client_options.api_endpoint:
         base = client_options.api_endpoint
@@ -673,6 +680,10 @@ def build_from_document(
             if use_mtls_endpoint == "always" or (
                 use_mtls_endpoint == "auto" and client_cert_to_use
             ):
+                if HAS_UNIVERSE and universe_domain != universe.DEFAULT_UNIVERSE:
+                    raise MutualTLSChannelError(
+                        f"mTLS is not supported in any universe other than {universe.DEFAULT_UNIVERSE}."
+                    )
                 base = mtls_endpoint
 
     if model is None:
@@ -688,6 +699,7 @@ def build_from_document(
         resourceDesc=service,
         rootDesc=service,
         schema=schema,
+        universe_domain=universe_domain,
     )
 
 
@@ -1517,6 +1529,7 @@ class Resource(object):
                         resourceDesc=methodDesc,
                         rootDesc=rootDesc,
                         schema=schema,
+                        universe_domain=self._universe_domain,
                     )
 
                 setattr(methodResource, "__doc__", "A collection resource.")


Test patch:
diff --git a/tests/test_discovery.py b/tests/test_discovery.py
index 635d2acd5..2b888bce0 100644
--- a/tests/test_discovery.py
+++ b/tests/test_discovery.py
@@ -2343,6 +2343,25 @@ class MediaGet(unittest.TestCase):
 if HAS_UNIVERSE:
 
     class Universe(unittest.TestCase):
+        def test_validate_credentials_with_no_client_options(self):
+            http = build_http()
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+            )
+            assert service._validate_credentials()
+
+        def test_validate_credentials_with_client_options_without_universe(self):
+            http = build_http()
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(),
+            )
+            assert service._validate_credentials()
+
         def test_validate_credentials_with_no_universe(self):
             fake_universe = "foo.com"
 
@@ -2360,9 +2379,20 @@ if HAS_UNIVERSE:
 
             assert service._validate_credentials()
 
-            # TODO(google-api-python-client/issues/2365): Add test case for no configured  universe and fake credentials' universe.
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=None, http=build_http()
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=fake_universe
+                ),
+            )
 
-            # TODO(google-api-python-client/issues/2365): Add test case for not specifying universe domain via client option.
+            with self.assertRaises(universe.UniverseMismatchError):
+                service._validate_credentials()
 
         def test_validate_credentials_with_default_universe(self):
             fake_universe = "foo.com"
@@ -2382,12 +2412,39 @@ if HAS_UNIVERSE:
 
             assert service._validate_credentials()
 
-            # TODO(google-api-python-client/issues/2365): # Add test case for "googleapis.com" configured universe and fake credentials' universe.
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=mock.Mock(universe_domain=universe.DEFAULT_UNIVERSE),
+                http=build_http(),
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=fake_universe
+                ),
+            )
+
+            with self.assertRaises(universe.UniverseMismatchError):
+                service._validate_credentials()
 
         def test_validate_credentials_with_a_different_universe(self):
             fake_universe = "foo.com"
 
-            # TODO(google-api-python-client/issues/2365): Add test case for fake configured universe and fake credentials' universe.
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=mock.Mock(universe_domain=fake_universe),
+                http=build_http(),
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=fake_universe
+                ),
+            )
+
+            assert service._validate_credentials()
 
             http = google_auth_httplib2.AuthorizedHttp(
                 credentials=mock.Mock(universe_domain=fake_universe), http=build_http()
@@ -2426,21 +2483,34 @@ if HAS_UNIVERSE:
             # Calling service._validate_credentials() again returns service.credentials_validated.
             assert service._validate_credentials()
 
-            # TODO(google-api-python-client/issues/2365): Add test case for fake configured universe and fake credentials' universe.
-
-        def test_validate_credentials_before_api_request(self):
-            fake_universe = "foo.com"
-
             http = google_auth_httplib2.AuthorizedHttp(
-                credentials=mock.Mock(universe_domain=universe.DEFAULT_UNIVERSE),
-                http=build_http(),
+                credentials=mock.Mock(universe_domain=fake_universe), http=build_http()
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=fake_universe
+                ),
             )
+
+            assert service._validate_credentials()
+            assert service._credentials_validated
+
+            # Calling service._validate_credentials() again returns service.credentials_validated.
+            assert service._validate_credentials()
+
+        def test_validate_credentials_before_api_request_success(self):
+            fake_universe = "foo.com"
+            credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+            credentials.universe_domain = fake_universe
             discovery = read_datafile("tasks.json")
             tasks = build_from_document(
                 discovery,
-                http=http,
+                credentials=credentials,
                 client_options=google.api_core.client_options.ClientOptions(
-                    universe_domain=universe.DEFAULT_UNIVERSE
+                    universe_domain=fake_universe
                 ),
             )
 
@@ -2450,13 +2520,14 @@ if HAS_UNIVERSE:
             # Check that credentials are indeed verified before request.
             assert tasklists._validate_credentials()
 
-            http = google_auth_httplib2.AuthorizedHttp(
-                credentials=mock.Mock(universe_domain=fake_universe), http=build_http()
-            )
+        def test_validate_credentials_before_api_request_failure(self):
+            fake_universe = "foo.com"
+            credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+            credentials.universe_domain = fake_universe
             discovery = read_datafile("tasks.json")
             tasks = build_from_document(
                 discovery,
-                http=http,
+                credentials=credentials,
                 client_options=google.api_core.client_options.ClientOptions(
                     universe_domain=universe.DEFAULT_UNIVERSE
                 ),
@@ -2466,6 +2537,69 @@ if HAS_UNIVERSE:
             with self.assertRaises(universe.UniverseMismatchError):
                 request = tasks.tasklists().list()
 
+        def test_validate_credentials_before_another_universe_api_request_failure(self):
+            fake_universe = "foo.com"
+            credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+            credentials.universe_domain = fake_universe
+            another_fake_universe = "bar.com"
+            discovery = read_datafile("tasks.json")
+            tasks = build_from_document(
+                discovery,
+                credentials=credentials,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=another_fake_universe
+                ),
+            )
+
+            # Check that credentials are verified before request.
+            with self.assertRaises(universe.UniverseMismatchError):
+                request = tasks.tasklists().list()
+
+        def test_client_options_with_empty_universe(self):
+            fake_universe = "foo.com"
+            credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+            discovery = read_datafile("tasks.json")
+
+            with self.assertRaises(universe.EmptyUniverseError):
+                tasks = build_from_document(
+                    discovery,
+                    credentials=credentials,
+                    client_options=google.api_core.client_options.ClientOptions(
+                        universe_domain=""
+                    ),
+                )
+
+        def test_client_options_universe_configured_with_mtls(self):
+            fake_universe = "foo.com"
+            discovery = read_datafile("tasks.json")
+
+            with self.assertRaises(MutualTLSChannelError):
+                with mock.patch.dict(
+                    "os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}
+                ):
+                    tasks = build_from_document(
+                        discovery,
+                        client_options=google.api_core.client_options.ClientOptions(
+                            universe_domain=fake_universe
+                        ),
+                    )
+
+        def test_client_options_universe_configured_with_api_override(self):
+            fake_universe = "foo.com"
+            fake_api_endpoint = "https://www.bar.com/"
+            credentials = mock.Mock(universe_domain=fake_universe)
+            discovery = read_datafile("tasks.json")
+
+            tasks = build_from_document(
+                discovery,
+                credentials=credentials,
+                client_options=google.api_core.client_options.ClientOptions(
+                    api_endpoint=fake_api_endpoint, universe_domain=fake_universe
+                ),
+            )
+
+            assert tasks._baseUrl == fake_api_endpoint
+
 
 if __name__ == "__main__":
     unittest.main()