diff --git a/packages/google-cloud-storage/google/cloud/storage/asyncio/async_read_object_stream.py b/packages/google-cloud-storage/google/cloud/storage/asyncio/async_read_object_stream.py index cd7ae067c631..8fd98d623571 100644 --- a/packages/google-cloud-storage/google/cloud/storage/asyncio/async_read_object_stream.py +++ b/packages/google-cloud-storage/google/cloud/storage/asyncio/async_read_object_stream.py @@ -79,6 +79,9 @@ def __init__( self.socket_like_rpc: Optional[AsyncBidiRpc] = None self._is_stream_open: bool = False self.persisted_size: Optional[int] = None + self.is_finalized: bool = False + self.full_obj_server_crc32c: Optional[int] = None + self.object_metadata: Optional[_storage_v2.Object] = None async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None: """Opens the bidi-gRPC connection to read from the object. @@ -132,6 +135,18 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None: self.generation_number = response.metadata.generation # update persisted size self.persisted_size = response.metadata.size + self.object_metadata = response.metadata + if ( + hasattr(response.metadata, "finalize_time") + and response.metadata.finalize_time + and response.metadata.finalize_time.second > 0 + ): + self.is_finalized = True + if ( + hasattr(response.metadata, "checksums") + and response.metadata.checksums + ): + self.full_obj_server_crc32c = response.metadata.checksums.crc32c if response and response.read_handle: self.read_handle = response.read_handle diff --git a/packages/google-cloud-storage/tests/unit/asyncio/test_async_read_object_stream.py b/packages/google-cloud-storage/tests/unit/asyncio/test_async_read_object_stream.py index f5783be6bf94..a8f64422765e 100644 --- a/packages/google-cloud-storage/tests/unit/asyncio/test_async_read_object_stream.py +++ b/packages/google-cloud-storage/tests/unit/asyncio/test_async_read_object_stream.py @@ -38,9 +38,11 @@ async def instantiate_read_obj_stream(mock_client, mock_cls_async_bidi_rpc, open socket_like_rpc.open = AsyncMock() recv_response = mock.MagicMock(spec=_storage_v2.BidiReadObjectResponse) - recv_response.metadata = mock.MagicMock(spec=_storage_v2.Object) + recv_response.metadata = mock.MagicMock() recv_response.metadata.generation = _TEST_GENERATION_NUMBER recv_response.metadata.size = _TEST_OBJECT_SIZE + recv_response.metadata.finalize_time.second = 30 + recv_response.metadata.checksums.crc32c = 98765 recv_response.read_handle = _TEST_READ_HANDLE socket_like_rpc.recv = AsyncMock(return_value=recv_response) @@ -130,6 +132,8 @@ async def test_open(mock_client, mock_cls_async_bidi_rpc): assert read_obj_stream.generation_number == _TEST_GENERATION_NUMBER assert read_obj_stream.read_handle == _TEST_READ_HANDLE assert read_obj_stream.persisted_size == _TEST_OBJECT_SIZE + assert read_obj_stream.is_finalized is True + assert read_obj_stream.full_obj_server_crc32c == 98765 assert read_obj_stream.is_stream_open @@ -381,3 +385,36 @@ async def test_recv_updates_read_handle_on_refresh( await stream.recv() assert stream.read_handle == refreshed_handle + + +@mock.patch("google.cloud.storage.asyncio.async_read_object_stream.AsyncBidiRpc") +@mock.patch( + "google.cloud.storage.asyncio.async_grpc_client.AsyncGrpcClient.grpc_client" +) +@pytest.mark.asyncio +async def test_open_unfinalized_object_skips_checksum( + mock_client, mock_cls_async_bidi_rpc +): + socket_like_rpc = AsyncMock() + mock_cls_async_bidi_rpc.return_value = socket_like_rpc + socket_like_rpc.open = AsyncMock() + + recv_response = mock.MagicMock(spec=_storage_v2.BidiReadObjectResponse) + recv_response.metadata = mock.MagicMock() + recv_response.metadata.generation = _TEST_GENERATION_NUMBER + recv_response.metadata.size = _TEST_OBJECT_SIZE + recv_response.metadata.finalize_time.second = 0 # NOT finalized! + recv_response.metadata.checksums.crc32c = 98765 + recv_response.read_handle = _TEST_READ_HANDLE + socket_like_rpc.recv = AsyncMock(return_value=recv_response) + + read_obj_stream = _AsyncReadObjectStream( + client=mock_client, + bucket_name=_TEST_BUCKET_NAME, + object_name=_TEST_OBJECT_NAME, + ) + + await read_obj_stream.open() + + assert read_obj_stream.is_finalized is False + assert read_obj_stream.full_obj_server_crc32c is None diff --git a/packages/google-cloud-storage/tests/unit/conftest.py b/packages/google-cloud-storage/tests/unit/conftest.py new file mode 100644 index 000000000000..2eeabdc990e6 --- /dev/null +++ b/packages/google-cloud-storage/tests/unit/conftest.py @@ -0,0 +1,32 @@ +# -*- coding: utf-8 -*- +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio +import pytest + + +@pytest.fixture(autouse=True) +def set_event_loop(): + try: + asyncio.get_running_loop() + yield + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + yield + finally: + loop.close() + asyncio.set_event_loop(None)