Fix minor issues

Author: nbayati

google/auth/_regional_access_boundary_utils.py

@@ -3,8 +3,11 @@
 import datetime
 import threading
 
+import logging
+
 from google.auth import _helpers
-from google.auth._default import _LOGGER
+
+_LOGGER = logging.getLogger(__name__)
 
 
 # The default lifetime for a cached Regional Access Boundary.

google/auth/compute_engine/credentials.py

@@ -242,6 +242,7 @@ def with_scopes(self, scopes, default_scopes=None):
     def with_universe_domain(self, universe_domain):
         creds = self._make_copy()
         creds._universe_domain = universe_domain
+        creds._universe_domain_cached = True
         return creds
 
 

google/auth/external_account.py

@@ -348,6 +348,7 @@ def with_scopes(self, scopes, default_scopes=None):
         scoped = self.__class__(**kwargs)
         scoped._cred_file_path = self._cred_file_path
         scoped._metrics_options = self._metrics_options
+        self._copy_regional_access_boundary_state(scoped)
         return scoped
 
     @abc.abstractmethod
@@ -505,7 +506,11 @@ def _build_regional_access_boundary_lookup_url(self):
             return url
         else:
             # If both fail, the audience format is invalid.
-            raise exceptions.InvalidValue("Invalid audience format.")
+            _LOGGER.error(
+                "Invalid audience format for Regional Access Boundary lookup: %s",
+                self._audience,
+            )
+            return None
 
     def _make_copy(self):
         kwargs = self._constructor_args()

google/oauth2/_service_account_async.py

@@ -77,10 +77,10 @@ async def refresh(self, request):
 
     @_helpers.copy_docstring(credentials_async.Credentials)
     async def before_request(self, request, method, url, headers):
+        # Explicit override to bypass synchronous CredentialsWithRegionalAccessBoundary.
         await credentials_async.Credentials.before_request(
             self, request, method, url, headers
         )
-        self._maybe_start_regional_access_boundary_refresh(request, url)
 
 
 class IDTokenCredentials(
@@ -137,3 +137,11 @@ async def refresh(self, request):
         )
         self.token = access_token
         self.expiry = expiry
+
+    @_helpers.copy_docstring(credentials_async.Credentials)
+    async def before_request(self, request, method, url, headers):
+        # Explicit override to bypass synchronous CredentialsWithRegionalAccessBoundary
+        # and disable Regional Access Boundary refresh for async credentials.
+        await credentials_async.Credentials.before_request(
+            self, request, method, url, headers
+        )

tests/oauth2/test__client.py

@@ -699,6 +699,7 @@ def test_lookup_regional_access_boundary_non_retryable_error(status_code):
     # Non-retryable errors should only be called once.
     mock_request.assert_called_once_with(method="GET", url=url, headers=headers)
 
+
 def test_lookup_regional_access_boundary_internal_failure_and_retry_failure_error():
     retryable_error = mock.create_autospec(transport.Response, instance=True)
     retryable_error.status = http_client.BAD_REQUEST

tests/oauth2/test_service_account.py

@@ -242,7 +242,9 @@ def test_build_regional_access_boundary_lookup_url(self):
         credentials = self.make_credentials()
         expected_url = (
             "https://iamcredentials.googleapis.com/v1/projects/-/"
-            "serviceAccounts/{}/allowedLocations".format(credentials.service_account_email)
+            "serviceAccounts/{}/allowedLocations".format(
+                credentials.service_account_email
+            )
         )
         assert credentials._build_regional_access_boundary_lookup_url() == expected_url
 

tests/test_external_account.py

@@ -351,6 +351,16 @@ def test_with_scopes(self):
         assert scoped_credentials.has_scopes(["email"])
         assert not scoped_credentials.requires_scopes
 
+    def test_with_scopes_copies_regional_access_boundary(self):
+        credentials = self.make_credentials()
+        credentials = credentials._with_regional_access_boundary(
+            self.VALID_TRUST_BOUNDARY
+        )
+        scoped_credentials = credentials.with_scopes(["email"])
+
+        assert scoped_credentials.has_scopes(["email"])
+        assert scoped_credentials._regional_access_boundary == self.VALID_TRUST_BOUNDARY
+
     def test_with_scopes_workforce_pool(self):
         credentials = self.make_workforce_pool_credentials(
             workforce_pool_user_project=self.WORKFORCE_POOL_USER_PROJECT
@@ -1750,8 +1760,7 @@ def test_build_regional_access_boundary_lookup_url_workforce(self):
     def test_build_regional_access_boundary_lookup_url_invalid_audience(self, audience):
         credentials = self.make_credentials()
         credentials._audience = audience
-        with pytest.raises(exceptions.InvalidValue, match="Invalid audience format."):
-            credentials._build_regional_access_boundary_lookup_url()
+        assert credentials._build_regional_access_boundary_lookup_url() is None
 
     def test_with_regional_access_boundary(self):
         credentials = self.make_credentials()