diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index f3fb40a85..c8ea80699 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -111,28 +111,28 @@ repos:
     name: mypy-bigquery
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.2
+    - gcloud-aio-auth==5.4.4
     - types-requests==2.32.4.20260107
     files: bigquery/
   - <<: *mypy
     name: mypy-datastore
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.2
+    - gcloud-aio-auth==5.4.4
     - types-requests==2.32.4.20260107
     files: datastore/
   - <<: *mypy
     name: mypy-kms
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.2
+    - gcloud-aio-auth==5.4.4
     - types-requests==2.32.4.20260107
     files: kms/
   - <<: *mypy
     name: mypy-pubsub
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.2
+    - gcloud-aio-auth==5.4.4
     - prometheus-client==0.24.1
     - types-requests==2.32.4.20260107
     files: pubsub/
@@ -140,7 +140,7 @@ repos:
     name: mypy-storage
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.2
+    - gcloud-aio-auth==5.4.4
     - rsa==4.9.1
     - types-aiofiles==25.1.0.20251011
     - types-requests==2.32.4.20260107
@@ -149,7 +149,7 @@ repos:
     name: mypy-taskqueue
     additional_dependencies:
     - aiohttp==3.13.3
-    - gcloud-aio-auth==5.4.2
+    - gcloud-aio-auth==5.4.4
     - types-requests==2.32.4.20260107
     files: taskqueue/
 - repo: https://github.com/asottile/yesqa
diff --git a/auth/gcloud/aio/auth/session.py b/auth/gcloud/aio/auth/session.py
index c791b9ed4..1ea608b07 100644
--- a/auth/gcloud/aio/auth/session.py
+++ b/auth/gcloud/aio/auth/session.py
@@ -52,7 +52,7 @@ async def get(
         self, url: str, headers: Mapping[str, str] | None,
         timeout: float, params: Mapping[str, int | str] | None,
         stream: bool,
-        auto_decompress: bool,
+        auto_decompress: bool | None,
     ) -> Response:
         pass
 
@@ -187,7 +187,7 @@ async def get(  # type: ignore[override]
             timeout: Timeout = 10,
             params: Mapping[str, int | str] | None = None,
             stream: bool | None = None,
-            auto_decompress: bool = True,
+            auto_decompress: bool | None = True,
         ) -> aiohttp.ClientResponse:
             if not isinstance(timeout, aiohttp.ClientTimeout):
                 timeout = aiohttp.ClientTimeout(total=timeout)
@@ -326,9 +326,9 @@ async def get(
             timeout: float = 10,
             params: Mapping[str, int | str] | None = None,
             stream: bool = False,
-            auto_decompress: bool = True,
+            auto_decompress: bool | None = True,
         ) -> Response:
