check universe is not supported for api-core <= 2.18.0

Author: ohmayr

googleapiclient/discovery.py

@@ -130,6 +130,11 @@
 STACK_QUERY_PARAMETERS = frozenset(["trace", "pp", "userip", "strict"])
 STACK_QUERY_PARAMETER_DEFAULT_VALUE = {"type": "string", "location": "query"}
 
+class APICoreVersionError(ValueError):
+    def __init__(self):
+        message = "google-api-core >= 2.18.0 is required to use the universe domain feature."
+        super().__init__(message)
+
 # Library-specific reserved words beyond Python keywords.
 RESERVED_WORDS = frozenset(["body"])
 
@@ -559,6 +564,9 @@ def build_from_document(
             client_options.universe_domain, universe_domain_env
         )
         base = base.replace(universe.DEFAULT_UNIVERSE, universe_domain)
+    else:
+        if client_options.universe_domain:
+            raise APICoreVersionError
 
     audience_for_self_signed_jwt = base
     if client_options.api_endpoint:
@@ -613,7 +621,7 @@ def build_from_document(
         ):
             credentials = credentials.with_always_use_jwt_access(always_use_jwt_access)
             credentials._create_self_signed_jwt(audience_for_self_signed_jwt)
-
+        
         # If credentials are provided, create an authorized http instance;
         # otherwise, skip authentication.
         if credentials:
@@ -688,6 +696,12 @@ def build_from_document(
                     )
                 base = mtls_endpoint
 
+    if not HAS_UNIVERSE:
+        credentials = getattr(http, "credentials", None)
+        universe_domain = getattr(credentials, "universe_domain", None)
+        if universe_domain != "googleapis.com":
+                raise APICoreVersionError
+
     if model is None:
         features = service.get("features", [])
         model = JsonModel("dataWrapper" in features)

tests/test_discovery.py

@@ -71,6 +71,7 @@
     V1_DISCOVERY_URI,
     V2_DISCOVERY_URI,
     ResourceMethodParameters,
+    APICoreVersionError,
     _fix_up_media_path_base_url,
     _fix_up_media_upload,
     _fix_up_method_description,
@@ -2501,170 +2502,196 @@ def test_validate_credentials_with_already_validated_credentials(self):
             # 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,
-            credentials=credentials,
-            client_options=google.api_core.client_options.ClientOptions(
-                universe_domain=fake_universe
-            ),
-        )
-
-        tasklists = tasks.tasklists()
-        request = tasklists.list()
-
-        # Check that credentials are indeed verified before request.
-        assert tasklists._validate_credentials()
-
-    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,
-            credentials=credentials,
-            client_options=google.api_core.client_options.ClientOptions(
-                universe_domain=universe.DEFAULT_UNIVERSE
-            ),
-        )
+        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,
+                credentials=credentials,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=fake_universe
+                ),
+            )
 
-        # Check that credentials are verified before request.
-        with self.assertRaises(universe.UniverseMismatchError):
-            request = tasks.tasklists().list()
+            tasklists = tasks.tasklists()
+            request = 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 indeed verified before request.
+            assert tasklists._validate_credentials()
 
-        # Check that credentials are verified before request.
-        with self.assertRaises(universe.UniverseMismatchError):
-            request = tasks.tasklists().list()
+        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,
+                credentials=credentials,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=universe.DEFAULT_UNIVERSE
+                ),
+            )
 
-    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")
+            # Check that credentials are verified before request.
+            with self.assertRaises(universe.UniverseMismatchError):
+                request = tasks.tasklists().list()
 
-        with self.assertRaises(universe.EmptyUniverseError):
+        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=""
+                    universe_domain=another_fake_universe
                 ),
             )
 
-    def test_client_options_universe_configured_with_mtls(self):
-        fake_universe = "foo.com"
-        discovery = read_datafile("tasks.json")
+            # Check that credentials are verified before request.
+            with self.assertRaises(universe.UniverseMismatchError):
+                request = tasks.tasklists().list()
 
