From ccceba4952ae11ea407730a869d0b3e078fbdf2f Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 16:29:40 -0700
Subject: [PATCH 01/16] add sync to async adapter

---
 .../%service/transports/rest_asyncio.py.j2    | 58 +++++++++++---
 .../%name_%version/%sub/test_%service.py.j2   | 18 +++++
 .../gapic/%name_%version/%sub/test_macros.j2  | 76 ++++++++++++++++++
 .../cloud_redis/transports/rest_asyncio.py    | 54 +++++++++++--
 .../unit/gapic/redis_v1/test_cloud_redis.py   | 77 +++++++++++++++++++
 .../cloud_redis/transports/rest_asyncio.py    | 54 +++++++++++--
 .../unit/gapic/redis_v1/test_cloud_redis.py   | 77 +++++++++++++++++++
 7 files changed, 392 insertions(+), 22 deletions(-)

diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
index 0f79d6e1ffef..dc09bdb97628 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
@@ -17,7 +17,10 @@ try:
 except ImportError as e:  # pragma: NO COVER
     raise ImportError("`rest_asyncio` transport requires the library to be installed with the `async_rest` extra. Install the library with the `async_rest` extra using `pip install {{ api.naming.warehouse_package_name }}[async_rest]`") from e
 
+import asyncio
+from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
+import google.auth.transport.requests  # type: ignore
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
@@ -93,6 +96,35 @@ class Async{{service.name}}RestStub:
     _host: str
     _interceptor: Async{{service.name}}RestInterceptor
 
+
+class _SyncToAsyncCredentialsAdapter(ga_credentials_async.Credentials):
+    """Adapts synchronous credentials to the asynchronous credentials interface.
+
+    ``AsyncAuthorizedSession`` only accepts :class:`google.auth.aio.credentials.Credentials`.
+    This adapter allows synchronous :class:`google.auth.credentials.Credentials`
+    (e.g. application default credentials) to be used with the asynchronous
+    REST transport by delegating to the wrapped credentials. Blocking operations,
+    i.e. refreshing the access token, are executed in a worker thread so that the
+    event loop is not blocked.
+    """
+
+    def __init__(self, credentials: ga_credentials.Credentials):
+        super().__init__()
+        self._credentials = credentials
+        # Synchronous credentials require a synchronous transport to refresh the access token.
+        self._request = google.auth.transport.requests.Request()
+
+    async def apply(self, headers, token=None):
+        self._credentials.apply(headers, token=token)
+
+    async def refresh(self, request):
+        # `request` is an asynchronous transport which cannot be used by synchronous credentials.
+        await asyncio.to_thread(self._credentials.refresh, self._request)
+
+    async def before_request(self, request, method, url, headers):
+        await asyncio.to_thread(self._credentials.before_request, self._request, method, url, headers)
+
+
 class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
     """Asynchronous REST backend transport for {{ service.name }}.
 
@@ -107,7 +139,7 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
     def __init__(self,
             *,
             host: str{% if service.host %} = '{{ service.host }}'{% endif %},
-            credentials: Optional[ga_credentials_async.Credentials] = None,
+            credentials: Optional[Union[ga_credentials.Credentials, ga_credentials_async.Credentials]] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             url_scheme: str = 'https',
             interceptor: Optional[Async{{ service.name }}RestInterceptor] = None,
@@ -123,11 +155,13 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
         Args:
             host ({% if service.host %}Optional[str]{% else %}str{% endif %}):
                 {{ ' ' }}The hostname to connect to {% if service.host %}(default: '{{ service.host }}'){% endif %}.
-            credentials (Optional[google.auth.aio.credentials.Credentials]): The
+            credentials (Optional[Union[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
-                credentials from the environment.
+                credentials from the environment. Synchronous credentials are
+                adapted for use with the asynchronous transport; any blocking
+                token refresh is performed in a worker thread.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):
                 The client info used to send a user-agent string along with
                 API requests. If ``None``, then default info will be used.
@@ -148,11 +182,19 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             url_scheme=url_scheme,
             api_audience=None
         )
+        # The credentials provided to (or resolved by) the base transport may be synchronous,
+        # e.g. when they are obtained from the environment. Adapt them for use with the
+        # asynchronous session while leaving `self._credentials` untouched.
+        self._async_credentials = (
+            _SyncToAsyncCredentialsAdapter(self._credentials)
+            if isinstance(self._credentials, ga_credentials.Credentials)
+            else self._credentials
+        )
         {# Note: Type for creds is ignored because of incorrect type hint for creds in the client layer.
          # TODO(https://github.com/googleapis/gapic-generator-python/issues/2177): Remove `# type: ignore` once
          # we update the type hints for credentials to include asynchronous credentials in the client layer.
         #}
-        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
+        self._session = AsyncAuthorizedSession(self._async_credentials)  # type: ignore
         self._interceptor = interceptor or Async{{ service.name }}RestInterceptor()
         self._wrap_with_kind = True
         self._prep_wrapped_messages(client_info)
@@ -300,12 +342,8 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
 
             rest_transport = operations_v1.AsyncOperationsRestTransport(  # type: ignore
                     host=self._host,
-                    # use the credentials which are saved
-                    {# Note: Type for creds is ignored because of incorrect type hint for creds in the client layer.
-                     # TODO(https://github.com/googleapis/gapic-generator-python/issues/2177): Remove `# type: ignore` once
-                     # we update the type hints for credentials to include asynchronous credentials in the client layer.
-                    #}
-                    credentials=self._credentials,  # type: ignore
+                    # use the credentials which are saved, adapted for asynchronous use
+                    credentials=self._async_credentials,
                     http_options=http_options,
                     path_prefix="{{ service.client_package_version }}"
             )
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index d6880c73605a..9804796f1dd4 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -55,6 +55,9 @@ except ImportError: # pragma: NO COVER
 {% filter sort_lines %}
 import google.auth
 from google.auth import credentials as ga_credentials
+{% if rest_async_io_enabled %}
+import google.auth.transport.requests
+{% endif %}
 
 
 from google.auth.exceptions import MutualTLSChannelError
@@ -281,9 +284,16 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
     {% endif %}
     {% if 'rest' in opts.transport %}
     ({{ service.client_name }}, "rest"),
+    {% if rest_async_io_enabled %}
+    ({{ service.async_client_name }}, "rest_asyncio"),
+    {% endif %}
     {% endif %}
 ])
 def test_{{ service.client_name|snake_case }}_from_service_account_info(client_class, transport_name):
+    {% if rest_async_io_enabled %}
+    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    {% endif %}
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_info') as factory:
         factory.return_value = creds
@@ -332,9 +342,16 @@ def test_{{ service.client_name|snake_case }}_service_account_always_use_jwt(tra
     {% endif %}
     {% if 'rest' in opts.transport %}
     ({{ service.client_name }}, "rest"),
+    {% if rest_async_io_enabled %}
+    ({{ service.async_client_name }}, "rest_asyncio"),
+    {% endif %}
     {% endif %}
 ])
 def test_{{ service.client_name|snake_case }}_from_service_account_file(client_class, transport_name):
+    {% if rest_async_io_enabled %}
+    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    {% endif %}
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_file') as factory:
         factory.return_value = creds
@@ -1117,6 +1134,7 @@ def test_transport_adc(transport_class):
 {% if rest_async_io_enabled %}
 {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove the below macro when async rest is GA and supports all required parameters. #}
 {{ test_macros.async_rest_unsupported_params_test(service) }}
+{{ test_macros.async_rest_credentials_test(service) }}
 {% endif %}
 {% if 'grpc' in opts.transport %}
 def test_transport_grpc_default():
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
index 05c60cbdc9c3..1169d797f2fd 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
@@ -1688,6 +1688,82 @@ def test_unsupported_parameter_rest_asyncio():
 
 {% endmacro %}
 
+{# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Fold these tests into the generic
+ # transport tests (e.g. `test_transport_adc`) when async rest is GA. #}
+{% macro async_rest_credentials_test(service) %}
+@pytest.mark.asyncio
+async def test_sync_credentials_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Synchronous credentials are accepted, consistent with every other transport.
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(transports.rest_asyncio, "AsyncAuthorizedSession", autospec=True) as session_class:
+        transport = {{ get_client(service, True) }}.get_transport_class("rest_asyncio")(credentials=creds)
+    # The provided credentials are preserved on the transport.
+    assert transport._credentials is creds
+    # The asynchronous session is created with an adapted version of the credentials.
+    async_creds = transport._async_credentials
+    assert isinstance(async_creds, ga_credentials_async.Credentials)
+    session_class.assert_called_once_with(async_creds)
+
+    # The adapted credentials delegate to the synchronous credentials using a
+    # synchronous transport, rather than the asynchronous transport of the session.
+    headers = {}
+    with mock.patch.object(creds, "before_request") as before_request:
+        await async_creds.before_request(mock.sentinel.async_request, "GET", "https://example.com", headers)
+    before_request.assert_called_once_with(mock.ANY, "GET", "https://example.com", headers)
+    assert isinstance(before_request.call_args[0][0], google.auth.transport.requests.Request)
+
+    with mock.patch.object(creds, "refresh") as refresh:
+        await async_creds.refresh(mock.sentinel.async_request)
+    refresh.assert_called_once_with(mock.ANY)
+    assert isinstance(refresh.call_args[0][0], google.auth.transport.requests.Request)
+
+    with mock.patch.object(creds, "apply") as apply:
+        await async_creds.apply(headers, token="token")
+    apply.assert_called_once_with(headers, token="token")
+
+
+def test_async_credentials_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Asynchronous credentials are used as is.
+    creds = ga_credentials_async.AnonymousCredentials()
+    transport = {{ get_client(service, True) }}.get_transport_class("rest_asyncio")(credentials=creds)
+    assert transport._credentials is creds
+    assert transport._async_credentials is creds
+
+
+def test_transport_adc_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Test default credentials are used if not provided.
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
+        adc.return_value = (creds, None)
+        transport = {{ get_client(service, True) }}.get_transport_class("rest_asyncio")()
+        adc.assert_called_once()
+    assert transport._credentials is creds
+    assert isinstance(transport._async_credentials, ga_credentials_async.Credentials)
+
+{% if service.has_lro %}
+
+def test_{{ service.name|snake_case }}_rest_asyncio_lro_client_sync_credentials():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = {{ get_client(service, True) }}(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest_asyncio",
+    )
+    transport = client.transport
+
+    # Ensure that the operations client uses the adapted credentials.
+    assert isinstance(transport.operations_client, operations_v1.AsyncOperationsRestClient)
+    assert transport.operations_client.transport._credentials is transport._async_credentials
+{% endif %}{# service.has_lro #}
+
+{% endmacro %}
+
 {# get_await_prefix sets an "await" keyword
  # to a method call if is_async=True.
 #}
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
index 4d629a5a8443..53e81849a64e 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
@@ -23,7 +23,10 @@
 except ImportError as e:  # pragma: NO COVER
     raise ImportError("`rest_asyncio` transport requires the library to be installed with the `async_rest` extra. Install the library with the `async_rest` extra using `pip install google-cloud-redis[async_rest]`") from e
 
+import asyncio
+from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
+import google.auth.transport.requests  # type: ignore
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
@@ -736,6 +739,35 @@ class AsyncCloudRedisRestStub:
     _host: str
     _interceptor: AsyncCloudRedisRestInterceptor
 
+
+class _SyncToAsyncCredentialsAdapter(ga_credentials_async.Credentials):
+    """Adapts synchronous credentials to the asynchronous credentials interface.
+
+    ``AsyncAuthorizedSession`` only accepts :class:`google.auth.aio.credentials.Credentials`.
+    This adapter allows synchronous :class:`google.auth.credentials.Credentials`
+    (e.g. application default credentials) to be used with the asynchronous
+    REST transport by delegating to the wrapped credentials. Blocking operations,
+    i.e. refreshing the access token, are executed in a worker thread so that the
+    event loop is not blocked.
+    """
+
+    def __init__(self, credentials: ga_credentials.Credentials):
+        super().__init__()
+        self._credentials = credentials
+        # Synchronous credentials require a synchronous transport to refresh the access token.
+        self._request = google.auth.transport.requests.Request()
+
+    async def apply(self, headers, token=None):
+        self._credentials.apply(headers, token=token)
+
+    async def refresh(self, request):
+        # `request` is an asynchronous transport which cannot be used by synchronous credentials.
+        await asyncio.to_thread(self._credentials.refresh, self._request)
+
+    async def before_request(self, request, method, url, headers):
+        await asyncio.to_thread(self._credentials.before_request, self._request, method, url, headers)
+
+
 class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     """Asynchronous REST backend transport for CloudRedis.
 
