test(auth): address RAB feedback with robust coverage and docs

- Add test for token refresh and RAB lookup sequencing in before_request.
- Add failure test to verify blocking RAB lookups are not retried.
- Restore and refine test for skipping RAB lookup when URL is None.
- Fix swapped url/method arguments in before_request test calls.
- Document why OAuth2 credentials skip independent RAB lookups.
- Internalize blocking lookup method with leading underscore.

Author: macastelaz

packages/google-auth/google/auth/credentials.py

@@ -377,8 +377,13 @@ def _with_regional_access_boundary(self, seed):
         creds._rab_manager.set_initial_regional_access_boundary(seed)
         return creds
 
-    def with_blocking_regional_access_boundary_lookup(self):
+    def _with_blocking_regional_access_boundary_lookup(self):
         """Returns a copy of these credentials with the blocking lookup mode enabled.
+        This is intended for internal use only as blocking lookup requires additional
+        care and consideration. Currently this is unsed by the gcloud CLI and
+        therefore changes to the contract MUST be backwards compatible (e.g. the
+        method signature must be unchanged and a copy of the credentials with the
+        blocking lookup flag set to true must be returned).
 
         Returns:
             google.auth.credentials.Credentials: A new credentials instance.

packages/google-auth/google/oauth2/credentials.py

@@ -210,7 +210,6 @@ def __setstate__(self, d):
         self._rab_manager = d.get("_rab_manager") or (
             _regional_access_boundary_utils._RegionalAccessBoundaryManager()
         )
-        self._use_blocking_regional_access_boundary_lookup = False
 
     @property
     def refresh_token(self):
@@ -363,6 +362,15 @@ def _metric_header_for_usage(self):
         return metrics.CRED_TYPE_USER
 
     def _build_regional_access_boundary_lookup_url(self, request=None):
+        """Builds the URL for Regional Access Boundary lookup.
+
+        OAuth 2.0 credentials do not support independent Regional Access Boundary
+        lookup. However, they may support a seeded Regional Access Boundary
+        provided externally (e.g., from gcloud).
+
+        Returns:
+            None: This credential type does not support RAB lookup.
+        """
         return None
 
     def _perform_refresh_token(self, request):

packages/google-auth/tests/compute_engine/test_credentials.py

@@ -455,7 +455,7 @@ def test_with_blocking_regional_access_boundary_lookup(self):
         creds = self.credentials
         assert not creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-        new_creds = creds.with_blocking_regional_access_boundary_lookup()
+        new_creds = creds._with_blocking_regional_access_boundary_lookup()
         assert new_creds._rab_manager._use_blocking_regional_access_boundary_lookup
         assert new_creds is not creds
 

packages/google-auth/tests/oauth2/test__client.py

@@ -810,3 +810,23 @@ def test_lookup_regional_access_boundary_blocking():
     mock_request.assert_called_once_with(
         method="GET", url=url, headers=headers, timeout=3
     )
+
+def test_lookup_regional_access_boundary_blocking_error():
+    mock_response = mock.create_autospec(transport.Response, instance=True)
+    mock_response.status = http_client.INTERNAL_SERVER_ERROR
+    mock_response.data = "this is an error message"
+
+    mock_request = mock.create_autospec(transport.Request)
+    mock_request.return_value = mock_response
+
+    url = "http://example.com"
+    headers = {"Authorization": "Bearer access_token"}
+    # Even if the error is retryable, blocking=True should prevent retries.
+    result = _client._lookup_regional_access_boundary(
+        mock_request, url, headers=headers, blocking=True
+    )
+    assert result is None
+
+    mock_request.assert_called_once_with(
+        method="GET", url=url, headers=headers, timeout=3
+    )

packages/google-auth/tests/oauth2/test_credentials.py

@@ -100,7 +100,7 @@ def test_with_blocking_regional_access_boundary_lookup(self):
         creds = self.make_credentials()
         assert not creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-        new_creds = creds.with_blocking_regional_access_boundary_lookup()
+        new_creds = creds._with_blocking_regional_access_boundary_lookup()
         assert new_creds._rab_manager._use_blocking_regional_access_boundary_lookup
         assert new_creds is not creds
 

packages/google-auth/tests/test__regional_access_boundary_utils.py

@@ -22,6 +22,7 @@
 from google.auth import _regional_access_boundary_utils
 from google.auth import credentials
 from google.auth import environment_vars
+from google.oauth2 import credentials as oauth2_credentials
 
 
 class CredentialsImpl(credentials.CredentialsWithRegionalAccessBoundary):
@@ -227,7 +228,7 @@ def test_with_blocking_regional_access_boundary_lookup(self):
         creds = CredentialsImpl()
         assert not creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-        new_creds = creds.with_blocking_regional_access_boundary_lookup()
+        new_creds = creds._with_blocking_regional_access_boundary_lookup()
         assert new_creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
     def test_with_regional_access_boundary(self):
@@ -405,6 +406,12 @@ def test_lookup_regional_access_boundary_failure(self, mock_lookup_rab):
         assert rab_manager._data.expiry is None
         assert rab_manager._data.cooldown_expiry is not None
 
+    def test_lookup_regional_access_boundary_null_url(self):
+        creds = oauth2_credentials.Credentials(token="token")
+        request = mock.Mock()
+        result = creds._lookup_regional_access_boundary(request)
+        assert result is None
+
     def test_credentials_with_regional_access_boundary_initialization(self):
         creds = CredentialsImpl()
         assert creds._rab_manager._data.encoded_locations is None

packages/google-auth/tests/test_credentials.py

@@ -38,7 +38,7 @@ def _perform_refresh_token(self, request):
     def with_quota_project(self, quota_project_id):
         raise NotImplementedError()
 
-    def _build_regional_access_boundary_lookup_url(self):
+    def _build_regional_access_boundary_lookup_url(self, request=None):
         # Using self.token here to make the URL dynamic for testing purposes
         return "http://mock.url/lookup_for_{}".format(self.token)
 
@@ -107,7 +107,7 @@ def test_before_request():
     headers = {}
 
     # First call should call refresh, setting the token.
-    credentials.before_request(request, "http://example.com", "GET", headers)
+    credentials.before_request(request, "GET", "http://example.com", headers)
     assert credentials.valid
     assert credentials.token == "refreshed-token"
     assert headers["authorization"] == "Bearer refreshed-token"
@@ -117,7 +117,7 @@ def test_before_request():
     headers = {}
 
     # Second call shouldn't call refresh.
-    credentials.before_request(request, "http://example.com", "GET", headers)
+    credentials.before_request(request, "GET", "http://example.com", headers)
     assert credentials.valid
     assert credentials.token == "refreshed-token"
     assert headers["authorization"] == "Bearer refreshed-token"
@@ -137,7 +137,7 @@ def test_before_request_with_regional_access_boundary():
     headers = {}
 
     # First call should call refresh, setting the token.
-    creds.before_request(request, "http://example.com", "GET", headers)
+    creds.before_request(request, "GET", "http://example.com", headers)
     assert creds.valid
     assert creds.token == "refreshed-token"
     assert headers["authorization"] == "Bearer refreshed-token"
@@ -147,7 +147,7 @@ def test_before_request_with_regional_access_boundary():
     headers = {}
 
     # Second call shouldn't call refresh.
-    creds.before_request(request, "http://example.com", "GET", headers)
+    creds.before_request(request, "GET", "http://example.com", headers)
     assert creds.valid
     assert creds.token == "refreshed-token"
     assert headers["authorization"] == "Bearer refreshed-token"
@@ -159,7 +159,7 @@ def test_before_request_metrics():
     request = "token"
     headers = {}
 
-    credentials.before_request(request, "http://example.com", "GET", headers)
+    credentials.before_request(request, "GET", "http://example.com", headers)
     assert headers["x-goog-api-client"] == "foo"
 
 
@@ -282,7 +282,7 @@ def test_nonblocking_refresh_fresh_credentials():
     assert c.token_state == credentials.TokenState.FRESH
 
     c.with_non_blocking_refresh()
-    c.before_request(request, "http://example.com", "GET", {})
+    c.before_request(request, "GET", "http://example.com", {})
 
 
 def test_nonblocking_refresh_invalid_credentials():
@@ -294,7 +294,7 @@ def test_nonblocking_refresh_invalid_credentials():
 
     assert c.token_state == credentials.TokenState.INVALID
 
-    c.before_request(request, "http://example.com", "GET", headers)
+    c.before_request(request, "GET", "http://example.com", headers)
     assert c.token_state == credentials.TokenState.FRESH
     assert c.valid
     assert c.token == "refreshed-token"
@@ -310,7 +310,7 @@ def test_nonblocking_refresh_stale_credentials():
     headers = {}
 
     # Invalid credentials MUST require a blocking refresh.
-    c.before_request(request, "http://example.com", "GET", headers)
+    c.before_request(request, "GET", "http://example.com", headers)
     assert c.token_state == credentials.TokenState.FRESH
     assert not c._refresh_worker._worker
 
@@ -322,7 +322,7 @@ def test_nonblocking_refresh_stale_credentials():
 
     # STALE credentials SHOULD spawn a non-blocking worker
     assert c.token_state == credentials.TokenState.STALE
-    c.before_request(request, "http://example.com", "GET", headers)
+    c.before_request(request, "GET", "http://example.com", headers)
     assert c._refresh_worker._worker is not None
 
     assert c.token_state == credentials.TokenState.FRESH
@@ -340,7 +340,7 @@ def test_nonblocking_refresh_failed_credentials():
     headers = {}
 
     # Invalid credentials MUST require a blocking refresh.
-    c.before_request(request, "http://example.com", "GET", headers)
+    c.before_request(request, "GET", "http://example.com", headers)
     assert c.token_state == credentials.TokenState.FRESH
     assert not c._refresh_worker._worker
 
@@ -354,7 +354,7 @@ def test_nonblocking_refresh_failed_credentials():
     assert c.token_state == credentials.TokenState.STALE
     c._refresh_worker._worker = mock.MagicMock()
     c._refresh_worker._worker._error_info = "Some Error"
-    c.before_request(request, "http://example.com", "GET", headers)
+    c.before_request(request, "GET", "http://example.com", headers)
     assert c._refresh_worker._worker is not None
 
     assert c.token_state == credentials.TokenState.FRESH
@@ -373,7 +373,7 @@ def test_token_state_no_expiry():
     c.expiry = None
     assert c.token_state == credentials.TokenState.FRESH
 
-    c.before_request(request, "http://example.com", "GET", {})
+    c.before_request(request, "GET", "http://example.com", {})
 
 
 def test_credentials_with_trust_boundary_bridge():
@@ -389,3 +389,34 @@ def _build_trust_boundary_lookup_url(self):
     # Verify that calling the new method delegates to the old method
     with pytest.warns(DeprecationWarning):
         assert creds._build_regional_access_boundary_lookup_url() == "http://legacy.url"
+
+def test_before_request_triggers_rab_refresh():
+    with mock.patch(
+        "google.auth._regional_access_boundary_utils.is_regional_access_boundary_enabled",
+        return_value=True,
+    ):
+        with mock.patch("google.oauth2._client._lookup_regional_access_boundary") as lookup:
+            lookup.return_value = {"encodedLocations": "0xA30"}
+
+            creds = CredentialsImpl()
+            creds = creds._with_blocking_regional_access_boundary_lookup()
+
+            request = mock.Mock()
+            headers = {}
+
+            # Initial state: no token
+            assert creds.token is None
+
+            # before_request should trigger token refresh and THEN RAB refresh.
+            # We verify this by checking that the RAB lookup was called with
+            # the URL containing the refreshed token.
+            creds.before_request(request, "GET", "http://example.com", headers)
+
+            assert creds.token == "refreshed-token"
+            assert headers["authorization"] == "Bearer refreshed-token"
+            assert headers["x-allowed-locations"] == "0xA30"
+
+            # Verify lookup was called with the refreshed token's URL
+            lookup.assert_called_once()
+            args, kwargs = lookup.call_args
+            assert args[1] == "http://mock.url/lookup_for_refreshed-token"