-        with self.assertRaises(MutualTLSChannelError):
-            with mock.patch.dict(
-                "os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}
-            ):
+        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=fake_universe
+                        universe_domain=""
                     ),
                 )
 
-    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")
+        def test_client_options_universe_configured_with_mtls(self):
+            fake_universe = "foo.com"
+            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
-            ),
-        )
+            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
+                        ),
+                    )
 
-        assert tasks._baseUrl == fake_api_endpoint
+        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")
 
-    def test_universe_env_var_configured_empty(self):
-        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
-        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
+
+        def test_universe_env_var_configured_empty(self):
+            credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+            discovery = read_datafile("tasks.json")
+
+            with self.assertRaises(universe.EmptyUniverseError):
+                with mock.patch.dict(
+                    "os.environ", {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": ""}
+                ):
+                    tasks = build_from_document(
+                        discovery,
+                        credentials=credentials,
+                    )
+
+        def test_universe_env_var_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",
+                        "GOOGLE_CLOUD_UNIVERSE_DOMAIN": fake_universe,
+                    },
+                ):
+                    tasks = build_from_document(discovery)
+
+        def test_universe_env_var_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")
 
-        with self.assertRaises(universe.EmptyUniverseError):
             with mock.patch.dict(
-                "os.environ", {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": ""}
+                "os.environ", {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": fake_universe}
             ):
                 tasks = build_from_document(
                     discovery,
                     credentials=credentials,
+                    client_options=google.api_core.client_options.ClientOptions(
+                        api_endpoint=fake_api_endpoint
+                    ),
                 )
 
-    def test_universe_env_var_configured_with_mtls(self):
-        fake_universe = "foo.com"
-        discovery = read_datafile("tasks.json")
+            assert tasks._baseUrl == fake_api_endpoint
+
+        def test_universe_env_var_configured_with_client_options_universe(self):
+            fake_universe = "foo.com"
+            another_fake_universe = "bar.com"
+            credentials = mock.Mock(universe_domain=fake_universe)
+            discovery = read_datafile("tasks.json")
 
-        with self.assertRaises(MutualTLSChannelError):
             with mock.patch.dict(
-                "os.environ",
-                {
-                    "GOOGLE_API_USE_MTLS_ENDPOINT": "always",
-                    "GOOGLE_CLOUD_UNIVERSE_DOMAIN": fake_universe,
-                },
+                "os.environ", {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": another_fake_universe}
             ):
-                tasks = build_from_document(discovery)
+                tasks = build_from_document(
+                    discovery,
+                    credentials=credentials,
+                    client_options=google.api_core.client_options.ClientOptions(
+                        universe_domain=fake_universe
+                    ),
+                )
 
-    def test_universe_env_var_configured_with_api_override(self):
+            assert tasks._universe_domain == fake_universe
+    
+    def test_client_options_universe_with_older_version_of_api_core(self):
         fake_universe = "foo.com"
-        fake_api_endpoint = "https://www.bar.com/"
-        credentials = mock.Mock(universe_domain=fake_universe)
+        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+        credentials.universe_domain = fake_universe
         discovery = read_datafile("tasks.json")
-
-        with mock.patch.dict(
-            "os.environ", {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": fake_universe}
-        ):
+        with self.assertRaises(APICoreVersionError):
             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
 
-    def test_universe_env_var_configured_with_client_options_universe(self):
+    def test_credentials_universe_with_older_version_of_api_core(self):
         fake_universe = "foo.com"
-        another_fake_universe = "bar.com"
-        credentials = mock.Mock(universe_domain=fake_universe)
+        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+        credentials.universe_domain = fake_universe
         discovery = read_datafile("tasks.json")
-
-        with mock.patch.dict(
-            "os.environ", {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": another_fake_universe}
-        ):
+        with self.assertRaises(APICoreVersionError):
             tasks = build_from_document(
                 discovery,
                 credentials=credentials,
-                client_options=google.api_core.client_options.ClientOptions(
-                    universe_domain=fake_universe
-                ),
             )
-
-        assert tasks._universe_domain == fake_universe
-
+            
 
 if __name__ == "__main__":
     unittest.main()