@@ -770,7 +802,7 @@ class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     def __init__(self,
             *,
             host: str = 'redis.googleapis.com',
-            credentials: Optional[ga_credentials_async.Credentials] = None,
+            credentials: Optional[Union[ga_credentials.Credentials, ga_credentials_async.Credentials]] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             url_scheme: str = 'https',
             interceptor: Optional[AsyncCloudRedisRestInterceptor] = None,
@@ -784,11 +816,13 @@ def __init__(self,
         Args:
             host (Optional[str]):
                  The hostname to connect to (default: 'redis.googleapis.com').
-            credentials (Optional[google.auth.aio.credentials.Credentials]): The
+            credentials (Optional[Union[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
-                credentials from the environment.
+                credentials from the environment. Synchronous credentials are
+                adapted for use with the asynchronous transport; any blocking
+                token refresh is performed in a worker thread.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):
                 The client info used to send a user-agent string along with
                 API requests. If ``None``, then default info will be used.
@@ -809,7 +843,15 @@ def __init__(self,
             url_scheme=url_scheme,
             api_audience=None
         )
-        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
+        # The credentials provided to (or resolved by) the base transport may be synchronous,
+        # e.g. when they are obtained from the environment. Adapt them for use with the
+        # asynchronous session while leaving `self._credentials` untouched.
+        self._async_credentials = (
+            _SyncToAsyncCredentialsAdapter(self._credentials)
+            if isinstance(self._credentials, ga_credentials.Credentials)
+            else self._credentials
+        )
+        self._session = AsyncAuthorizedSession(self._async_credentials)  # type: ignore
         self._interceptor = interceptor or AsyncCloudRedisRestInterceptor()
         self._wrap_with_kind = True
         self._prep_wrapped_messages(client_info)
@@ -2412,8 +2454,8 @@ def operations_client(self) -> AsyncOperationsRestClient:
 
             rest_transport = operations_v1.AsyncOperationsRestTransport(  # type: ignore
                     host=self._host,
-                    # use the credentials which are saved
-                    credentials=self._credentials,  # type: ignore
+                    # use the credentials which are saved, adapted for asynchronous use
+                    credentials=self._async_credentials,
                     http_options=http_options,
                     path_prefix="v1"
             )
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 6bd8b8b5009c..38ef0b9182d7 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -69,6 +69,7 @@
 from google.oauth2 import service_account
 import google.api_core.operation_async as operation_async  # type: ignore
 import google.auth
+import google.auth.transport.requests
 import google.protobuf.duration_pb2 as duration_pb2  # type: ignore
 import google.protobuf.empty_pb2 as empty_pb2  # type: ignore
 import google.protobuf.field_mask_pb2 as field_mask_pb2  # type: ignore
@@ -197,8 +198,11 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
+    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_info(client_class, transport_name):
+    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_info') as factory:
         factory.return_value = creds
@@ -236,8 +240,11 @@ def test_cloud_redis_client_service_account_always_use_jwt(transport_class, tran
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
+    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_file(client_class, transport_name):
+    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_file') as factory:
         factory.return_value = creds
@@ -11258,6 +11265,76 @@ def test_unsupported_parameter_rest_asyncio():
     )
 
 
+@pytest.mark.asyncio
+async def test_sync_credentials_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Synchronous credentials are accepted, consistent with every other transport.
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(transports.rest_asyncio, "AsyncAuthorizedSession", autospec=True) as session_class:
+        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
+    # The provided credentials are preserved on the transport.
+    assert transport._credentials is creds
+    # The asynchronous session is created with an adapted version of the credentials.
+    async_creds = transport._async_credentials
+    assert isinstance(async_creds, ga_credentials_async.Credentials)
+    session_class.assert_called_once_with(async_creds)
+
+    # The adapted credentials delegate to the synchronous credentials using a
+    # synchronous transport, rather than the asynchronous transport of the session.
+    headers = {}
+    with mock.patch.object(creds, "before_request") as before_request:
+        await async_creds.before_request(mock.sentinel.async_request, "GET", "https://example.com", headers)
+    before_request.assert_called_once_with(mock.ANY, "GET", "https://example.com", headers)
+    assert isinstance(before_request.call_args[0][0], google.auth.transport.requests.Request)
+
+    with mock.patch.object(creds, "refresh") as refresh:
+        await async_creds.refresh(mock.sentinel.async_request)
+    refresh.assert_called_once_with(mock.ANY)
+    assert isinstance(refresh.call_args[0][0], google.auth.transport.requests.Request)
+
+    with mock.patch.object(creds, "apply") as apply:
+        await async_creds.apply(headers, token="token")
+    apply.assert_called_once_with(headers, token="token")
+
+
+def test_async_credentials_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Asynchronous credentials are used as is.
+    creds = ga_credentials_async.AnonymousCredentials()
+    transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
+    assert transport._credentials is creds
+    assert transport._async_credentials is creds
+
+
+def test_transport_adc_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Test default credentials are used if not provided.
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
+        adc.return_value = (creds, None)
+        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")()
+        adc.assert_called_once()
+    assert transport._credentials is creds
+    assert isinstance(transport._async_credentials, ga_credentials_async.Credentials)
+
+
+def test_cloud_redis_rest_asyncio_lro_client_sync_credentials():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest_asyncio",
+    )
+    transport = client.transport
+
+    # Ensure that the operations client uses the adapted credentials.
+    assert isinstance(transport.operations_client, operations_v1.AsyncOperationsRestClient)
+    assert transport.operations_client.transport._credentials is transport._async_credentials
+
+
 def test_transport_grpc_default():
     # A client should use the gRPC transport by default.
     client = CloudRedisClient(
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
index 960d9639a214..da2d892e8ed9 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
@@ -23,7 +23,10 @@
 except ImportError as e:  # pragma: NO COVER
     raise ImportError("`rest_asyncio` transport requires the library to be installed with the `async_rest` extra. Install the library with the `async_rest` extra using `pip install google-cloud-redis[async_rest]`") from e
 
+import asyncio
+from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
+import google.auth.transport.requests  # type: ignore
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
@@ -472,6 +475,35 @@ class AsyncCloudRedisRestStub:
     _host: str
     _interceptor: AsyncCloudRedisRestInterceptor
 
+
+class _SyncToAsyncCredentialsAdapter(ga_credentials_async.Credentials):
+    """Adapts synchronous credentials to the asynchronous credentials interface.
+
+    ``AsyncAuthorizedSession`` only accepts :class:`google.auth.aio.credentials.Credentials`.
+    This adapter allows synchronous :class:`google.auth.credentials.Credentials`
+    (e.g. application default credentials) to be used with the asynchronous
+    REST transport by delegating to the wrapped credentials. Blocking operations,
+    i.e. refreshing the access token, are executed in a worker thread so that the
+    event loop is not blocked.
+    """
+
+    def __init__(self, credentials: ga_credentials.Credentials):
+        super().__init__()
+        self._credentials = credentials
+        # Synchronous credentials require a synchronous transport to refresh the access token.
+        self._request = google.auth.transport.requests.Request()
+
+    async def apply(self, headers, token=None):
+        self._credentials.apply(headers, token=token)
+
+    async def refresh(self, request):
+        # `request` is an asynchronous transport which cannot be used by synchronous credentials.
+        await asyncio.to_thread(self._credentials.refresh, self._request)
+
+    async def before_request(self, request, method, url, headers):
+        await asyncio.to_thread(self._credentials.before_request, self._request, method, url, headers)
+
+
 class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     """Asynchronous REST backend transport for CloudRedis.
 
@@ -506,7 +538,7 @@ class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     def __init__(self,
             *,
             host: str = 'redis.googleapis.com',
-            credentials: Optional[ga_credentials_async.Credentials] = None,
+            credentials: Optional[Union[ga_credentials.Credentials, ga_credentials_async.Credentials]] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             url_scheme: str = 'https',
             interceptor: Optional[AsyncCloudRedisRestInterceptor] = None,
@@ -520,11 +552,13 @@ def __init__(self,
         Args:
             host (Optional[str]):
                  The hostname to connect to (default: 'redis.googleapis.com').
-            credentials (Optional[google.auth.aio.credentials.Credentials]): The
+            credentials (Optional[Union[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
-                credentials from the environment.
+                credentials from the environment. Synchronous credentials are
+                adapted for use with the asynchronous transport; any blocking
+                token refresh is performed in a worker thread.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):
                 The client info used to send a user-agent string along with
                 API requests. If ``None``, then default info will be used.
@@ -545,7 +579,15 @@ def __init__(self,
             url_scheme=url_scheme,
             api_audience=None
         )
-        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
+        # The credentials provided to (or resolved by) the base transport may be synchronous,
+        # e.g. when they are obtained from the environment. Adapt them for use with the
+        # asynchronous session while leaving `self._credentials` untouched.
+        self._async_credentials = (
+            _SyncToAsyncCredentialsAdapter(self._credentials)
+            if isinstance(self._credentials, ga_credentials.Credentials)
+            else self._credentials
+        )
+        self._session = AsyncAuthorizedSession(self._async_credentials)  # type: ignore
         self._interceptor = interceptor or AsyncCloudRedisRestInterceptor()
         self._wrap_with_kind = True
         self._prep_wrapped_messages(client_info)
@@ -1324,8 +1366,8 @@ def operations_client(self) -> AsyncOperationsRestClient:
 
             rest_transport = operations_v1.AsyncOperationsRestTransport(  # type: ignore
                     host=self._host,
-                    # use the credentials which are saved
-                    credentials=self._credentials,  # type: ignore
+                    # use the credentials which are saved, adapted for asynchronous use
+                    credentials=self._async_credentials,
                     http_options=http_options,
                     path_prefix="v1"
             )
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 44a69d3d2277..49b5e2f614ef 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -69,6 +69,7 @@
 from google.oauth2 import service_account
 import google.api_core.operation_async as operation_async  # type: ignore
 import google.auth
+import google.auth.transport.requests
 import google.protobuf.duration_pb2 as duration_pb2  # type: ignore
 import google.protobuf.empty_pb2 as empty_pb2  # type: ignore
 import google.protobuf.field_mask_pb2 as field_mask_pb2  # type: ignore
@@ -197,8 +198,11 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
+    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_info(client_class, transport_name):
+    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_info') as factory:
         factory.return_value = creds
@@ -236,8 +240,11 @@ def test_cloud_redis_client_service_account_always_use_jwt(transport_class, tran
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
+    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_file(client_class, transport_name):
+    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_file') as factory:
         factory.return_value = creds
@@ -6502,6 +6509,76 @@ def test_unsupported_parameter_rest_asyncio():
     )
 
 
+@pytest.mark.asyncio
+async def test_sync_credentials_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Synchronous credentials are accepted, consistent with every other transport.
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(transports.rest_asyncio, "AsyncAuthorizedSession", autospec=True) as session_class:
+        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
+    # The provided credentials are preserved on the transport.
+    assert transport._credentials is creds
+    # The asynchronous session is created with an adapted version of the credentials.
+    async_creds = transport._async_credentials
+    assert isinstance(async_creds, ga_credentials_async.Credentials)
+    session_class.assert_called_once_with(async_creds)
+
+    # The adapted credentials delegate to the synchronous credentials using a
+    # synchronous transport, rather than the asynchronous transport of the session.
+    headers = {}
+    with mock.patch.object(creds, "before_request") as before_request:
+        await async_creds.before_request(mock.sentinel.async_request, "GET", "https://example.com", headers)
+    before_request.assert_called_once_with(mock.ANY, "GET", "https://example.com", headers)
+    assert isinstance(before_request.call_args[0][0], google.auth.transport.requests.Request)
+
+    with mock.patch.object(creds, "refresh") as refresh:
+        await async_creds.refresh(mock.sentinel.async_request)
+    refresh.assert_called_once_with(mock.ANY)
+    assert isinstance(refresh.call_args[0][0], google.auth.transport.requests.Request)
+
+    with mock.patch.object(creds, "apply") as apply:
+        await async_creds.apply(headers, token="token")
+    apply.assert_called_once_with(headers, token="token")
+
+
+def test_async_credentials_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Asynchronous credentials are used as is.
+    creds = ga_credentials_async.AnonymousCredentials()
+    transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
+    assert transport._credentials is creds
+    assert transport._async_credentials is creds
+
+
+def test_transport_adc_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    # Test default credentials are used if not provided.
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
+        adc.return_value = (creds, None)
+        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")()
+        adc.assert_called_once()
+    assert transport._credentials is creds
+    assert isinstance(transport._async_credentials, ga_credentials_async.Credentials)
+
+
+def test_cloud_redis_rest_asyncio_lro_client_sync_credentials():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest_asyncio",
+    )
+    transport = client.transport
+
+    # Ensure that the operations client uses the adapted credentials.
+    assert isinstance(transport.operations_client, operations_v1.AsyncOperationsRestClient)
+    assert transport.operations_client.transport._credentials is transport._async_credentials
+
+
 def test_transport_grpc_default():
     # A client should use the gRPC transport by default.
     client = CloudRedisClient(

From b997aed7cccd5f2f32628c23749b160d896007e3 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 17:20:09 -0700
Subject: [PATCH 02/16] added adapter to auth

---
 .../google/auth/aio/transport/sessions.py     |  75 ++++++++++--
 .../tests/transport/aio/test_sessions.py      | 113 +++++++++++++++++-
 2 files changed, 178 insertions(+), 10 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index d2375604b127..2221adca8600 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -24,6 +24,8 @@
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Mapping, Optional, Union
 
+import google.auth.credentials
+import google.auth.transport
 import google.auth.transport._mtls_helper
 from google.auth import _exponential_backoff, exceptions
 from google.auth.aio import transport
@@ -104,6 +106,56 @@ async def with_timeout(coro):
         _remaining_time()
 
 
+class _SyncCredentialsAdapter(Credentials):
+    """Adapts synchronous credentials to the asynchronous credentials interface.
+
+    :class:`AsyncAuthorizedSession` wraps :class:`google.auth.credentials.Credentials`
+    (e.g. application default credentials) with this adapter so that they can be
+    used with an asynchronous transport. Calls are delegated to the wrapped
+    credentials using a synchronous transport, and blocking calls such as
+    refreshing the access token run in a worker thread so that the event loop
+    is not blocked.
+
+    Args:
+        credentials (google.auth.credentials.Credentials): The synchronous
+            credentials to adapt.
+    """
+
+    def __init__(self, credentials: google.auth.credentials.Credentials):
+        super().__init__()
+        self._credentials = credentials
+        self._sync_request: Optional[google.auth.transport.Request] = None
+
+    def _get_sync_request(self) -> google.auth.transport.Request:
+        """Returns the synchronous transport used to call the wrapped credentials.
+
+        Synchronous credentials cannot use the asynchronous transport of the
+        session, so a synchronous transport is created on first use and reused.
+        """
+        if self._sync_request is None:
+            # Imported lazily because `requests` is an optional dependency of
+            # google-auth. It is installed alongside `aiohttp` by the `aiohttp` extra.
+            from google.auth.transport import requests as sync_requests
+
+            self._sync_request = sync_requests.Request()
+        return self._sync_request
+
+    async def apply(self, headers, token=None):
+        self._credentials.apply(headers, token=token)
+
+    async def refresh(self, request):
+        await asyncio.to_thread(self._credentials.refresh, self._get_sync_request())
+
+    async def before_request(self, request, method, url, headers):
+        await asyncio.to_thread(
+            self._credentials.before_request,
+            self._get_sync_request(),
+            method,
+            url,
+            headers,
+        )
+
+
 class AsyncAuthorizedSession:
     """This is an asynchronous implementation of :class:`google.auth.requests.AuthorizedSession` class.
     We utilize an instance of a class that implements :class:`google.auth.aio.transport.Request` configured
@@ -126,8 +178,12 @@ class AsyncAuthorizedSession:
     credentials' headers to the request and refreshing credentials as needed.
 
     Args:
-        credentials (google.auth.aio.credentials.Credentials):
-            The credentials to add to the request.
+        credentials (Union[google.auth.aio.credentials.Credentials, google.auth.credentials.Credentials]):
+            The credentials to add to the request. Synchronous credentials
+            (e.g. application default credentials) are adapted to the
+            asynchronous credentials interface: they are invoked using a
+            synchronous transport in a worker thread so that the event loop
+            is not blocked.
         auth_request (Optional[google.auth.aio.transport.Request]):
             An instance of a class that implements
             :class:`~google.auth.aio.transport.Request` used to make requests
@@ -139,17 +195,22 @@ class AsyncAuthorizedSession:
         - google.auth.exceptions.TransportError: If `auth_request` is `None`
             and the external package `aiohttp` is not installed.
         - google.auth.exceptions.InvalidType: If the provided credentials are
-            not of type `google.auth.aio.credentials.Credentials`.
+            not of type `google.auth.aio.credentials.Credentials` or
+            `google.auth.credentials.Credentials`.
     """
 
     def __init__(
-        self, credentials: Credentials, auth_request: Optional[transport.Request] = None
+        self,
+        credentials: Union[Credentials, google.auth.credentials.Credentials],
+        auth_request: Optional[transport.Request] = None,
     ):
-        if not isinstance(credentials, Credentials):
+        if isinstance(credentials, google.auth.credentials.Credentials):
+            credentials = _SyncCredentialsAdapter(credentials)
+        elif not isinstance(credentials, Credentials):
             raise exceptions.InvalidType(
-                f"The configured credentials of type {type(credentials)} are invalid and must be of type `google.auth.aio.credentials.Credentials`"
+                f"The configured credentials of type {type(credentials)} are invalid and must be of type `google.auth.aio.credentials.Credentials` or `google.auth.credentials.Credentials`"
             )
-        self._credentials = credentials
+        self._credentials: Credentials = credentials
         _auth_request = auth_request
         if not _auth_request and AIOHTTP_INSTALLED:
             _auth_request = AiohttpRequest()
diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index 69793fe49f5b..73c063e02607 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -13,13 +13,17 @@
 # limitations under the License.
 
 import asyncio
+import http.client as http_client
+import threading
 from typing import AsyncGenerator
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest  # type: ignore
 from aioresponses import aioresponses  # type: ignore
 
-from google.auth.aio.credentials import AnonymousCredentials
+import google.auth.credentials
+import google.auth.transport.requests
+from google.auth.aio.credentials import AnonymousCredentials, Credentials
 from google.auth.aio.transport import (
     _DEFAULT_TIMEOUT_SECONDS,
     DEFAULT_MAX_RETRY_ATTEMPTS,
@@ -203,9 +207,71 @@ async def test_constructor_raises_incorrect_credentials_error(self):
             sessions.AsyncAuthorizedSession(credentials)
 
         exc.match(
-            f"The configured credentials of type {type(credentials)} are invalid and must be of type `google.auth.aio.credentials.Credentials`"
+            f"The configured credentials of type {type(credentials)} are invalid and must be of type `google.auth.aio.credentials.Credentials` or `google.auth.credentials.Credentials`"
         )
 
+    @pytest.mark.asyncio
+    async def test_constructor_with_sync_credentials(self):
+        sync_credentials = google.auth.credentials.AnonymousCredentials()
+        authed_session = sessions.AsyncAuthorizedSession(
+            sync_credentials, auth_request=MockRequest()
+        )
+
+        # Synchronous credentials are adapted to the asynchronous credentials interface.
+        assert isinstance(authed_session._credentials, Credentials)
+        assert isinstance(authed_session._credentials, sessions._SyncCredentialsAdapter)
+        assert authed_session._credentials._credentials is sync_credentials
+        await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_request_with_sync_credentials_success(self, mocked_content):
+        sync_credentials = Mock(spec=google.auth.credentials.Credentials)
+        mocked_response = MockResponse(
+            status_code=http_client.OK,
+            headers={"Content-Type": "application/json"},
+            content=mocked_content,
+        )
+        auth_request = MockRequest(mocked_response)
+        authed_session = sessions.AsyncAuthorizedSession(sync_credentials, auth_request)
+
+        response = await authed_session.request(
+            "GET", self.TEST_URL, headers={"x-test": "value"}
+        )
+
+        assert response.status_code == http_client.OK
+        assert await response.read() == b"Cavefish have no sight."
+        # The synchronous credentials are invoked with a synchronous transport rather
+        # than the asynchronous transport of the session.
+        sync_credentials.before_request.assert_called_once()
+        request, method, url, headers = sync_credentials.before_request.call_args.args
+        assert isinstance(request, google.auth.transport.requests.Request)
+        assert method == "GET"
+        assert url == self.TEST_URL
+        assert headers == {"x-test": "value"}
+        await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_request_with_sync_credentials_refreshes_on_unauthorized(self):
+        sync_credentials = Mock(spec=google.auth.credentials.Credentials)
+        unauthorized_response = MockResponse(status_code=http_client.UNAUTHORIZED)
+        ok_response = MockResponse(status_code=http_client.OK)
+        auth_request = AsyncMock(side_effect=[unauthorized_response, ok_response])
+        authed_session = sessions.AsyncAuthorizedSession(
+            sync_credentials, auth_request=auth_request
+        )
+
+        response = await authed_session.request("GET", self.TEST_URL)
+
+        assert response is ok_response
+        assert auth_request.call_count == 2
+        assert unauthorized_response._close
+        sync_credentials.refresh.assert_called_once()
+        (refresh_request,) = sync_credentials.refresh.call_args.args
+        assert isinstance(refresh_request, google.auth.transport.requests.Request)
+        # The same synchronous transport is reused for every call to the credentials.
+        assert refresh_request is sync_credentials.before_request.call_args.args[0]
+        await authed_session.close()
+
     @pytest.mark.asyncio
     async def test_request_default_auth_request_success(self):
         with aioresponses() as m:
@@ -368,3 +434,44 @@ def test_mock_request_clone():
     request = MockRequest()
     cloned = request._clone()
     assert cloned is request
+
+
+class TestSyncCredentialsAdapter(object):
+    TEST_URL = "http://example.com/"
+
+    @pytest.mark.asyncio
+    async def test_delegates_to_sync_credentials(self):
+        sync_credentials = Mock(spec=google.auth.credentials.Credentials)
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+        headers = {}
+
+        await adapter.before_request(Mock(), "GET", self.TEST_URL, headers)
+        await adapter.refresh(Mock())
+        await adapter.apply(headers, token="token")
+
+        # A synchronous transport is created lazily and reused for every call.
+        sync_request = sync_credentials.before_request.call_args.args[0]
+        assert isinstance(sync_request, google.auth.transport.requests.Request)
+        sync_credentials.before_request.assert_called_once_with(
+            sync_request, "GET", self.TEST_URL, headers
+        )
+        sync_credentials.refresh.assert_called_once_with(sync_request)
+        sync_credentials.apply.assert_called_once_with(headers, token="token")
+
+    @pytest.mark.asyncio
+    async def test_blocking_calls_run_off_the_event_loop_thread(self):
+        sync_credentials = Mock(spec=google.auth.credentials.Credentials)
+        thread_ids = []
+        sync_credentials.before_request.side_effect = lambda *args: thread_ids.append(
+            threading.get_ident()
+        )
+        sync_credentials.refresh.side_effect = lambda *args: thread_ids.append(
+            threading.get_ident()
+        )
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+
+        await adapter.before_request(Mock(), "GET", self.TEST_URL, {})
+        await adapter.refresh(Mock())
+
+        assert len(thread_ids) == 2
+        assert threading.get_ident() not in thread_ids

From 9cea1deb14dca517b477fd6ebfee4475ede2d158 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 17:36:45 -0700
Subject: [PATCH 03/16] reverted generator changes

---
 .../%service/transports/rest_asyncio.py.j2    | 58 +++-----------
 .../%name_%version/%sub/test_%service.py.j2   | 18 -----
 .../gapic/%name_%version/%sub/test_macros.j2  | 76 ------------------
 .../cloud_redis/transports/rest_asyncio.py    | 54 ++-----------
 .../unit/gapic/redis_v1/test_cloud_redis.py   | 77 -------------------
 .../cloud_redis/transports/rest_asyncio.py    | 54 ++-----------
 .../unit/gapic/redis_v1/test_cloud_redis.py   | 77 -------------------
 7 files changed, 22 insertions(+), 392 deletions(-)

diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
index dc09bdb97628..0f79d6e1ffef 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
@@ -17,10 +17,7 @@ try:
 except ImportError as e:  # pragma: NO COVER
     raise ImportError("`rest_asyncio` transport requires the library to be installed with the `async_rest` extra. Install the library with the `async_rest` extra using `pip install {{ api.naming.warehouse_package_name }}[async_rest]`") from e
 
-import asyncio
-from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
-import google.auth.transport.requests  # type: ignore
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
@@ -96,35 +93,6 @@ class Async{{service.name}}RestStub:
     _host: str
     _interceptor: Async{{service.name}}RestInterceptor
 
-
-class _SyncToAsyncCredentialsAdapter(ga_credentials_async.Credentials):
-    """Adapts synchronous credentials to the asynchronous credentials interface.
-
-    ``AsyncAuthorizedSession`` only accepts :class:`google.auth.aio.credentials.Credentials`.
-    This adapter allows synchronous :class:`google.auth.credentials.Credentials`
-    (e.g. application default credentials) to be used with the asynchronous
-    REST transport by delegating to the wrapped credentials. Blocking operations,
-    i.e. refreshing the access token, are executed in a worker thread so that the
-    event loop is not blocked.
-    """
-
-    def __init__(self, credentials: ga_credentials.Credentials):
-        super().__init__()
-        self._credentials = credentials
-        # Synchronous credentials require a synchronous transport to refresh the access token.
-        self._request = google.auth.transport.requests.Request()
-
-    async def apply(self, headers, token=None):
-        self._credentials.apply(headers, token=token)
-
-    async def refresh(self, request):
-        # `request` is an asynchronous transport which cannot be used by synchronous credentials.
-        await asyncio.to_thread(self._credentials.refresh, self._request)
-
-    async def before_request(self, request, method, url, headers):
-        await asyncio.to_thread(self._credentials.before_request, self._request, method, url, headers)
-
-
 class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
     """Asynchronous REST backend transport for {{ service.name }}.
 
@@ -139,7 +107,7 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
     def __init__(self,
             *,
             host: str{% if service.host %} = '{{ service.host }}'{% endif %},
-            credentials: Optional[Union[ga_credentials.Credentials, ga_credentials_async.Credentials]] = None,
+            credentials: Optional[ga_credentials_async.Credentials] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             url_scheme: str = 'https',
             interceptor: Optional[Async{{ service.name }}RestInterceptor] = None,
@@ -155,13 +123,11 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
         Args:
             host ({% if service.host %}Optional[str]{% else %}str{% endif %}):
                 {{ ' ' }}The hostname to connect to {% if service.host %}(default: '{{ service.host }}'){% endif %}.
-            credentials (Optional[Union[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]]): The
+            credentials (Optional[google.auth.aio.credentials.Credentials]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
-                credentials from the environment. Synchronous credentials are
-                adapted for use with the asynchronous transport; any blocking
-                token refresh is performed in a worker thread.
+                credentials from the environment.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):
                 The client info used to send a user-agent string along with
                 API requests. If ``None``, then default info will be used.
@@ -182,19 +148,11 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             url_scheme=url_scheme,
             api_audience=None
         )
-        # The credentials provided to (or resolved by) the base transport may be synchronous,
-        # e.g. when they are obtained from the environment. Adapt them for use with the
-        # asynchronous session while leaving `self._credentials` untouched.
-        self._async_credentials = (
-            _SyncToAsyncCredentialsAdapter(self._credentials)
-            if isinstance(self._credentials, ga_credentials.Credentials)
-            else self._credentials
-        )
         {# Note: Type for creds is ignored because of incorrect type hint for creds in the client layer.
          # TODO(https://github.com/googleapis/gapic-generator-python/issues/2177): Remove `# type: ignore` once
          # we update the type hints for credentials to include asynchronous credentials in the client layer.
         #}
-        self._session = AsyncAuthorizedSession(self._async_credentials)  # type: ignore
+        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
         self._interceptor = interceptor or Async{{ service.name }}RestInterceptor()
         self._wrap_with_kind = True
         self._prep_wrapped_messages(client_info)
@@ -342,8 +300,12 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
 
             rest_transport = operations_v1.AsyncOperationsRestTransport(  # type: ignore
                     host=self._host,
-                    # use the credentials which are saved, adapted for asynchronous use
-                    credentials=self._async_credentials,
+                    # use the credentials which are saved
+                    {# Note: Type for creds is ignored because of incorrect type hint for creds in the client layer.
+                     # TODO(https://github.com/googleapis/gapic-generator-python/issues/2177): Remove `# type: ignore` once
+                     # we update the type hints for credentials to include asynchronous credentials in the client layer.
+                    #}
+                    credentials=self._credentials,  # type: ignore
                     http_options=http_options,
                     path_prefix="{{ service.client_package_version }}"
             )
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index 9804796f1dd4..d6880c73605a 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -55,9 +55,6 @@ except ImportError: # pragma: NO COVER
 {% filter sort_lines %}
 import google.auth
 from google.auth import credentials as ga_credentials
-{% if rest_async_io_enabled %}
-import google.auth.transport.requests
-{% endif %}
 
 
 from google.auth.exceptions import MutualTLSChannelError
@@ -284,16 +281,9 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
     {% endif %}
     {% if 'rest' in opts.transport %}
     ({{ service.client_name }}, "rest"),
-    {% if rest_async_io_enabled %}
-    ({{ service.async_client_name }}, "rest_asyncio"),
-    {% endif %}
     {% endif %}
 ])
 def test_{{ service.client_name|snake_case }}_from_service_account_info(client_class, transport_name):
-    {% if rest_async_io_enabled %}
-    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    {% endif %}
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_info') as factory:
         factory.return_value = creds
@@ -342,16 +332,9 @@ def test_{{ service.client_name|snake_case }}_service_account_always_use_jwt(tra
     {% endif %}
     {% if 'rest' in opts.transport %}
     ({{ service.client_name }}, "rest"),
-    {% if rest_async_io_enabled %}
-    ({{ service.async_client_name }}, "rest_asyncio"),
-    {% endif %}
     {% endif %}
 ])
 def test_{{ service.client_name|snake_case }}_from_service_account_file(client_class, transport_name):
-    {% if rest_async_io_enabled %}
-    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    {% endif %}
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_file') as factory:
         factory.return_value = creds
@@ -1134,7 +1117,6 @@ def test_transport_adc(transport_class):
 {% if rest_async_io_enabled %}
 {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove the below macro when async rest is GA and supports all required parameters. #}
 {{ test_macros.async_rest_unsupported_params_test(service) }}
-{{ test_macros.async_rest_credentials_test(service) }}
 {% endif %}
 {% if 'grpc' in opts.transport %}
 def test_transport_grpc_default():
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
index 1169d797f2fd..05c60cbdc9c3 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
@@ -1688,82 +1688,6 @@ def test_unsupported_parameter_rest_asyncio():
 
 {% endmacro %}
 
-{# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Fold these tests into the generic
- # transport tests (e.g. `test_transport_adc`) when async rest is GA. #}
-{% macro async_rest_credentials_test(service) %}
-@pytest.mark.asyncio
-async def test_sync_credentials_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Synchronous credentials are accepted, consistent with every other transport.
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(transports.rest_asyncio, "AsyncAuthorizedSession", autospec=True) as session_class:
-        transport = {{ get_client(service, True) }}.get_transport_class("rest_asyncio")(credentials=creds)
-    # The provided credentials are preserved on the transport.
-    assert transport._credentials is creds
-    # The asynchronous session is created with an adapted version of the credentials.
-    async_creds = transport._async_credentials
-    assert isinstance(async_creds, ga_credentials_async.Credentials)
-    session_class.assert_called_once_with(async_creds)
-
-    # The adapted credentials delegate to the synchronous credentials using a
-    # synchronous transport, rather than the asynchronous transport of the session.
-    headers = {}
-    with mock.patch.object(creds, "before_request") as before_request:
-        await async_creds.before_request(mock.sentinel.async_request, "GET", "https://example.com", headers)
-    before_request.assert_called_once_with(mock.ANY, "GET", "https://example.com", headers)
-    assert isinstance(before_request.call_args[0][0], google.auth.transport.requests.Request)
-
-    with mock.patch.object(creds, "refresh") as refresh:
-        await async_creds.refresh(mock.sentinel.async_request)
-    refresh.assert_called_once_with(mock.ANY)
-    assert isinstance(refresh.call_args[0][0], google.auth.transport.requests.Request)
-
-    with mock.patch.object(creds, "apply") as apply:
-        await async_creds.apply(headers, token="token")
-    apply.assert_called_once_with(headers, token="token")
-
-
-def test_async_credentials_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Asynchronous credentials are used as is.
-    creds = ga_credentials_async.AnonymousCredentials()
-    transport = {{ get_client(service, True) }}.get_transport_class("rest_asyncio")(credentials=creds)
-    assert transport._credentials is creds
-    assert transport._async_credentials is creds
-
-
-def test_transport_adc_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Test default credentials are used if not provided.
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
-        adc.return_value = (creds, None)
-        transport = {{ get_client(service, True) }}.get_transport_class("rest_asyncio")()
-        adc.assert_called_once()
-    assert transport._credentials is creds
-    assert isinstance(transport._async_credentials, ga_credentials_async.Credentials)
-
-{% if service.has_lro %}
-
-def test_{{ service.name|snake_case }}_rest_asyncio_lro_client_sync_credentials():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    client = {{ get_client(service, True) }}(
-        credentials=ga_credentials.AnonymousCredentials(),
-        transport="rest_asyncio",
-    )
-    transport = client.transport
-
-    # Ensure that the operations client uses the adapted credentials.
-    assert isinstance(transport.operations_client, operations_v1.AsyncOperationsRestClient)
-    assert transport.operations_client.transport._credentials is transport._async_credentials
-{% endif %}{# service.has_lro #}
-
-{% endmacro %}
-
 {# get_await_prefix sets an "await" keyword
  # to a method call if is_async=True.
 #}
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
index 53e81849a64e..4d629a5a8443 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
@@ -23,10 +23,7 @@
 except ImportError as e:  # pragma: NO COVER
     raise ImportError("`rest_asyncio` transport requires the library to be installed with the `async_rest` extra. Install the library with the `async_rest` extra using `pip install google-cloud-redis[async_rest]`") from e
 
-import asyncio
-from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
-import google.auth.transport.requests  # type: ignore
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
@@ -739,35 +736,6 @@ class AsyncCloudRedisRestStub:
     _host: str
     _interceptor: AsyncCloudRedisRestInterceptor
 
-
-class _SyncToAsyncCredentialsAdapter(ga_credentials_async.Credentials):
-    """Adapts synchronous credentials to the asynchronous credentials interface.
-
-    ``AsyncAuthorizedSession`` only accepts :class:`google.auth.aio.credentials.Credentials`.
-    This adapter allows synchronous :class:`google.auth.credentials.Credentials`
-    (e.g. application default credentials) to be used with the asynchronous
-    REST transport by delegating to the wrapped credentials. Blocking operations,
-    i.e. refreshing the access token, are executed in a worker thread so that the
-    event loop is not blocked.
-    """
-
-    def __init__(self, credentials: ga_credentials.Credentials):
-        super().__init__()
-        self._credentials = credentials
-        # Synchronous credentials require a synchronous transport to refresh the access token.
-        self._request = google.auth.transport.requests.Request()
-
-    async def apply(self, headers, token=None):
-        self._credentials.apply(headers, token=token)
-
-    async def refresh(self, request):
-        # `request` is an asynchronous transport which cannot be used by synchronous credentials.
-        await asyncio.to_thread(self._credentials.refresh, self._request)
-
-    async def before_request(self, request, method, url, headers):
-        await asyncio.to_thread(self._credentials.before_request, self._request, method, url, headers)
-
-
 class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     """Asynchronous REST backend transport for CloudRedis.
 
@@ -802,7 +770,7 @@ class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     def __init__(self,
             *,
             host: str = 'redis.googleapis.com',
-            credentials: Optional[Union[ga_credentials.Credentials, ga_credentials_async.Credentials]] = None,
+            credentials: Optional[ga_credentials_async.Credentials] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             url_scheme: str = 'https',
             interceptor: Optional[AsyncCloudRedisRestInterceptor] = None,
@@ -816,13 +784,11 @@ def __init__(self,
         Args:
             host (Optional[str]):
                  The hostname to connect to (default: 'redis.googleapis.com').
-            credentials (Optional[Union[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]]): The
+            credentials (Optional[google.auth.aio.credentials.Credentials]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
-                credentials from the environment. Synchronous credentials are
-                adapted for use with the asynchronous transport; any blocking
-                token refresh is performed in a worker thread.
+                credentials from the environment.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):
                 The client info used to send a user-agent string along with
                 API requests. If ``None``, then default info will be used.
@@ -843,15 +809,7 @@ def __init__(self,
             url_scheme=url_scheme,
             api_audience=None
         )
-        # The credentials provided to (or resolved by) the base transport may be synchronous,
-        # e.g. when they are obtained from the environment. Adapt them for use with the
-        # asynchronous session while leaving `self._credentials` untouched.
-        self._async_credentials = (
-            _SyncToAsyncCredentialsAdapter(self._credentials)
-            if isinstance(self._credentials, ga_credentials.Credentials)
-            else self._credentials
-        )
-        self._session = AsyncAuthorizedSession(self._async_credentials)  # type: ignore
+        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
         self._interceptor = interceptor or AsyncCloudRedisRestInterceptor()
         self._wrap_with_kind = True
         self._prep_wrapped_messages(client_info)
@@ -2454,8 +2412,8 @@ def operations_client(self) -> AsyncOperationsRestClient:
 
             rest_transport = operations_v1.AsyncOperationsRestTransport(  # type: ignore
                     host=self._host,
-                    # use the credentials which are saved, adapted for asynchronous use
-                    credentials=self._async_credentials,
+                    # use the credentials which are saved
+                    credentials=self._credentials,  # type: ignore
                     http_options=http_options,
                     path_prefix="v1"
             )
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 38ef0b9182d7..6bd8b8b5009c 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -69,7 +69,6 @@
 from google.oauth2 import service_account
 import google.api_core.operation_async as operation_async  # type: ignore
 import google.auth
-import google.auth.transport.requests
 import google.protobuf.duration_pb2 as duration_pb2  # type: ignore
 import google.protobuf.empty_pb2 as empty_pb2  # type: ignore
 import google.protobuf.field_mask_pb2 as field_mask_pb2  # type: ignore
@@ -198,11 +197,8 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
-    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_info(client_class, transport_name):
-    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_info') as factory:
         factory.return_value = creds
@@ -240,11 +236,8 @@ def test_cloud_redis_client_service_account_always_use_jwt(transport_class, tran
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
-    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_file(client_class, transport_name):
-    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_file') as factory:
         factory.return_value = creds
@@ -11265,76 +11258,6 @@ def test_unsupported_parameter_rest_asyncio():
     )
 
 
-@pytest.mark.asyncio
-async def test_sync_credentials_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Synchronous credentials are accepted, consistent with every other transport.
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(transports.rest_asyncio, "AsyncAuthorizedSession", autospec=True) as session_class:
-        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
-    # The provided credentials are preserved on the transport.
-    assert transport._credentials is creds
-    # The asynchronous session is created with an adapted version of the credentials.
-    async_creds = transport._async_credentials
-    assert isinstance(async_creds, ga_credentials_async.Credentials)
-    session_class.assert_called_once_with(async_creds)
-
-    # The adapted credentials delegate to the synchronous credentials using a
-    # synchronous transport, rather than the asynchronous transport of the session.
-    headers = {}
-    with mock.patch.object(creds, "before_request") as before_request:
-        await async_creds.before_request(mock.sentinel.async_request, "GET", "https://example.com", headers)
-    before_request.assert_called_once_with(mock.ANY, "GET", "https://example.com", headers)
-    assert isinstance(before_request.call_args[0][0], google.auth.transport.requests.Request)
-
-    with mock.patch.object(creds, "refresh") as refresh:
-        await async_creds.refresh(mock.sentinel.async_request)
-    refresh.assert_called_once_with(mock.ANY)
-    assert isinstance(refresh.call_args[0][0], google.auth.transport.requests.Request)
-
-    with mock.patch.object(creds, "apply") as apply:
-        await async_creds.apply(headers, token="token")
-    apply.assert_called_once_with(headers, token="token")
-
-
-def test_async_credentials_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Asynchronous credentials are used as is.
-    creds = ga_credentials_async.AnonymousCredentials()
-    transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
-    assert transport._credentials is creds
-    assert transport._async_credentials is creds
-
-
-def test_transport_adc_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Test default credentials are used if not provided.
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
-        adc.return_value = (creds, None)
-        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")()
-        adc.assert_called_once()
-    assert transport._credentials is creds
-    assert isinstance(transport._async_credentials, ga_credentials_async.Credentials)
-
-
-def test_cloud_redis_rest_asyncio_lro_client_sync_credentials():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    client = CloudRedisAsyncClient(
-        credentials=ga_credentials.AnonymousCredentials(),
-        transport="rest_asyncio",
-    )
-    transport = client.transport
-
-    # Ensure that the operations client uses the adapted credentials.
-    assert isinstance(transport.operations_client, operations_v1.AsyncOperationsRestClient)
-    assert transport.operations_client.transport._credentials is transport._async_credentials
-
-
 def test_transport_grpc_default():
     # A client should use the gRPC transport by default.
     client = CloudRedisClient(
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
index da2d892e8ed9..960d9639a214 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
@@ -23,10 +23,7 @@
 except ImportError as e:  # pragma: NO COVER
     raise ImportError("`rest_asyncio` transport requires the library to be installed with the `async_rest` extra. Install the library with the `async_rest` extra using `pip install google-cloud-redis[async_rest]`") from e
 
-import asyncio
-from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
-import google.auth.transport.requests  # type: ignore
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
@@ -475,35 +472,6 @@ class AsyncCloudRedisRestStub:
     _host: str
     _interceptor: AsyncCloudRedisRestInterceptor
 
-
-class _SyncToAsyncCredentialsAdapter(ga_credentials_async.Credentials):
-    """Adapts synchronous credentials to the asynchronous credentials interface.
-
-    ``AsyncAuthorizedSession`` only accepts :class:`google.auth.aio.credentials.Credentials`.
-    This adapter allows synchronous :class:`google.auth.credentials.Credentials`
-    (e.g. application default credentials) to be used with the asynchronous
-    REST transport by delegating to the wrapped credentials. Blocking operations,
-    i.e. refreshing the access token, are executed in a worker thread so that the
-    event loop is not blocked.
-    """
-
-    def __init__(self, credentials: ga_credentials.Credentials):
-        super().__init__()
-        self._credentials = credentials
-        # Synchronous credentials require a synchronous transport to refresh the access token.
-        self._request = google.auth.transport.requests.Request()
-
-    async def apply(self, headers, token=None):
-        self._credentials.apply(headers, token=token)
-
-    async def refresh(self, request):
-        # `request` is an asynchronous transport which cannot be used by synchronous credentials.
-        await asyncio.to_thread(self._credentials.refresh, self._request)
-
-    async def before_request(self, request, method, url, headers):
-        await asyncio.to_thread(self._credentials.before_request, self._request, method, url, headers)
-
-
 class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     """Asynchronous REST backend transport for CloudRedis.
 
@@ -538,7 +506,7 @@ class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
     def __init__(self,
             *,
             host: str = 'redis.googleapis.com',
-            credentials: Optional[Union[ga_credentials.Credentials, ga_credentials_async.Credentials]] = None,
+            credentials: Optional[ga_credentials_async.Credentials] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             url_scheme: str = 'https',
             interceptor: Optional[AsyncCloudRedisRestInterceptor] = None,
@@ -552,13 +520,11 @@ def __init__(self,
         Args:
             host (Optional[str]):
                  The hostname to connect to (default: 'redis.googleapis.com').
-            credentials (Optional[Union[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]]): The
+            credentials (Optional[google.auth.aio.credentials.Credentials]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
-                credentials from the environment. Synchronous credentials are
-                adapted for use with the asynchronous transport; any blocking
-                token refresh is performed in a worker thread.
+                credentials from the environment.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):
                 The client info used to send a user-agent string along with
                 API requests. If ``None``, then default info will be used.
@@ -579,15 +545,7 @@ def __init__(self,
             url_scheme=url_scheme,
             api_audience=None
         )
-        # The credentials provided to (or resolved by) the base transport may be synchronous,
-        # e.g. when they are obtained from the environment. Adapt them for use with the
-        # asynchronous session while leaving `self._credentials` untouched.
-        self._async_credentials = (
-            _SyncToAsyncCredentialsAdapter(self._credentials)
-            if isinstance(self._credentials, ga_credentials.Credentials)
-            else self._credentials
-        )
-        self._session = AsyncAuthorizedSession(self._async_credentials)  # type: ignore
+        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
         self._interceptor = interceptor or AsyncCloudRedisRestInterceptor()
         self._wrap_with_kind = True
         self._prep_wrapped_messages(client_info)
@@ -1366,8 +1324,8 @@ def operations_client(self) -> AsyncOperationsRestClient:
 
             rest_transport = operations_v1.AsyncOperationsRestTransport(  # type: ignore
                     host=self._host,
-                    # use the credentials which are saved, adapted for asynchronous use
-                    credentials=self._async_credentials,
+                    # use the credentials which are saved
+                    credentials=self._credentials,  # type: ignore
                     http_options=http_options,
                     path_prefix="v1"
             )
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 49b5e2f614ef..44a69d3d2277 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -69,7 +69,6 @@
 from google.oauth2 import service_account
 import google.api_core.operation_async as operation_async  # type: ignore
 import google.auth
-import google.auth.transport.requests
 import google.protobuf.duration_pb2 as duration_pb2  # type: ignore
 import google.protobuf.empty_pb2 as empty_pb2  # type: ignore
 import google.protobuf.field_mask_pb2 as field_mask_pb2  # type: ignore
@@ -198,11 +197,8 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
-    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_info(client_class, transport_name):
-    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_info') as factory:
         factory.return_value = creds
@@ -240,11 +236,8 @@ def test_cloud_redis_client_service_account_always_use_jwt(transport_class, tran
     (CloudRedisClient, "grpc"),
     (CloudRedisAsyncClient, "grpc_asyncio"),
     (CloudRedisClient, "rest"),
-    (CloudRedisAsyncClient, "rest_asyncio"),
 ])
 def test_cloud_redis_client_from_service_account_file(client_class, transport_name):
-    if transport_name == "rest_asyncio" and not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
     creds = ga_credentials.AnonymousCredentials()
     with mock.patch.object(service_account.Credentials, 'from_service_account_file') as factory:
         factory.return_value = creds
@@ -6509,76 +6502,6 @@ def test_unsupported_parameter_rest_asyncio():
     )
 
 
-@pytest.mark.asyncio
-async def test_sync_credentials_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Synchronous credentials are accepted, consistent with every other transport.
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(transports.rest_asyncio, "AsyncAuthorizedSession", autospec=True) as session_class:
-        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
-    # The provided credentials are preserved on the transport.
-    assert transport._credentials is creds
-    # The asynchronous session is created with an adapted version of the credentials.
-    async_creds = transport._async_credentials
-    assert isinstance(async_creds, ga_credentials_async.Credentials)
-    session_class.assert_called_once_with(async_creds)
-
-    # The adapted credentials delegate to the synchronous credentials using a
-    # synchronous transport, rather than the asynchronous transport of the session.
-    headers = {}
-    with mock.patch.object(creds, "before_request") as before_request:
-        await async_creds.before_request(mock.sentinel.async_request, "GET", "https://example.com", headers)
-    before_request.assert_called_once_with(mock.ANY, "GET", "https://example.com", headers)
-    assert isinstance(before_request.call_args[0][0], google.auth.transport.requests.Request)
-
-    with mock.patch.object(creds, "refresh") as refresh:
-        await async_creds.refresh(mock.sentinel.async_request)
-    refresh.assert_called_once_with(mock.ANY)
-    assert isinstance(refresh.call_args[0][0], google.auth.transport.requests.Request)
-
-    with mock.patch.object(creds, "apply") as apply:
-        await async_creds.apply(headers, token="token")
-    apply.assert_called_once_with(headers, token="token")
-
-
-def test_async_credentials_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Asynchronous credentials are used as is.
-    creds = ga_credentials_async.AnonymousCredentials()
-    transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")(credentials=creds)
-    assert transport._credentials is creds
-    assert transport._async_credentials is creds
-
-
-def test_transport_adc_rest_asyncio():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    # Test default credentials are used if not provided.
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
-        adc.return_value = (creds, None)
-        transport = CloudRedisAsyncClient.get_transport_class("rest_asyncio")()
-        adc.assert_called_once()
-    assert transport._credentials is creds
-    assert isinstance(transport._async_credentials, ga_credentials_async.Credentials)
-
-
-def test_cloud_redis_rest_asyncio_lro_client_sync_credentials():
-    if not HAS_ASYNC_REST_EXTRA:
-        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
-    client = CloudRedisAsyncClient(
-        credentials=ga_credentials.AnonymousCredentials(),
-        transport="rest_asyncio",
-    )
-    transport = client.transport
-
-    # Ensure that the operations client uses the adapted credentials.
-    assert isinstance(transport.operations_client, operations_v1.AsyncOperationsRestClient)
-    assert transport.operations_client.transport._credentials is transport._async_credentials
-
-
 def test_transport_grpc_default():
     # A client should use the gRPC transport by default.
     client = CloudRedisClient(

From ff5dd833dce2d268d5dd65e9c7c41b84b48163bd Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 17:36:58 -0700
Subject: [PATCH 04/16] updated docstring

---
 packages/google-auth/google/auth/aio/transport/sessions.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 2221adca8600..872b17751df2 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -180,10 +180,7 @@ class AsyncAuthorizedSession:
     Args:
         credentials (Union[google.auth.aio.credentials.Credentials, google.auth.credentials.Credentials]):
             The credentials to add to the request. Synchronous credentials
-            (e.g. application default credentials) are adapted to the
-            asynchronous credentials interface: they are invoked using a
-            synchronous transport in a worker thread so that the event loop
-            is not blocked.
+            (e.g. application default credentials) are also supported.
         auth_request (Optional[google.auth.aio.transport.Request]):
             An instance of a class that implements
             :class:`~google.auth.aio.transport.Request` used to make requests

From 1a301176a026603f8e882baff62de86a36785b19 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 18:07:38 -0700
Subject: [PATCH 05/16] added concurrency

---
 .../google/auth/aio/transport/sessions.py     | 17 +++-
 .../tests/transport/aio/test_sessions.py      | 82 +++++++++++++++++++
 2 files changed, 98 insertions(+), 1 deletion(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 872b17751df2..4ef92dda1176 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -125,6 +125,9 @@ def __init__(self, credentials: google.auth.credentials.Credentials):
         super().__init__()
         self._credentials = credentials
         self._sync_request: Optional[google.auth.transport.Request] = None
+        # Synchronous credentials are not safe to refresh concurrently, which
+        # concurrent requests would otherwise do from multiple worker threads.
+        self._refresh_lock = asyncio.Lock()
 
     def _get_sync_request(self) -> google.auth.transport.Request:
         """Returns the synchronous transport used to call the wrapped credentials.
@@ -144,9 +147,21 @@ async def apply(self, headers, token=None):
         self._credentials.apply(headers, token=token)
 
     async def refresh(self, request):
-        await asyncio.to_thread(self._credentials.refresh, self._get_sync_request())
+        async with self._refresh_lock:
+            await asyncio.to_thread(self._credentials.refresh, self._get_sync_request())
 
     async def before_request(self, request, method, url, headers):
+        if not self._credentials.valid:
+            # Only one task refreshes the credentials; the others wait for it
+            # and then reuse the refreshed token. Requests with valid
+            # credentials never wait for the lock.
+            async with self._refresh_lock:
+                if not self._credentials.valid:
+                    await self._before_request_in_thread(method, url, headers)
+                    return
+        await self._before_request_in_thread(method, url, headers)
+
+    async def _before_request_in_thread(self, method, url, headers):
         await asyncio.to_thread(
             self._credentials.before_request,
             self._get_sync_request(),
diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index 73c063e02607..28b8dc10bd4c 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -436,6 +436,29 @@ def test_mock_request_clone():
     assert cloned is request
 
 
+class BlockingRefreshCredentials(google.auth.credentials.Credentials):
+    """Synchronous credentials whose refresh blocks until released by the test."""
+
+    def __init__(self):
+        super().__init__()
+        self.refresh_calls = 0
+        self.in_flight_refreshes = 0
+        self.max_in_flight_refreshes = 0
+        self.refresh_started = threading.Event()
+        self.release_refresh = threading.Event()
+
+    def refresh(self, request):
+        self.refresh_calls += 1
+        self.in_flight_refreshes += 1
+        self.max_in_flight_refreshes = max(
+            self.max_in_flight_refreshes, self.in_flight_refreshes
+        )
+        self.refresh_started.set()
+        self.release_refresh.wait(timeout=5)
+        self.in_flight_refreshes -= 1
+        self.token = "token"
+
+
 class TestSyncCredentialsAdapter(object):
     TEST_URL = "http://example.com/"
 
@@ -475,3 +498,62 @@ async def test_blocking_calls_run_off_the_event_loop_thread(self):
 
         assert len(thread_ids) == 2
         assert threading.get_ident() not in thread_ids
+
+    @pytest.mark.asyncio
+    async def test_concurrent_before_request_refreshes_once(self):
+        sync_credentials = BlockingRefreshCredentials()
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+        headers = [{} for _ in range(5)]
+
+        tasks = [
+            asyncio.create_task(adapter.before_request(Mock(), "GET", self.TEST_URL, h))
+            for h in headers
+        ]
+        await asyncio.to_thread(sync_credentials.refresh_started.wait, 5)
+        # Give the remaining tasks the opportunity to start a refresh of their own.
+        await asyncio.sleep(0.1)
+        sync_credentials.release_refresh.set()
+        await asyncio.gather(*tasks)
+
+        assert sync_credentials.refresh_calls == 1
+        assert all(h["authorization"] == "Bearer token" for h in headers)
+
+    @pytest.mark.asyncio
+    async def test_refresh_is_serialized_with_before_request(self):
+        sync_credentials = BlockingRefreshCredentials()
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+        headers = {}
+
+        refresh_task = asyncio.create_task(adapter.refresh(Mock()))
+        await asyncio.to_thread(sync_credentials.refresh_started.wait, 5)
+        before_request_task = asyncio.create_task(
+            adapter.before_request(Mock(), "GET", self.TEST_URL, headers)
+        )
+        await asyncio.sleep(0.1)
+        sync_credentials.release_refresh.set()
+        await asyncio.gather(refresh_task, before_request_task)
+
+        assert sync_credentials.max_in_flight_refreshes == 1
+        assert sync_credentials.refresh_calls == 1
+        assert headers["authorization"] == "Bearer token"
+
+    @pytest.mark.asyncio
+    async def test_before_request_with_valid_credentials_does_not_wait_for_refresh(
+        self,
+    ):
+        sync_credentials = BlockingRefreshCredentials()
+        sync_credentials.token = "token"
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+        headers = {}
+
+        refresh_task = asyncio.create_task(adapter.refresh(Mock()))
+        await asyncio.to_thread(sync_credentials.refresh_started.wait, 5)
+        # Requests that already have a valid token must not wait for the refresh.
+        await asyncio.wait_for(
+            adapter.before_request(Mock(), "GET", self.TEST_URL, headers), timeout=5
+        )
+        assert headers["authorization"] == "Bearer token"
+
+        sync_credentials.release_refresh.set()
+        await refresh_task
+        assert sync_credentials.refresh_calls == 1

From 577eaff97267f404e9dedfbba069bd804ef2b477 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 18:58:09 -0700
Subject: [PATCH 06/16] removed api_core test against auth credentials

---
 .../test_operations_rest_client.py            | 28 +++++--------------
 1 file changed, 7 insertions(+), 21 deletions(-)

diff --git a/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py b/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py
index cdb0722474d6..a60e4a26d0f7 100644
--- a/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py
+++ b/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py
@@ -1235,30 +1235,16 @@ def test_operations_base_transport_with_adc():
         adc.assert_called_once()
 
 
-@pytest.mark.parametrize(
-    "client_class",
-    PYPARAM_CLIENT,
-)
-def test_operations_auth_adc(client_class):
+def test_operations_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
     with mock.patch.object(google.auth, "default", autospec=True) as adc:
         adc.return_value = (ga_credentials.AnonymousCredentials(), None)
-
-        if "async" in str(client_class).lower():
-            # TODO(): Add support for adc to async REST transport.
-            # NOTE: Ideally, the logic for adc shouldn't be called if transport
-            # is set to async REST. If the user does not configure credentials
-            # of type `google.auth.aio.credentials.Credentials`,
-            # we should raise an exception to avoid the adc workflow.
-            with pytest.raises(google.auth.exceptions.InvalidType):
-                client_class()
-        else:
-            client_class()
-            adc.assert_called_once_with(
-                scopes=None,
-                default_scopes=(),
-                quota_project_id=None,
-            )
+        AbstractOperationsClient()
+        adc.assert_called_once_with(
+            scopes=None,
+            default_scopes=(),
+            quota_project_id=None,
+        )
 
 
 # TODO(https://github.com/googleapis/python-api-core/issues/705): Add

From 74b334441f32fc154560040b7ce0522a0f4cd146 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 18:59:35 -0700
Subject: [PATCH 07/16] use shared refresh

---
 .../google/auth/aio/transport/sessions.py     | 31 +++++++-----
 .../tests/transport/aio/test_sessions.py      | 49 ++++++++++++++++++-
 2 files changed, 65 insertions(+), 15 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 4ef92dda1176..aef6b5600fc2 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -127,7 +127,8 @@ def __init__(self, credentials: google.auth.credentials.Credentials):
         self._sync_request: Optional[google.auth.transport.Request] = None
         # Synchronous credentials are not safe to refresh concurrently, which
         # concurrent requests would otherwise do from multiple worker threads.
-        self._refresh_lock = asyncio.Lock()
+        # Instead, at most one refresh is in flight and concurrent callers share it.
+        self._pending_refresh: Optional["asyncio.Future[None]"] = None
 
     def _get_sync_request(self) -> google.auth.transport.Request:
         """Returns the synchronous transport used to call the wrapped credentials.
@@ -143,25 +144,29 @@ def _get_sync_request(self) -> google.auth.transport.Request:
             self._sync_request = sync_requests.Request()
         return self._sync_request
 
+    async def _refresh_shared(self):
+        """Refreshes the wrapped credentials, joining a refresh already in flight.
+
+        The refresh is shielded from cancellation: a caller that is cancelled
+        while waiting (e.g. because of a timeout) stops waiting, but the refresh
+        completes so that the next caller joins it rather than starting a
+        second, concurrent refresh.
+        """
+        if self._pending_refresh is None or self._pending_refresh.done():
+            self._pending_refresh = asyncio.ensure_future(
+                asyncio.to_thread(self._credentials.refresh, self._get_sync_request())
+            )
+        await asyncio.shield(self._pending_refresh)
+
     async def apply(self, headers, token=None):
         self._credentials.apply(headers, token=token)
 
     async def refresh(self, request):
-        async with self._refresh_lock:
-            await asyncio.to_thread(self._credentials.refresh, self._get_sync_request())
+        await self._refresh_shared()
 
     async def before_request(self, request, method, url, headers):
         if not self._credentials.valid:
-            # Only one task refreshes the credentials; the others wait for it
-            # and then reuse the refreshed token. Requests with valid
-            # credentials never wait for the lock.
-            async with self._refresh_lock:
-                if not self._credentials.valid:
-                    await self._before_request_in_thread(method, url, headers)
-                    return
-        await self._before_request_in_thread(method, url, headers)
-
-    async def _before_request_in_thread(self, method, url, headers):
+            await self._refresh_shared()
         await asyncio.to_thread(
             self._credentials.before_request,
             self._get_sync_request(),
diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index 28b8dc10bd4c..37d211cc74f3 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -32,7 +32,12 @@
     Response,
     sessions,
 )
-from google.auth.exceptions import InvalidType, TimeoutError, TransportError
+from google.auth.exceptions import (
+    InvalidType,
+    RefreshError,
+    TimeoutError,
+    TransportError,
+)
 
 
 @pytest.fixture
@@ -519,7 +524,7 @@ async def test_concurrent_before_request_refreshes_once(self):
         assert all(h["authorization"] == "Bearer token" for h in headers)
 
     @pytest.mark.asyncio
-    async def test_refresh_is_serialized_with_before_request(self):
+    async def test_refresh_and_before_request_share_one_refresh(self):
         sync_credentials = BlockingRefreshCredentials()
         adapter = sessions._SyncCredentialsAdapter(sync_credentials)
         headers = {}
@@ -537,6 +542,46 @@ async def test_refresh_is_serialized_with_before_request(self):
         assert sync_credentials.refresh_calls == 1
         assert headers["authorization"] == "Bearer token"
 
+    @pytest.mark.asyncio
+    async def test_cancelled_caller_does_not_abandon_refresh(self):
+        sync_credentials = BlockingRefreshCredentials()
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+
+        cancelled_task = asyncio.create_task(
+            adapter.before_request(Mock(), "GET", self.TEST_URL, {})
+        )
+        await asyncio.to_thread(sync_credentials.refresh_started.wait, 5)
+        cancelled_task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await cancelled_task
+
+        # The refresh started on behalf of the cancelled request is still running:
+        # a new request must wait for it rather than start a second refresh.
+        headers = {}
+        waiting_task = asyncio.create_task(
+            adapter.before_request(Mock(), "GET", self.TEST_URL, headers)
+        )
+        await asyncio.sleep(0.1)
+        assert not waiting_task.done()
+        sync_credentials.release_refresh.set()
+        await waiting_task
+
+        assert sync_credentials.refresh_calls == 1
+        assert sync_credentials.max_in_flight_refreshes == 1
+        assert headers["authorization"] == "Bearer token"
+
+    @pytest.mark.asyncio
+    async def test_failed_refresh_is_not_reused(self):
+        sync_credentials = Mock(spec=google.auth.credentials.Credentials)
+        sync_credentials.refresh.side_effect = [RefreshError("refresh failed"), None]
+        adapter = sessions._SyncCredentialsAdapter(sync_credentials)
+
+        with pytest.raises(RefreshError):
+            await adapter.refresh(Mock())
+        await adapter.refresh(Mock())
+
+        assert sync_credentials.refresh.call_count == 2
+
     @pytest.mark.asyncio
     async def test_before_request_with_valid_credentials_does_not_wait_for_refresh(
         self,

From bdc873272e17eecca07de65c9adee6992fbcbe4b Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 1 Oct 2026 19:31:19 -0700
Subject: [PATCH 08/16] removed lazy sync request

---
 .../google/auth/aio/transport/sessions.py     | 27 +++++++------------
 .../tests/transport/aio/test_sessions.py      |  2 +-
 2 files changed, 10 insertions(+), 19 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index aef6b5600fc2..a1129afd32d0 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -25,7 +25,6 @@
 from typing import TYPE_CHECKING, Mapping, Optional, Union
 
 import google.auth.credentials
-import google.auth.transport
 import google.auth.transport._mtls_helper
 from google.auth import _exponential_backoff, exceptions
 from google.auth.aio import transport
@@ -123,27 +122,19 @@ class _SyncCredentialsAdapter(Credentials):
 
     def __init__(self, credentials: google.auth.credentials.Credentials):
         super().__init__()
+        # Imported here because `requests` is an optional dependency of
+        # google-auth. It is installed alongside `aiohttp` by the `aiohttp` extra.
+        from google.auth.transport import requests as sync_requests
+
         self._credentials = credentials
-        self._sync_request: Optional[google.auth.transport.Request] = None
+        # Synchronous credentials cannot use the asynchronous transport of the
+        # session, so they are called with a synchronous transport instead.
+        self._sync_request = sync_requests.Request()
         # Synchronous credentials are not safe to refresh concurrently, which
         # concurrent requests would otherwise do from multiple worker threads.
         # Instead, at most one refresh is in flight and concurrent callers share it.
         self._pending_refresh: Optional["asyncio.Future[None]"] = None
 
-    def _get_sync_request(self) -> google.auth.transport.Request:
-        """Returns the synchronous transport used to call the wrapped credentials.
-
-        Synchronous credentials cannot use the asynchronous transport of the
-        session, so a synchronous transport is created on first use and reused.
-        """
-        if self._sync_request is None:
-            # Imported lazily because `requests` is an optional dependency of
-            # google-auth. It is installed alongside `aiohttp` by the `aiohttp` extra.
-            from google.auth.transport import requests as sync_requests
-
-            self._sync_request = sync_requests.Request()
-        return self._sync_request
-
     async def _refresh_shared(self):
         """Refreshes the wrapped credentials, joining a refresh already in flight.
 
@@ -154,7 +145,7 @@ async def _refresh_shared(self):
         """
         if self._pending_refresh is None or self._pending_refresh.done():
             self._pending_refresh = asyncio.ensure_future(
-                asyncio.to_thread(self._credentials.refresh, self._get_sync_request())
+                asyncio.to_thread(self._credentials.refresh, self._sync_request)
             )
         await asyncio.shield(self._pending_refresh)
 
@@ -169,7 +160,7 @@ async def before_request(self, request, method, url, headers):
             await self._refresh_shared()
         await asyncio.to_thread(
             self._credentials.before_request,
-            self._get_sync_request(),
+            self._sync_request,
             method,
             url,
             headers,
diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index 37d211cc74f3..0a3369806cbe 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -477,7 +477,7 @@ async def test_delegates_to_sync_credentials(self):
         await adapter.refresh(Mock())
         await adapter.apply(headers, token="token")
 
-        # A synchronous transport is created lazily and reused for every call.
+        # The same synchronous transport is used for every call.
         sync_request = sync_credentials.before_request.call_args.args[0]
         assert isinstance(sync_request, google.auth.transport.requests.Request)
         sync_credentials.before_request.assert_called_once_with(

From 8589cdcd0657849be56d46dd6c674debc0355b44 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Fri, 2 Oct 2026 09:26:20 -0700
Subject: [PATCH 09/16] use task instead of future

---
 packages/google-auth/google/auth/aio/transport/sessions.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index a1129afd32d0..344f7ffa7ebd 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -133,7 +133,7 @@ def __init__(self, credentials: google.auth.credentials.Credentials):
         # Synchronous credentials are not safe to refresh concurrently, which
         # concurrent requests would otherwise do from multiple worker threads.
         # Instead, at most one refresh is in flight and concurrent callers share it.
-        self._pending_refresh: Optional["asyncio.Future[None]"] = None
+        self._pending_refresh: Optional["asyncio.Task[None]"] = None
 
     async def _refresh_shared(self):
         """Refreshes the wrapped credentials, joining a refresh already in flight.
@@ -144,7 +144,7 @@ async def _refresh_shared(self):
         second, concurrent refresh.
         """
         if self._pending_refresh is None or self._pending_refresh.done():
-            self._pending_refresh = asyncio.ensure_future(
+            self._pending_refresh = asyncio.create_task(
                 asyncio.to_thread(self._credentials.refresh, self._sync_request)
             )
         await asyncio.shield(self._pending_refresh)

From 990f47cc060f78816f9ddd9693dcf0684e7bea37 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <sanche@google.com>
Date: Fri, 2 Oct 2026 10:30:49 -0700
Subject: [PATCH 10/16] extended test based on feedback

Co-authored-by: Anthonios Partheniou <partheniou@google.com>
---
 .../google-auth/tests/transport/aio/test_sessions.py     | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index 0a3369806cbe..aea934ec1261 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -470,6 +470,10 @@ class TestSyncCredentialsAdapter(object):
     @pytest.mark.asyncio
     async def test_delegates_to_sync_credentials(self):
         sync_credentials = Mock(spec=google.auth.credentials.Credentials)
+        sync_credentials.token = "sync-token"
+        sync_credentials.expiry = Mock()
+        sync_credentials.valid = True
+        sync_credentials.expired = False
         adapter = sessions._SyncCredentialsAdapter(sync_credentials)
         headers = {}
 
@@ -477,6 +481,11 @@ async def test_delegates_to_sync_credentials(self):
         await adapter.refresh(Mock())
         await adapter.apply(headers, token="token")
 
+        assert adapter.token == "sync-token"
+        assert adapter.expiry is sync_credentials.expiry
+        assert adapter.valid is True
+        assert adapter.expired is False
+
         # The same synchronous transport is used for every call.
         sync_request = sync_credentials.before_request.call_args.args[0]
         assert isinstance(sync_request, google.auth.transport.requests.Request)

From 768328925adeda43e2cf63be3440fbc4d3574917 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <sanche@google.com>
Date: Fri, 2 Oct 2026 10:31:20 -0700
Subject: [PATCH 11/16] extended test based on feedback

Co-authored-by: Anthonios Partheniou <partheniou@google.com>
---
 packages/google-auth/tests/transport/aio/test_sessions.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index aea934ec1261..19e251aab3e8 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -494,6 +494,10 @@ async def test_delegates_to_sync_credentials(self):
         )
         sync_credentials.refresh.assert_called_once_with(sync_request)
         sync_credentials.apply.assert_called_once_with(headers, token="token")
+        with patch.object(sync_request.session, "close") as mock_close:
+            adapter.close()
+            mock_close.assert_called_once()
+        
 
     @pytest.mark.asyncio
     async def test_blocking_calls_run_off_the_event_loop_thread(self):

From 13af2515a3ff9cbd952f1ea4f7d60e479faa66a4 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Fri, 2 Oct 2026 09:43:17 -0700
Subject: [PATCH 12/16] added back parametrization

---
 .../operations_v1/test_operations_rest_client.py   | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git a/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py b/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py
index a60e4a26d0f7..17d9150e5f2b 100644
--- a/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py
+++ b/packages/google-api-core/tests/unit/operations_v1/test_operations_rest_client.py
@@ -1235,11 +1235,21 @@ def test_operations_base_transport_with_adc():
         adc.assert_called_once()
 
 
-def test_operations_auth_adc():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_operations_auth_adc(client_class):
     # If no credentials are provided, we should use ADC credentials.
+    is_async = "async" in str(client_class).lower()
+    if is_async and parse_version_to_tuple(auth_version) < (2, 60, 0):
+        # Older versions of google-auth do not accept the synchronous credentials
+        # returned by ADC in `AsyncAuthorizedSession`.
+        pytest.skip("ADC with the async REST transport requires google-auth >= 2.60.0")
+
     with mock.patch.object(google.auth, "default", autospec=True) as adc:
         adc.return_value = (ga_credentials.AnonymousCredentials(), None)
-        AbstractOperationsClient()
+        client_class()
         adc.assert_called_once_with(
             scopes=None,
             default_scopes=(),

From fc51e19f2d30e58f536f38874871f0f761a6b53c Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Fri, 2 Oct 2026 10:53:00 -0700
Subject: [PATCH 13/16] extended delegated interface

---
 .../google/auth/aio/transport/sessions.py     | 36 +++++++++++++++++--
 1 file changed, 34 insertions(+), 2 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 344f7ffa7ebd..c84a0a9c397f 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -26,7 +26,7 @@
 
 import google.auth.credentials
 import google.auth.transport._mtls_helper
-from google.auth import _exponential_backoff, exceptions
+from google.auth import _exponential_backoff, _helpers, exceptions
 from google.auth.aio import transport
 from google.auth.aio.credentials import Credentials
 from google.auth.aio.transport import mtls
@@ -125,7 +125,6 @@ def __init__(self, credentials: google.auth.credentials.Credentials):
         # Imported here because `requests` is an optional dependency of
         # google-auth. It is installed alongside `aiohttp` by the `aiohttp` extra.
         from google.auth.transport import requests as sync_requests
-
         self._credentials = credentials
         # Synchronous credentials cannot use the asynchronous transport of the
         # session, so they are called with a synchronous transport instead.
@@ -135,6 +134,37 @@ def __init__(self, credentials: google.auth.credentials.Credentials):
         # Instead, at most one refresh is in flight and concurrent callers share it.
         self._pending_refresh: Optional["asyncio.Task[None]"] = None
 
+
+    def close(self):
+        if (
+            self._sync_request_instance is not None
+            and hasattr(self._sync_request_instance, "session")
+            and self._sync_request_instance.session is not None
+        ):
+            self._sync_request_instance.session.close()
+
+    @property
+    def token(self):
+        """Optional[str]: The bearer token that can be used in HTTP headers to make
+        authenticated requests."""
+        return self._credentials.token
+
+    @property
+    def expiry(self):
+        """Optional[datetime]: When the token expires and is no longer valid.
+        If this is None, the token is assumed to never expire."""
+        return self._credentials.expiry
+
+    @property
+    @_helpers.copy_docstring(google.auth.credentials.Credentials)
+    def valid(self):
+        return self._credentials.valid
+
+    @property
+    @_helpers.copy_docstring(google.auth.credentials.Credentials)
+    def expired(self):
+        return self._credentials.expired
+
     async def _refresh_shared(self):
         """Refreshes the wrapped credentials, joining a refresh already in flight.
 
@@ -915,6 +945,8 @@ async def close(self) -> None:
                 if inspect.isawaitable(res):
                     await res
         finally:
+            if hasattr(self._credentials, "close"):
+                self._credentials.close()
             for old_request in self._old_auth_requests:
                 try:
                     if hasattr(old_request, "close"):

From cebe8ae6b696c356442f64c699c1cb5693b3e33b Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Fri, 2 Oct 2026 10:53:16 -0700
Subject: [PATCH 14/16] lazy sync transport

---
 .../google/auth/aio/transport/sessions.py         | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index c84a0a9c397f..2f8905f7acf9 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -121,19 +121,24 @@ class _SyncCredentialsAdapter(Credentials):
     """
 
     def __init__(self, credentials: google.auth.credentials.Credentials):
-        super().__init__()
-        # Imported here because `requests` is an optional dependency of
-        # google-auth. It is installed alongside `aiohttp` by the `aiohttp` extra.
-        from google.auth.transport import requests as sync_requests
         self._credentials = credentials
         # Synchronous credentials cannot use the asynchronous transport of the
         # session, so they are called with a synchronous transport instead.
-        self._sync_request = sync_requests.Request()
+        self._sync_request_instance = None
         # Synchronous credentials are not safe to refresh concurrently, which
         # concurrent requests would otherwise do from multiple worker threads.
         # Instead, at most one refresh is in flight and concurrent callers share it.
         self._pending_refresh: Optional["asyncio.Task[None]"] = None
 
+    @property
+    def _sync_request(self):
+        if self._sync_request_instance is None:
+            # Imported here because `requests` is an optional dependency of
+            # google-auth. It is installed alongside `aiohttp` by the `aiohttp` extra.
+            from google.auth.transport import requests as sync_requests
+
+            self._sync_request_instance = sync_requests.Request()
+        return self._sync_request_instance
 
     def close(self):
         if (

From 58b8bff80f9aecac482a9cc04811feb6c32014e5 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Fri, 2 Oct 2026 10:53:32 -0700
Subject: [PATCH 15/16] copy docstrings

---
 packages/google-auth/google/auth/aio/transport/sessions.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 2f8905f7acf9..14e950427b5f 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -184,12 +184,15 @@ async def _refresh_shared(self):
             )
         await asyncio.shield(self._pending_refresh)
 
+    @_helpers.copy_docstring(Credentials)
     async def apply(self, headers, token=None):
         self._credentials.apply(headers, token=token)
 
+    @_helpers.copy_docstring(Credentials)
     async def refresh(self, request):
         await self._refresh_shared()
 
+    @_helpers.copy_docstring(Credentials)
     async def before_request(self, request, method, url, headers):
         if not self._credentials.valid:
             await self._refresh_shared()

From e4464a903f53dc81eb2dbc69bf426be0b0f460d3 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Fri, 2 Oct 2026 10:53:45 -0700
Subject: [PATCH 16/16] updated tests

---
 .../tests/transport/aio/test_sessions.py       | 18 ++++++++++++------
 1 file changed, 12 insertions(+), 6 deletions(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index 19e251aab3e8..0b4ce134fa9c 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -226,7 +226,13 @@ async def test_constructor_with_sync_credentials(self):
         assert isinstance(authed_session._credentials, Credentials)
         assert isinstance(authed_session._credentials, sessions._SyncCredentialsAdapter)
         assert authed_session._credentials._credentials is sync_credentials
-        await authed_session.close()
+        with patch.object(
+            authed_session._credentials,
+            "close",
+            wraps=authed_session._credentials.close,
+        ) as mock_close:
+            await authed_session.close()
+            mock_close.assert_called_once()
 
     @pytest.mark.asyncio
     async def test_request_with_sync_credentials_success(self, mocked_content):
@@ -497,7 +503,6 @@ async def test_delegates_to_sync_credentials(self):
         with patch.object(sync_request.session, "close") as mock_close:
             adapter.close()
             mock_close.assert_called_once()
-        
 
     @pytest.mark.asyncio
     async def test_blocking_calls_run_off_the_event_loop_thread(self):
@@ -528,8 +533,8 @@ async def test_concurrent_before_request_refreshes_once(self):
             for h in headers
         ]
         await asyncio.to_thread(sync_credentials.refresh_started.wait, 5)
-        # Give the remaining tasks the opportunity to start a refresh of their own.
-        await asyncio.sleep(0.1)
+        # Yield to the event loop so all tasks run until they await the refresh.
+        await asyncio.sleep(0)
         sync_credentials.release_refresh.set()
         await asyncio.gather(*tasks)
 
@@ -547,7 +552,8 @@ async def test_refresh_and_before_request_share_one_refresh(self):
         before_request_task = asyncio.create_task(
             adapter.before_request(Mock(), "GET", self.TEST_URL, headers)
         )
-        await asyncio.sleep(0.1)
+        # Yield to the event loop so before_request_task reaches the refresh.
+        await asyncio.sleep(0)
         sync_credentials.release_refresh.set()
         await asyncio.gather(refresh_task, before_request_task)
 
@@ -574,7 +580,7 @@ async def test_cancelled_caller_does_not_abandon_refresh(self):
         waiting_task = asyncio.create_task(
             adapter.before_request(Mock(), "GET", self.TEST_URL, headers)
         )
-        await asyncio.sleep(0.1)
+        await asyncio.sleep(0)
         assert not waiting_task.done()
         sync_credentials.release_refresh.set()
         await waiting_task