fix: log warning for unregistered read_id in stream multiplexer

Added a warning log when the stream multiplexer receives data for an unregistered read_id, and updated the corresponding test to verify the logging behavior. Also applied linter formatting to the modified files.

Author: zhixiangli

packages/google-cloud-storage/google/cloud/storage/asyncio/_stream_multiplexer.py

@@ -15,10 +15,11 @@
 from __future__ import annotations
 
 import asyncio
-import grpc
 import logging
 from typing import Awaitable, Callable, Dict, Optional, Set
 
+import grpc
+
 from google.cloud import _storage_v2
 from google.cloud.storage.asyncio.async_read_object_stream import (
     _AsyncReadObjectStream,
@@ -139,6 +140,10 @@ async def _recv_loop(self) -> None:
                         queue = self._queues.get(read_id)
                         if queue:
                             queues_to_notify.add(queue)
+                        else:
+                            logger.warning(
+                                f"Received data for unregistered read_id: {read_id}"
+                            )
                     await asyncio.gather(
                         *(
                             self._put_with_timeout(queue, response)

packages/google-cloud-storage/tests/unit/asyncio/test_stream_multiplexer.py

@@ -13,9 +13,9 @@
 # limitations under the License.
 
 import asyncio
-import grpc
 from unittest.mock import AsyncMock
 
+import grpc
 import pytest
 
 from google.cloud import _storage_v2
@@ -325,7 +325,7 @@ async def test_error_uses_put_nowait(self):
         assert err.exception is exc
 
     @pytest.mark.asyncio
-    async def test_unknown_read_id_is_dropped(self):
+    async def test_unknown_read_id_is_dropped(self, caplog):
         # Given a multiplexer and a response with an unknown read ID
         mock_stream = AsyncMock()
         resp = _make_response(read_id=999)
@@ -341,6 +341,9 @@ async def test_unknown_read_id_is_dropped(self):
         end = await asyncio.wait_for(queue.get(), timeout=1)
         assert isinstance(end, _StreamEnd)
 
+        # And a warning is logged
+        assert "Received data for unregistered read_id: 999" in caplog.text
+
 
 class TestSend:
     @pytest.mark.asyncio