-            if not auto_decompress and not stream:
+            if auto_decompress is False and not stream:
                 warnings.warn(
                     'the requests library always decompresses responses when '
                     'outside of streaming mode; when auto_decompress is '
diff --git a/auth/pyproject.rest.toml b/auth/pyproject.rest.toml
index 653e6eaf7..1a90ecd48 100644
--- a/auth/pyproject.rest.toml
+++ b/auth/pyproject.rest.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gcloud-rest-auth"
-version = "5.4.3"
+version = "5.4.4"
 description = "Python Client for Google Cloud Auth"
 readme = "README.rst"
 
diff --git a/auth/pyproject.toml b/auth/pyproject.toml
index f52e0a891..0b9bd9caf 100644
--- a/auth/pyproject.toml
+++ b/auth/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gcloud-aio-auth"
-version = "5.4.3"
+version = "5.4.4"
 description = "Python Client for Google Cloud Auth"
 readme = "README.rst"
 
diff --git a/pubsub/tests/unit/subscriber_test.py b/pubsub/tests/unit/subscriber_test.py
index 86694ceba..cd58cef27 100644
--- a/pubsub/tests/unit/subscriber_test.py
+++ b/pubsub/tests/unit/subscriber_test.py
@@ -1,3 +1,4 @@
+# pylint: disable=no-name-in-module
 # pylint: disable=redefined-outer-name
 # pylint: disable=too-many-lines
 from gcloud.aio.auth import BUILD_GCLOUD_REST
@@ -10,6 +11,7 @@
     import asyncio
     import time
     import logging
+    from unittest.mock import AsyncMock
     from unittest.mock import call
     from unittest.mock import MagicMock
     from unittest.mock import patch
@@ -37,39 +39,24 @@ def message():
     def subscriber_client(message):
         mock = MagicMock()
 
-        f = asyncio.Future()
-        f.set_result({'ackDeadlineSeconds': 42})
-        mock.get_subscription = MagicMock(return_value=f)
-
-        async def g(*_args, **_kwargs):
-            return [message]
-        mock.pull = g
-
-        f = asyncio.Future()
-        f.set_result(None)
-        mock.acknowledge = MagicMock(return_value=f)
-
-        f = asyncio.Future()
-        f.set_result(None)
-        mock.modify_ack_deadline = MagicMock(return_value=f)
+        mock.get_subscription = AsyncMock(
+            return_value={'ackDeadlineSeconds': 42},
+        )
+        mock.pull = AsyncMock(return_value=[message])
+        mock.acknowledge = AsyncMock(return_value=None)
+        mock.modify_ack_deadline = AsyncMock(return_value=None)
 
         return mock
 
     @pytest.fixture(scope='function')
     def ack_deadline_cache():
-        f = asyncio.Future()
-        f.set_result(float('inf'))
-
         mock = MagicMock()
-        mock.get = MagicMock(return_value=f)
+        mock.get = AsyncMock(return_value=float('inf'))
         return mock
 
     @pytest.fixture(scope='function')
     def application_callback():
-        f = asyncio.Future()
-        f.set_result(None)
-
-        return MagicMock(return_value=f)
+        return AsyncMock(return_value=None)
 
     # ================
     # AckDeadlineCache
@@ -126,10 +113,8 @@ async def test_ack_deadline_cache_refresh_updates_value_and_last_refresh(
     async def test_ack_deadline_cache_refresh_is_cool_about_failures(
         subscriber_client,
     ):
-        f = asyncio.Future()
-        f.set_exception(RuntimeError)
-        subscriber_client.get_subscription = MagicMock(
-            return_value=f,
+        subscriber_client.get_subscription = AsyncMock(
+            side_effect=RuntimeError,
         )
         cache = AckDeadlineCache(
             subscriber_client, 'fake_subscription', 1,
@@ -196,10 +181,8 @@ async def test_ack_deadline_cache_get_refreshes_if_outdated(
 
     @pytest.mark.asyncio
     async def test_ack_deadline_cache_first_get_failed(subscriber_client):
-        f = asyncio.Future()
-        f.set_exception(RuntimeError)
-        subscriber_client.get_subscription = MagicMock(
-            return_value=f,
+        subscriber_client.get_subscription = AsyncMock(
+            side_effect=RuntimeError,
         )
         cache = AckDeadlineCache(
             subscriber_client, 'fake_subscription', 10,
@@ -444,9 +427,7 @@ async def test_consumer_drops_expired_messages(
         message,
         application_callback,
     ):
-        f = asyncio.Future()
-        f.set_result(0.0)
-        ack_deadline_cache.get = MagicMock(return_value=f)
+        ack_deadline_cache.get = AsyncMock(return_value=0.0)
 
         queue = asyncio.Queue()
         ack_queue = asyncio.Queue()
@@ -946,7 +927,7 @@ def exception_handler(_loop, context) -> None:
         # Ensure the test fails on unhandled exceptions
         asyncio.get_running_loop().set_exception_handler(exception_handler)
 
-        pull_ret = asyncio.Future()
+        pull_ret = asyncio.get_running_loop().create_future()
         pull_called = asyncio.Event()
 
         async def pull(*_args, **_kwargs):
diff --git a/storage/gcloud/aio/storage/storage.py b/storage/gcloud/aio/storage/storage.py
index 0661c1a94..a78722000 100644
--- a/storage/gcloud/aio/storage/storage.py
+++ b/storage/gcloud/aio/storage/storage.py
@@ -622,14 +622,18 @@ async def _download(
         headers.update(await self._headers())
 
         # aiohttp and requests automatically decompress the body if this
-        # argument is not passed. We assume that if the Accept-Encoding header
-        # is present, then the client will handle the decompression
-        auto_decompress = 'accept-encoding' not in {k.lower() for k in headers}
+        # argument is not passed, unless a user has explicitly disabled that
+        # option at the session level in the case of aiohttp. We follow the
+        # user setting by default (by passing None) and only explicitly disable
+        # it when a caller explicitly requests a compressed payload.
+        auto_decompress = None
+        if 'accept-encoding' in {k.lower() for k in headers}:
+            auto_decompress = False
 
         s = AioSession(session) if session else self.session
 
         data: bytes
-        if not auto_decompress and BUILD_GCLOUD_REST:
+        if auto_decompress is False and BUILD_GCLOUD_REST:
             # Requests lib has a different way of reading compressed data. We
             # must pass the stream=True argument and read the response using
             # the 'raw' property.
@@ -666,6 +670,15 @@ async def _download_stream(
         headers = headers or {}
         headers.update(await self._headers())
 
+        # aiohttp and requests automatically decompress the body if this
+        # argument is not passed, unless a user has explicitly disabled that
+        # option at the session level in the case of aiohttp. We follow the
+        # user setting by default (by passing None) and only explicitly disable
+        # it when a caller explicitly requests a compressed payload.
+        auto_decompress = None
+        if 'accept-encoding' in {k.lower() for k in headers}:
+            auto_decompress = False
+
         s = AioSession(session) if session else self.session
 
         if BUILD_GCLOUD_REST:
@@ -680,7 +693,7 @@ async def _download_stream(
         return StreamResponse(
             await s.get(
                 url, headers=headers, params=params or {},
-                timeout=timeout,
+                timeout=timeout, auto_decompress=auto_decompress,
             ),
         )
 
diff --git a/storage/poetry.lock b/storage/poetry.lock
index ccf86e6de..a6cd7aaa2 100644
--- a/storage/poetry.lock
+++ b/storage/poetry.lock
@@ -590,7 +590,7 @@ files = [
 
 [[package]]
 name = "gcloud-aio-auth"
-version = "5.4.3"
+version = "5.4.4"
 description = "Python Client for Google Cloud Auth"
 optional = false
 python-versions = ">= 3.10, < 4.0"
@@ -1325,4 +1325,4 @@ propcache = ">=0.2.1"
 [metadata]
 lock-version = "2.1"
 python-versions = ">= 3.10, < 4.0"
-content-hash = "5b6421333a9cf444483a94b602774b656c5d9397e0606cd43675fcd557aa9980"
+content-hash = "720a18daa379404b8bc918471bef5b6f1a908fd8642d299e2a075be174de176d"
diff --git a/storage/poetry.rest.lock b/storage/poetry.rest.lock
index af8667566..614c462dc 100644
--- a/storage/poetry.rest.lock
+++ b/storage/poetry.rest.lock
@@ -364,7 +364,7 @@ test = ["pytest (>=6)"]
 
 [[package]]
 name = "gcloud-rest-auth"
-version = "5.4.3"
+version = "5.4.4"
 description = "Python Client for Google Cloud Auth"
 optional = false
 python-versions = ">= 3.10, < 4.0"
@@ -682,4 +682,4 @@ zstd = ["backports-zstd (>=1.0.0) ; python_version < \"3.14\""]
 [metadata]
 lock-version = "2.1"
 python-versions = ">= 3.10, < 4.0"
-content-hash = "d7b69bb7dc20f1e0046a5ec92b531ba1ee19bfe6f0df054e1601aa194c0e2b58"
+content-hash = "bb040f27baa3c12a381ddfe8a43d67ef7c93fd774a8338b4ae6d55df4b8167f3"
diff --git a/storage/pyproject.rest.toml b/storage/pyproject.rest.toml
index d74c9e6b2..260dcfc91 100644
--- a/storage/pyproject.rest.toml
+++ b/storage/pyproject.rest.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gcloud-rest-storage"
-version = "9.6.3"
+version = "9.6.4"
 description = "Python Client for Google Cloud Storage"
 readme = "README.rst"
 
@@ -22,7 +22,7 @@ classifiers = [
 [tool.poetry.dependencies]
 python = ">= 3.10, < 4.0"
 # aiofiles = ">=0.6.0, <26.0.0"
-gcloud-rest-auth = ">= 5.3.0, < 6.0.0"
+gcloud-rest-auth = ">= 5.4.4, < 6.0.0"
 pyasn1-modules = ">=0.2.1, <0.5.0"
 rsa = ">= 3.1.4, < 5.0.0"
 
diff --git a/storage/pyproject.toml b/storage/pyproject.toml
index f1f8cb12a..6b627c3e8 100644
--- a/storage/pyproject.toml
+++ b/storage/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gcloud-aio-storage"
-version = "9.6.3"
+version = "9.6.4"
 description = "Python Client for Google Cloud Storage"
 readme = "README.rst"
 
@@ -22,7 +22,7 @@ classifiers = [
 [tool.poetry.dependencies]
 python = ">= 3.10, < 4.0"
 aiofiles = ">=0.6.0, <26.0.0"
-gcloud-aio-auth = ">= 5.3.0, < 6.0.0"
+gcloud-aio-auth = ">= 5.4.4, < 6.0.0"
 pyasn1-modules = ">=0.2.1, <0.5.0"
 rsa = ">= 3.1.4, < 5.0.0"