Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions storages/backends/gcloud.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@

try:
from google import auth
from google.auth.credentials import TokenState
from google.auth.transport import requests
from google.cloud.exceptions import NotFound
from google.cloud.storage import Blob
Expand Down Expand Up @@ -356,7 +355,11 @@ def get_available_name(self, name, max_length=None):
return super().get_available_name(name, max_length)

def _get_iam_sign_blob_params(self):
if self.credentials.token_state != TokenState.FRESH:
# Check if credentials need refreshing before signing
# TokenState.FRESH = 1, STALE = 2, INVALID = 3
# We hardcode the value 1 (FRESH) to avoid importing TokenState,
# which was only added in google-auth 2.26.0, ensuring backward compatibility
if self.credentials.token_state != 1: # Not FRESH
self.credentials.refresh(requests.Request())

try:
Expand Down
17 changes: 17 additions & 0 deletions tests/test_gcloud.py
Original file line number Diff line number Diff line change
Expand Up @@ -578,6 +578,23 @@ def test_iam_sign_blob_with_sa_email_setting(self):
access_token=storage.credentials.token,
)

def test_iam_sign_blob_params_with_integer_token_state(self):
"""
Test that _get_iam_sign_blob_params works with integer token_state values
(backward compatibility)
"""
with override_settings(GS_IAM_SIGN_BLOB=True, GS_SA_EMAIL="[email protected]"):
storage = gcloud.GoogleCloudStorage()
storage.credentials = mock.MagicMock()
storage.credentials.token = "access_token_123"
storage.credentials.token_state = 2 # STALE - should refresh

email, token = storage._get_iam_sign_blob_params()

self.assertEqual(email, "[email protected]")
self.assertEqual(token, "access_token_123")
storage.credentials.refresh.assert_called_once()


class GoogleCloudGzipClientTests(GCloudTestCase):
def setUp(self):
Expand Down