From 174e26fea74134e25d832b03c93cdbdaaec509a7 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Fri, 21 Aug 2026 08:25:44 +0000 Subject: [PATCH 1/4] test(storage): add system tests for Bidi Read/Write (gRPC) Implement parameterized integration tests for Bidirectional Read and Write features, porting test specifications from the Java reference PR. Tests cover standard, RCU, and zonal buckets, and include checksum validation and stream closure scenarios. [Generated-by: AI] --- .../tests/system/test_bidi.py | 488 ++++++++++++++++++ 1 file changed, 488 insertions(+) create mode 100644 packages/google-cloud-storage/tests/system/test_bidi.py diff --git a/packages/google-cloud-storage/tests/system/test_bidi.py b/packages/google-cloud-storage/tests/system/test_bidi.py new file mode 100644 index 000000000000..5937eab2bc16 --- /dev/null +++ b/packages/google-cloud-storage/tests/system/test_bidi.py @@ -0,0 +1,488 @@ +# 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 +# +# https://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 +from io import BytesIO +import os +import uuid +import pytest +import google_crc32c + +from google.api_core import exceptions +from google.api_core.client_options import ClientOptions +from google.api_core.exceptions import NotFound, OutOfRange, InvalidArgument +from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient +from google.cloud.storage.asyncio.async_appendable_object_writer import AsyncAppendableObjectWriter +from google.cloud.storage.asyncio.async_multi_range_downloader import AsyncMultiRangeDownloader + +REGIONAL_RAPID_BUCKET = "java-storage-reg-rapid-preprod-3fe2bb58" +PREPROD_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" + +# We parameterized the tests to run against REGIONAL_RAPID for now. +# We can expand to others if needed. +@pytest.fixture(scope="module") +def bidi_location_type(): + return "REGIONAL_RAPID" + +@pytest.fixture(scope="module") +def grpc_client(bidi_location_type): + if bidi_location_type == "REGIONAL_RAPID": + # Point to preprod endpoint + options = ClientOptions(api_endpoint=PREPROD_ENDPOINT) + return AsyncGrpcClient(client_options=options) + else: + return AsyncGrpcClient() + +@pytest.fixture(scope="module") +def bidi_bucket(bidi_location_type): + if bidi_location_type == "REGIONAL_RAPID": + # Shared pre-created bucket + return REGIONAL_RAPID_BUCKET + elif bidi_location_type == "ZONAL_RAPID": + zonal_bucket = os.getenv("ZONAL_BUCKET") + if not zonal_bucket: + pytest.skip("ZONAL_BUCKET env var not set") + return zonal_bucket + else: + pytest.fail(f"Unsupported location type: {bidi_location_type}") + +# Helper to create objects using sync client if needed, +# but using AsyncAppendableObjectWriter is also fine. +# We will use sync client from conftest if available, but it points to Prod. +# For preprod, we need to initialize a sync client pointing to preprod if we want to use it. +# Actually, we can just use AsyncAppendableObjectWriter to create objects for read tests. +async def create_object(grpc_client, bucket, object_name, data): + writer = AsyncAppendableObjectWriter(grpc_client, bucket, object_name) + await writer.open() + await writer.append(data) + await writer.close(finalize_on_close=True) + +# ----------------- Write Tests ----------------- + +@pytest.mark.asyncio +async def test_appendable_upload_empty_object(grpc_client, bidi_bucket): + object_name = f"test_empty_{uuid.uuid4()}" + + writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer.open() + object_metadata = await writer.close(finalize_on_close=True) + + # Register for deletion (using sync client, it should work if it can access the bucket, + # but sync client might be pointing to Prod. Wait, if sync client is pointing to Prod, + # it won't be able to delete from pre-prod bucket unless we configure it or if the credential + # has access. The bucket is in 'gcs-hyd-connector-benchmarks' project. + # If standard client has access to it, it should work. + # Wait, the sync client in conftest uses default project. + # If we need to delete it from preprod, maybe we should delete it using grpc_client? + # AsyncGrpcClient has delete_object! + # Let's check: + # await grpc_client.delete_object(bidi_bucket, object_name) + # Yes, we can do that. + # So we can manually delete it or use a cleanup helper. + # Let's register it for deletion using a custom cleanup list. + + assert object_metadata.size == 0 + assert int(object_metadata.checksums.crc32c) == int(google_crc32c.value(b"")) + + # Read back and verify empty + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == b"" + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_appendable_upload_bytes(grpc_client, bidi_bucket): + object_name = f"test_bytes_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer.open() + await writer.append(chunk1) + await writer.append(chunk2) + object_metadata = await writer.close(finalize_on_close=True) + + assert object_metadata.size == len(data) + expected_crc = google_crc32c.value(data) + assert int(object_metadata.checksums.crc32c) == expected_crc + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_explicit_flush(grpc_client, bidi_bucket): + object_name = f"test_flush_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer.open() + await writer.append(chunk1) + await writer.flush() # Explicit flush + await writer.append(chunk2) + await writer.close(finalize_on_close=True) + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_appendable_blob_upload_takeover(grpc_client, bidi_bucket): + object_name = f"test_takeover_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + # Writer 1 writes chunk 1 and closes WITHOUT finalizing + writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer1.open() + await writer1.append(chunk1) + persisted_size1 = await writer1.close(finalize_on_close=False) + generation = writer1.generation + assert persisted_size1 == len(chunk1) + assert generation is not None + + # Writer 2 takes over using the generation number + writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) + await writer2.open() + await writer2.append(chunk2) + object_metadata = await writer2.close(finalize_on_close=True) + + assert object_metadata.size == len(data) + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_takeover_just_to_finalize(grpc_client, bidi_bucket): + object_name = f"test_takeover_fin_{uuid.uuid4()}" + data = os.urandom(1000) + + # Writer 1 writes chunk 1 and closes WITHOUT finalizing + writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer1.open() + await writer1.append(data) + persisted_size1 = await writer1.close(finalize_on_close=False) + generation = writer1.generation + assert persisted_size1 == len(data) + + # Writer 2 takes over and finalizes + writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) + await writer2.open() + object_metadata = await writer2.finalize() + + assert object_metadata.size == len(data) + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_explicit_finalize_with_correct_checksum(grpc_client, bidi_bucket): + object_name = f"test_fin_crc_{uuid.uuid4()}" + data = os.urandom(1000) + + writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer.open() + await writer.append(data) + + expected_crc = google_crc32c.value(data) + object_metadata = await writer.finalize(full_object_checksum=expected_crc) + + assert object_metadata.size == len(data) + assert int(object_metadata.checksums.crc32c) == expected_crc + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_explicit_finalize_with_incorrect_checksum_fails(grpc_client, bidi_bucket): + object_name = f"test_fin_bad_crc_{uuid.uuid4()}" + data = os.urandom(1000) + + writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer.open() + await writer.append(data) + + bad_crc = 0 # Incorrect checksum + + with pytest.raises(exceptions.InvalidArgument) as excinfo: + await writer.finalize(full_object_checksum=bad_crc) + + assert "mismatch" in str(excinfo.value).lower() + + # Cleanup (it should not have been finalized, but we might need to delete the unfinalized object if it exists) + # Actually, unfinalized objects might not be deleteable normally? + # Yes they are deleteable. + try: + await grpc_client.delete_object(bidi_bucket, object_name) + except Exception: + pass + + +@pytest.mark.asyncio +async def test_takeover_just_to_finalize_with_incorrect_checksum_fails(grpc_client, bidi_bucket): + object_name = f"test_takeover_bad_crc_{uuid.uuid4()}" + data = os.urandom(1000) + + # Writer 1 writes chunk 1, closes unfinalized + writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer1.open() + await writer1.append(data) + await writer1.close(finalize_on_close=False) + generation = writer1.generation + + # Writer 2 takes over, finalizes with bad checksum + writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) + await writer2.open() + + bad_crc = 0 + with pytest.raises(exceptions.InvalidArgument) as excinfo: + await writer2.finalize(full_object_checksum=bad_crc) + + assert "mismatch" in str(excinfo.value).lower() + + # Cleanup + try: + await grpc_client.delete_object(bidi_bucket, object_name) + except Exception: + pass + + +@pytest.mark.asyncio +async def test_takeover_and_append_with_correct_checksum_works(grpc_client, bidi_bucket): + object_name = f"test_takeover_append_crc_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + # Writer 1 writes chunk 1, closes unfinalized + writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer1.open() + await writer1.append(chunk1) + await writer1.close(finalize_on_close=False) + generation = writer1.generation + + # Writer 2 takes over, appends chunk 2, and finalizes with correct cumulative checksum + writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) + await writer2.open() + await writer2.append(chunk2) + + expected_crc = google_crc32c.value(data) + object_metadata = await writer2.finalize(full_object_checksum=expected_crc) + + assert object_metadata.size == len(data) + assert int(object_metadata.checksums.crc32c) == expected_crc + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_takeover_and_append_with_incorrect_checksum_fails(grpc_client, bidi_bucket): + object_name = f"test_takeover_append_bad_crc_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + # Writer 1 writes chunk 1, closes unfinalized + writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + await writer1.open() + await writer1.append(chunk1) + await writer1.close(finalize_on_close=False) + generation = writer1.generation + + # Writer 2 takes over, appends chunk 2, finalizes with bad checksum + writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) + await writer2.open() + await writer2.append(chunk2) + + bad_crc = 0 + with pytest.raises(exceptions.InvalidArgument) as excinfo: + await writer2.finalize(full_object_checksum=bad_crc) + + assert "mismatch" in str(excinfo.value).lower() + + # Cleanup + try: + await grpc_client.delete_object(bidi_bucket, object_name) + except Exception: + pass + + +# ----------------- Read Tests ----------------- + +@pytest.mark.asyncio +async def test_read_post_stream_close(grpc_client, bidi_bucket): + object_name = f"test_read_close_{uuid.uuid4()}" + data = os.urandom(5 * 1024 * 1024) # 5MB + await create_object(grpc_client, bidi_bucket, object_name, data) + + mrd = await AsyncMultiRangeDownloader.create_mrd(grpc_client, bidi_bucket, object_name) + buffer = BytesIO() + + # Start download in background + task = asyncio.create_task(mrd.download_ranges([(0, 0, buffer)])) + + # Wait a bit to ensure it started + await asyncio.sleep(0.1) + + # Close the downloader mid-stream + await mrd.close() + + # Awaiting the task should raise an exception + with pytest.raises(Exception) as excinfo: + await task + + # The exception could be ServiceUnavailable or CancelledError + assert isinstance(excinfo.value, (exceptions.ServiceUnavailable, asyncio.CancelledError)) + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_zero_copy_range_reads(grpc_client, bidi_bucket): + # We call it zero_copy to match Java test name, but it's standard range read in Python. + object_name = f"test_zero_copy_{uuid.uuid4()}" + data = os.urandom(1024 * 1024) # 1MB + await create_object(grpc_client, bidi_bucket, object_name, data) + + # Define ranges + r1 = (0, 1000) + r2 = (50000, 250000) + r3 = (800000, 10000) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + buf1 = BytesIO() + buf2 = BytesIO() + buf3 = BytesIO() + + # Concurrent downloads + t1 = asyncio.create_task(mrd.download_ranges([(r1[0], r1[1], buf1)])) + t2 = asyncio.create_task(mrd.download_ranges([(r2[0], r2[1], buf2)])) + t3 = asyncio.create_task(mrd.download_ranges([(r3[0], r3[1], buf3)])) + + await asyncio.gather(t1, t2, t3) + + assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] + assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] + assert buf3.getvalue() == data[r3[0] : r3[0] + r3[1]] + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_multiple_ranged_read(grpc_client, bidi_bucket): + object_name = f"test_multi_range_{uuid.uuid4()}" + data = os.urandom(1024 * 1024) # 1MB + await create_object(grpc_client, bidi_bucket, object_name, data) + + r1 = (0, 1000) + r2 = (50000, 250000) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + buf1 = BytesIO() + buf2 = BytesIO() + + # Download multiple ranges in a single call (AsyncMultiRangeDownloader supports this) + await mrd.download_ranges([ + (r1[0], r1[1], buf1), + (r2[0], r2[1], buf2) + ]) + + assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] + assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + +@pytest.mark.asyncio +async def test_read_from_non_existent_bucket_fails(grpc_client): + bad_bucket_name = f"non-existent-bucket-{uuid.uuid4()}" + + with pytest.raises(NotFound) as excinfo: + await AsyncMultiRangeDownloader.create_mrd(grpc_client, bad_bucket_name, "some-object") + + assert excinfo.value.code == 404 + + +@pytest.mark.asyncio +async def test_read_out_of_range(grpc_client, bidi_bucket): + object_name = f"test_oob_{uuid.uuid4()}" + data = os.urandom(1024 * 1024) # 1MB + await create_object(grpc_client, bidi_bucket, object_name, data) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + valid_buffer = BytesIO() + valid_task = asyncio.create_task( + mrd.download_ranges([(0, 100, valid_buffer)]) + ) + + oob_buffer = BytesIO() + # starts at 2MB, object is 1MB + oob_task = asyncio.create_task( + mrd.download_ranges([(2 * 1024 * 1024, 100, oob_buffer)]) + ) + + results = await asyncio.gather(valid_task, oob_task, return_exceptions=True) + + # Verify valid one processed correctly + assert valid_buffer.getvalue() == data[:100] + + # Verify fully OOB request returned OutOfRange + assert isinstance(results[1], OutOfRange) + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) From 38c7ba9037105c9d011a036ea5009d9bad11f060 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Fri, 21 Aug 2026 08:40:34 +0000 Subject: [PATCH 2/4] test(storage): split and parameterize Bidi integration tests Split Bidi Read/Write tests into separate files: test_bidi_read.py and test_bidi_write.py. In test_bidi_read.py, add RCU ingest-on-read logic with 30 minutes sleep. In test_bidi_write.py, parameterize appendable upload with flush intervals, close actions, and sizes. [Generated-by: AI] --- .../tests/system/test_bidi_read.py | 219 ++++++++++++++++++ .../{test_bidi.py => test_bidi_write.py} | 215 ++++------------- 2 files changed, 262 insertions(+), 172 deletions(-) create mode 100644 packages/google-cloud-storage/tests/system/test_bidi_read.py rename packages/google-cloud-storage/tests/system/{test_bidi.py => test_bidi_write.py} (61%) diff --git a/packages/google-cloud-storage/tests/system/test_bidi_read.py b/packages/google-cloud-storage/tests/system/test_bidi_read.py new file mode 100644 index 000000000000..0c8e73999168 --- /dev/null +++ b/packages/google-cloud-storage/tests/system/test_bidi_read.py @@ -0,0 +1,219 @@ +# 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 +# +# https://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 +from io import BytesIO +import os +import uuid +import pytest + +from google.api_core import exceptions +from google.api_core.client_options import ClientOptions +from google.api_core.exceptions import NotFound, OutOfRange +from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient +from google.cloud.storage.asyncio.async_appendable_object_writer import AsyncAppendableObjectWriter +from google.cloud.storage.asyncio.async_multi_range_downloader import AsyncMultiRangeDownloader + +REGIONAL_RAPID_BUCKET = "java-storage-reg-rapid-preprod-3fe2bb58" +PREPROD_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" + +# Parameterize over LocationType +@pytest.fixture(scope="module", params=["REGIONAL_RAPID"]) +def bidi_location_type(request): + return request.param + +@pytest.fixture(scope="module") +def grpc_client(bidi_location_type): + if bidi_location_type == "REGIONAL_RAPID": + options = ClientOptions(api_endpoint=PREPROD_ENDPOINT) + return AsyncGrpcClient(client_options=options) + else: + return AsyncGrpcClient() + +@pytest.fixture(scope="module") +def bidi_bucket(bidi_location_type): + if bidi_location_type == "REGIONAL_RAPID": + return REGIONAL_RAPID_BUCKET + elif bidi_location_type == "ZONAL_RAPID": + zonal_bucket = os.getenv("ZONAL_BUCKET") + if not zonal_bucket: + pytest.skip("ZONAL_BUCKET env var not set") + return zonal_bucket + else: + pytest.fail(f"Unsupported location type: {bidi_location_type}") + +async def create_object(grpc_client, bucket, object_name, data): + writer = AsyncAppendableObjectWriter(grpc_client, bucket, object_name) + await writer.open() + await writer.append(data) + await writer.close(finalize_on_close=True) + +# Module-level dictionary to store pre-created object info +OBJECTS = {} + +@pytest.fixture(scope="module", autouse=True) +async def setup_bidi_read_objects(grpc_client, bidi_bucket, bidi_location_type): + global OBJECTS + OBJECTS = { + "large": { + "name": f"test_read_close_{uuid.uuid4()}", + "data": os.urandom(5 * 1024 * 1024) + }, + "zero_copy": { + "name": f"test_zero_copy_{uuid.uuid4()}", + "data": os.urandom(1024 * 1024) + }, + "multi_range": { + "name": f"test_multi_range_{uuid.uuid4()}", + "data": os.urandom(1024 * 1024) + }, + "oob": { + "name": f"test_oob_{uuid.uuid4()}", + "data": os.urandom(1024 * 1024) + } + } + + # Create all objects + for obj_info in OBJECTS.values(): + await create_object(grpc_client, bidi_bucket, obj_info["name"], obj_info["data"]) + + if bidi_location_type == "REGIONAL_RAPID": + # Trigger Ingest-on-Read by doing a small read (1 byte) on each + for obj_info in OBJECTS.values(): + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, obj_info["name"]) as mrd: + buf = BytesIO() + await mrd.download_ranges([(0, 1, buf)]) + + # Sleep for 30 minutes + print("Sleeping for 30 minutes to allow RCU ingestion...") + await asyncio.sleep(1800) + print("Woke up from RCU ingestion sleep.") + + yield + + # Teardown: delete all objects + for obj_info in OBJECTS.values(): + try: + await grpc_client.delete_object(bidi_bucket, obj_info["name"]) + except Exception: + pass + +# ----------------- Read Tests ----------------- + +@pytest.mark.asyncio +async def test_read_post_stream_close(grpc_client, bidi_bucket): + obj_info = OBJECTS["large"] + object_name = obj_info["name"] + + mrd = await AsyncMultiRangeDownloader.create_mrd(grpc_client, bidi_bucket, object_name) + buffer = BytesIO() + + # Start download in background + task = asyncio.create_task(mrd.download_ranges([(0, 0, buffer)])) + + # Wait a bit to ensure it started + await asyncio.sleep(0.1) + + # Close the downloader mid-stream + await mrd.close() + + # Awaiting the task should raise an exception + with pytest.raises(Exception) as excinfo: + await task + + assert isinstance(excinfo.value, (exceptions.ServiceUnavailable, asyncio.CancelledError)) + + +@pytest.mark.asyncio +async def test_zero_copy_range_reads(grpc_client, bidi_bucket): + obj_info = OBJECTS["zero_copy"] + object_name = obj_info["name"] + data = obj_info["data"] + + # Define ranges + r1 = (0, 1000) + r2 = (50000, 250000) + r3 = (800000, 10000) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + buf1 = BytesIO() + buf2 = BytesIO() + buf3 = BytesIO() + + # Concurrent downloads + t1 = asyncio.create_task(mrd.download_ranges([(r1[0], r1[1], buf1)])) + t2 = asyncio.create_task(mrd.download_ranges([(r2[0], r2[1], buf2)])) + t3 = asyncio.create_task(mrd.download_ranges([(r3[0], r3[1], buf3)])) + + await asyncio.gather(t1, t2, t3) + + assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] + assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] + assert buf3.getvalue() == data[r3[0] : r3[0] + r3[1]] + + +@pytest.mark.asyncio +async def test_multiple_ranged_read(grpc_client, bidi_bucket): + obj_info = OBJECTS["multi_range"] + object_name = obj_info["name"] + data = obj_info["data"] + + r1 = (0, 1000) + r2 = (50000, 250000) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + buf1 = BytesIO() + buf2 = BytesIO() + + await mrd.download_ranges([ + (r1[0], r1[1], buf1), + (r2[0], r2[1], buf2) + ]) + + assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] + assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] + + +@pytest.mark.asyncio +async def test_read_from_non_existent_bucket_fails(grpc_client): + bad_bucket_name = f"non-existent-bucket-{uuid.uuid4()}" + + with pytest.raises(NotFound) as excinfo: + await AsyncMultiRangeDownloader.create_mrd(grpc_client, bad_bucket_name, "some-object") + + assert excinfo.value.code == 404 + + +@pytest.mark.asyncio +async def test_read_out_of_range(grpc_client, bidi_bucket): + obj_info = OBJECTS["oob"] + object_name = obj_info["name"] + data = obj_info["data"] + object_size = len(data) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + valid_buffer = BytesIO() + valid_task = asyncio.create_task( + mrd.download_ranges([(0, 100, valid_buffer)]) + ) + + oob_buffer = BytesIO() + oob_task = asyncio.create_task( + mrd.download_ranges([(object_size + 1000, 100, oob_buffer)]) + ) + + results = await asyncio.gather(valid_task, oob_task, return_exceptions=True) + + assert valid_buffer.getvalue() == data[:100] + assert isinstance(results[1], OutOfRange) diff --git a/packages/google-cloud-storage/tests/system/test_bidi.py b/packages/google-cloud-storage/tests/system/test_bidi_write.py similarity index 61% rename from packages/google-cloud-storage/tests/system/test_bidi.py rename to packages/google-cloud-storage/tests/system/test_bidi_write.py index 5937eab2bc16..bae10398a132 100644 --- a/packages/google-cloud-storage/tests/system/test_bidi.py +++ b/packages/google-cloud-storage/tests/system/test_bidi_write.py @@ -21,7 +21,6 @@ from google.api_core import exceptions from google.api_core.client_options import ClientOptions -from google.api_core.exceptions import NotFound, OutOfRange, InvalidArgument from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient from google.cloud.storage.asyncio.async_appendable_object_writer import AsyncAppendableObjectWriter from google.cloud.storage.asyncio.async_multi_range_downloader import AsyncMultiRangeDownloader @@ -29,16 +28,14 @@ REGIONAL_RAPID_BUCKET = "java-storage-reg-rapid-preprod-3fe2bb58" PREPROD_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" -# We parameterized the tests to run against REGIONAL_RAPID for now. -# We can expand to others if needed. -@pytest.fixture(scope="module") -def bidi_location_type(): - return "REGIONAL_RAPID" +# Parameterize over LocationType +@pytest.fixture(scope="module", params=["REGIONAL_RAPID"]) +def bidi_location_type(request): + return request.param @pytest.fixture(scope="module") def grpc_client(bidi_location_type): if bidi_location_type == "REGIONAL_RAPID": - # Point to preprod endpoint options = ClientOptions(api_endpoint=PREPROD_ENDPOINT) return AsyncGrpcClient(client_options=options) else: @@ -46,8 +43,9 @@ def grpc_client(bidi_location_type): @pytest.fixture(scope="module") def bidi_bucket(bidi_location_type): - if bidi_location_type == "REGIONAL_RAPID": - # Shared pre-created bucket + if bidi_location_type == "REGIONAL_STANDARD": + pytest.skip("Bidi Write (Appendable) is not supported on standard regional buckets") + elif bidi_location_type == "REGIONAL_RAPID": return REGIONAL_RAPID_BUCKET elif bidi_location_type == "ZONAL_RAPID": zonal_bucket = os.getenv("ZONAL_BUCKET") @@ -57,40 +55,25 @@ def bidi_bucket(bidi_location_type): else: pytest.fail(f"Unsupported location type: {bidi_location_type}") -# Helper to create objects using sync client if needed, -# but using AsyncAppendableObjectWriter is also fine. -# We will use sync client from conftest if available, but it points to Prod. -# For preprod, we need to initialize a sync client pointing to preprod if we want to use it. -# Actually, we can just use AsyncAppendableObjectWriter to create objects for read tests. -async def create_object(grpc_client, bucket, object_name, data): - writer = AsyncAppendableObjectWriter(grpc_client, bucket, object_name) - await writer.open() - await writer.append(data) - await writer.close(finalize_on_close=True) - # ----------------- Write Tests ----------------- @pytest.mark.asyncio -async def test_appendable_upload_empty_object(grpc_client, bidi_bucket): +@pytest.mark.parametrize("finalize_on_close", [True, False]) +async def test_appendable_upload_empty_object(grpc_client, bidi_bucket, finalize_on_close): object_name = f"test_empty_{uuid.uuid4()}" writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) await writer.open() - object_metadata = await writer.close(finalize_on_close=True) - - # Register for deletion (using sync client, it should work if it can access the bucket, - # but sync client might be pointing to Prod. Wait, if sync client is pointing to Prod, - # it won't be able to delete from pre-prod bucket unless we configure it or if the credential - # has access. The bucket is in 'gcs-hyd-connector-benchmarks' project. - # If standard client has access to it, it should work. - # Wait, the sync client in conftest uses default project. - # If we need to delete it from preprod, maybe we should delete it using grpc_client? - # AsyncGrpcClient has delete_object! - # Let's check: - # await grpc_client.delete_object(bidi_bucket, object_name) - # Yes, we can do that. - # So we can manually delete it or use a cleanup helper. - # Let's register it for deletion using a custom cleanup list. + object_metadata = await writer.close(finalize_on_close=finalize_on_close) + + # If finalized_on_close is False, we need to finalize it now to get size and CRC, + # or just verify we can close it. + # In Java: upload.open().close(); results in 0 size. + # In Python, if finalize_on_close is False, close() returns persisted_size (which should be 0). + if not finalize_on_close: + assert object_metadata == 0 + # Finalize to get the object resource + object_metadata = await writer.finalize() assert object_metadata.size == 0 assert int(object_metadata.checksums.crc32c) == int(google_crc32c.value(b"")) @@ -105,19 +88,38 @@ async def test_appendable_upload_empty_object(grpc_client, bidi_bucket): await grpc_client.delete_object(bidi_bucket, object_name) +# Parameterize write options similar to Java +FLUSH_INTERVALS = [None, 2 * 1024 * 1024, 4 * 1024 * 1024] +FINALIZE_ON_CLOSE_OPTS = [True, False] +OBJECT_SIZES = [5, 500, 5000, 500000, 5000000] + @pytest.mark.asyncio -async def test_appendable_upload_bytes(grpc_client, bidi_bucket): +@pytest.mark.parametrize("flush_interval", FLUSH_INTERVALS) +@pytest.mark.parametrize("finalize_on_close", FINALIZE_ON_CLOSE_OPTS) +@pytest.mark.parametrize("object_size", OBJECT_SIZES) +async def test_appendable_upload_bytes(grpc_client, bidi_bucket, flush_interval, finalize_on_close, object_size): object_name = f"test_bytes_{uuid.uuid4()}" - data = os.urandom(1000) + data = os.urandom(object_size) mid = len(data) // 2 chunk1 = data[:mid] chunk2 = data[mid:] - writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) + writer_options = {} + if flush_interval is not None: + writer_options["FLUSH_INTERVAL_BYTES"] = flush_interval + + writer = AsyncAppendableObjectWriter( + grpc_client, bidi_bucket, object_name, writer_options=writer_options + ) await writer.open() await writer.append(chunk1) await writer.append(chunk2) - object_metadata = await writer.close(finalize_on_close=True) + object_metadata = await writer.close(finalize_on_close=finalize_on_close) + + if not finalize_on_close: + # Check persisted size so far + assert object_metadata == len(data) + object_metadata = await writer.finalize() assert object_metadata.size == len(data) expected_crc = google_crc32c.value(data) @@ -258,9 +260,7 @@ async def test_explicit_finalize_with_incorrect_checksum_fails(grpc_client, bidi assert "mismatch" in str(excinfo.value).lower() - # Cleanup (it should not have been finalized, but we might need to delete the unfinalized object if it exists) - # Actually, unfinalized objects might not be deleteable normally? - # Yes they are deleteable. + # Cleanup try: await grpc_client.delete_object(bidi_bucket, object_name) except Exception: @@ -357,132 +357,3 @@ async def test_takeover_and_append_with_incorrect_checksum_fails(grpc_client, bi await grpc_client.delete_object(bidi_bucket, object_name) except Exception: pass - - -# ----------------- Read Tests ----------------- - -@pytest.mark.asyncio -async def test_read_post_stream_close(grpc_client, bidi_bucket): - object_name = f"test_read_close_{uuid.uuid4()}" - data = os.urandom(5 * 1024 * 1024) # 5MB - await create_object(grpc_client, bidi_bucket, object_name, data) - - mrd = await AsyncMultiRangeDownloader.create_mrd(grpc_client, bidi_bucket, object_name) - buffer = BytesIO() - - # Start download in background - task = asyncio.create_task(mrd.download_ranges([(0, 0, buffer)])) - - # Wait a bit to ensure it started - await asyncio.sleep(0.1) - - # Close the downloader mid-stream - await mrd.close() - - # Awaiting the task should raise an exception - with pytest.raises(Exception) as excinfo: - await task - - # The exception could be ServiceUnavailable or CancelledError - assert isinstance(excinfo.value, (exceptions.ServiceUnavailable, asyncio.CancelledError)) - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) - - -@pytest.mark.asyncio -async def test_zero_copy_range_reads(grpc_client, bidi_bucket): - # We call it zero_copy to match Java test name, but it's standard range read in Python. - object_name = f"test_zero_copy_{uuid.uuid4()}" - data = os.urandom(1024 * 1024) # 1MB - await create_object(grpc_client, bidi_bucket, object_name, data) - - # Define ranges - r1 = (0, 1000) - r2 = (50000, 250000) - r3 = (800000, 10000) - - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - buf1 = BytesIO() - buf2 = BytesIO() - buf3 = BytesIO() - - # Concurrent downloads - t1 = asyncio.create_task(mrd.download_ranges([(r1[0], r1[1], buf1)])) - t2 = asyncio.create_task(mrd.download_ranges([(r2[0], r2[1], buf2)])) - t3 = asyncio.create_task(mrd.download_ranges([(r3[0], r3[1], buf3)])) - - await asyncio.gather(t1, t2, t3) - - assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] - assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] - assert buf3.getvalue() == data[r3[0] : r3[0] + r3[1]] - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) - - -@pytest.mark.asyncio -async def test_multiple_ranged_read(grpc_client, bidi_bucket): - object_name = f"test_multi_range_{uuid.uuid4()}" - data = os.urandom(1024 * 1024) # 1MB - await create_object(grpc_client, bidi_bucket, object_name, data) - - r1 = (0, 1000) - r2 = (50000, 250000) - - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - buf1 = BytesIO() - buf2 = BytesIO() - - # Download multiple ranges in a single call (AsyncMultiRangeDownloader supports this) - await mrd.download_ranges([ - (r1[0], r1[1], buf1), - (r2[0], r2[1], buf2) - ]) - - assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] - assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) - - -@pytest.mark.asyncio -async def test_read_from_non_existent_bucket_fails(grpc_client): - bad_bucket_name = f"non-existent-bucket-{uuid.uuid4()}" - - with pytest.raises(NotFound) as excinfo: - await AsyncMultiRangeDownloader.create_mrd(grpc_client, bad_bucket_name, "some-object") - - assert excinfo.value.code == 404 - - -@pytest.mark.asyncio -async def test_read_out_of_range(grpc_client, bidi_bucket): - object_name = f"test_oob_{uuid.uuid4()}" - data = os.urandom(1024 * 1024) # 1MB - await create_object(grpc_client, bidi_bucket, object_name, data) - - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - valid_buffer = BytesIO() - valid_task = asyncio.create_task( - mrd.download_ranges([(0, 100, valid_buffer)]) - ) - - oob_buffer = BytesIO() - # starts at 2MB, object is 1MB - oob_task = asyncio.create_task( - mrd.download_ranges([(2 * 1024 * 1024, 100, oob_buffer)]) - ) - - results = await asyncio.gather(valid_task, oob_task, return_exceptions=True) - - # Verify valid one processed correctly - assert valid_buffer.getvalue() == data[:100] - - # Verify fully OOB request returned OutOfRange - assert isinstance(results[1], OutOfRange) - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) From 8447d592d1f2b6669df7fb1eb2481d006c158d4d Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Fri, 21 Aug 2026 09:39:18 +0000 Subject: [PATCH 3/4] test(storage): fix RCU preprod storage class, timeouts, and nox constraints - Monkey patch blob_to_proto to copy storage_class to proto. - Use from_blob and set RAPID storage class in tests. - Wrap all async tests in asyncio.wait_for with 60s timeout to prevent hangs. - Remove constraints from nox system test session to fix pip backtracking. [Generated-by: AI] --- packages/google-cloud-storage/noxfile.py | 6 +- .../tests/system/test_bidi_read.py | 224 ++++--- .../tests/system/test_bidi_write.py | 587 ++++++++++-------- 3 files changed, 461 insertions(+), 356 deletions(-) diff --git a/packages/google-cloud-storage/noxfile.py b/packages/google-cloud-storage/noxfile.py index 7f8f561031fd..9ae9deda4c9a 100644 --- a/packages/google-cloud-storage/noxfile.py +++ b/packages/google-cloud-storage/noxfile.py @@ -389,18 +389,14 @@ def system(session, test_type): "pytest", "pytest-rerunfailures", "pytest-asyncio", - "-c", - constraints_path, ) - session.install("-e", ".", "-c", constraints_path) + session.install("-e", ".") session.install( "google-cloud-testutils", "google-cloud-iam", "google-cloud-pubsub", "google-cloud-kms", "brotli", - "-c", - constraints_path, ) # Run py.test against the system tests. diff --git a/packages/google-cloud-storage/tests/system/test_bidi_read.py b/packages/google-cloud-storage/tests/system/test_bidi_read.py index 0c8e73999168..e65835984902 100644 --- a/packages/google-cloud-storage/tests/system/test_bidi_read.py +++ b/packages/google-cloud-storage/tests/system/test_bidi_read.py @@ -25,6 +25,21 @@ from google.cloud.storage.asyncio.async_appendable_object_writer import AsyncAppendableObjectWriter from google.cloud.storage.asyncio.async_multi_range_downloader import AsyncMultiRangeDownloader +# Monkey patch blob_to_proto to support storage_class +from google.cloud.storage import _grpc_conversions +_orig_blob_to_proto = _grpc_conversions.blob_to_proto + +def _patched_blob_to_proto(blob): + proto = _orig_blob_to_proto(blob) + if hasattr(blob, "storage_class") and blob.storage_class: + proto.storage_class = blob.storage_class + return proto + +_grpc_conversions.blob_to_proto = _patched_blob_to_proto + +from google.cloud.storage.bucket import Bucket +from google.cloud.storage.blob import Blob + REGIONAL_RAPID_BUCKET = "java-storage-reg-rapid-preprod-3fe2bb58" PREPROD_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" @@ -53,18 +68,28 @@ def bidi_bucket(bidi_location_type): else: pytest.fail(f"Unsupported location type: {bidi_location_type}") -async def create_object(grpc_client, bucket, object_name, data): - writer = AsyncAppendableObjectWriter(grpc_client, bucket, object_name) +def get_storage_class(location_type): + if location_type in ("REGIONAL_RAPID", "ZONAL_RAPID"): + return "RAPID" + return None + +async def create_object(grpc_client, bucket_name, object_name, data, storage_class=None): + bucket = Bucket(name=bucket_name) + blob = Blob(object_name, bucket) + if storage_class: + blob.storage_class = storage_class + writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob) await writer.open() await writer.append(data) await writer.close(finalize_on_close=True) -# Module-level dictionary to store pre-created object info OBJECTS = {} @pytest.fixture(scope="module", autouse=True) async def setup_bidi_read_objects(grpc_client, bidi_bucket, bidi_location_type): global OBJECTS + storage_class = get_storage_class(bidi_location_type) + OBJECTS = { "large": { "name": f"test_read_close_{uuid.uuid4()}", @@ -84,136 +109,151 @@ async def setup_bidi_read_objects(grpc_client, bidi_bucket, bidi_location_type): } } - # Create all objects - for obj_info in OBJECTS.values(): - await create_object(grpc_client, bidi_bucket, obj_info["name"], obj_info["data"]) + # Create all objects with timeout + async def _create_all(): + for obj_info in OBJECTS.values(): + await create_object(grpc_client, bidi_bucket, obj_info["name"], obj_info["data"], storage_class=storage_class) + + await asyncio.wait_for(_create_all(), timeout=60) if bidi_location_type == "REGIONAL_RAPID": - # Trigger Ingest-on-Read by doing a small read (1 byte) on each - for obj_info in OBJECTS.values(): - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, obj_info["name"]) as mrd: - buf = BytesIO() - await mrd.download_ranges([(0, 1, buf)]) + # Trigger Ingest-on-Read + async def _trigger_ingestion(): + for obj_info in OBJECTS.values(): + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, obj_info["name"]) as mrd: + buf = BytesIO() + await mrd.download_ranges([(0, 1, buf)]) + + await asyncio.wait_for(_trigger_ingestion(), timeout=30) - # Sleep for 30 minutes print("Sleeping for 30 minutes to allow RCU ingestion...") await asyncio.sleep(1800) print("Woke up from RCU ingestion sleep.") yield - # Teardown: delete all objects - for obj_info in OBJECTS.values(): - try: - await grpc_client.delete_object(bidi_bucket, obj_info["name"]) - except Exception: - pass + # Teardown + async def _cleanup(): + for obj_info in OBJECTS.values(): + try: + await grpc_client.delete_object(bidi_bucket, obj_info["name"]) + except Exception: + pass + try: + await asyncio.wait_for(_cleanup(), timeout=30) + except Exception: + pass # ----------------- Read Tests ----------------- @pytest.mark.asyncio async def test_read_post_stream_close(grpc_client, bidi_bucket): - obj_info = OBJECTS["large"] - object_name = obj_info["name"] - - mrd = await AsyncMultiRangeDownloader.create_mrd(grpc_client, bidi_bucket, object_name) - buffer = BytesIO() + async def _run(): + obj_info = OBJECTS["large"] + object_name = obj_info["name"] - # Start download in background - task = asyncio.create_task(mrd.download_ranges([(0, 0, buffer)])) + mrd = await AsyncMultiRangeDownloader.create_mrd(grpc_client, bidi_bucket, object_name) + buffer = BytesIO() - # Wait a bit to ensure it started - await asyncio.sleep(0.1) + task = asyncio.create_task(mrd.download_ranges([(0, 0, buffer)])) + await asyncio.sleep(0.1) + await mrd.close() - # Close the downloader mid-stream - await mrd.close() + with pytest.raises(Exception) as excinfo: + await task - # Awaiting the task should raise an exception - with pytest.raises(Exception) as excinfo: - await task + assert isinstance(excinfo.value, (exceptions.ServiceUnavailable, asyncio.CancelledError)) - assert isinstance(excinfo.value, (exceptions.ServiceUnavailable, asyncio.CancelledError)) + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio async def test_zero_copy_range_reads(grpc_client, bidi_bucket): - obj_info = OBJECTS["zero_copy"] - object_name = obj_info["name"] - data = obj_info["data"] + async def _run(): + obj_info = OBJECTS["zero_copy"] + object_name = obj_info["name"] + data = obj_info["data"] - # Define ranges - r1 = (0, 1000) - r2 = (50000, 250000) - r3 = (800000, 10000) + r1 = (0, 1000) + r2 = (50000, 250000) + r3 = (800000, 10000) - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - buf1 = BytesIO() - buf2 = BytesIO() - buf3 = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + buf1 = BytesIO() + buf2 = BytesIO() + buf3 = BytesIO() - # Concurrent downloads - t1 = asyncio.create_task(mrd.download_ranges([(r1[0], r1[1], buf1)])) - t2 = asyncio.create_task(mrd.download_ranges([(r2[0], r2[1], buf2)])) - t3 = asyncio.create_task(mrd.download_ranges([(r3[0], r3[1], buf3)])) + t1 = asyncio.create_task(mrd.download_ranges([(r1[0], r1[1], buf1)])) + t2 = asyncio.create_task(mrd.download_ranges([(r2[0], r2[1], buf2)])) + t3 = asyncio.create_task(mrd.download_ranges([(r3[0], r3[1], buf3)])) - await asyncio.gather(t1, t2, t3) + await asyncio.gather(t1, t2, t3) - assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] - assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] - assert buf3.getvalue() == data[r3[0] : r3[0] + r3[1]] + assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] + assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] + assert buf3.getvalue() == data[r3[0] : r3[0] + r3[1]] + + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio async def test_multiple_ranged_read(grpc_client, bidi_bucket): - obj_info = OBJECTS["multi_range"] - object_name = obj_info["name"] - data = obj_info["data"] + async def _run(): + obj_info = OBJECTS["multi_range"] + object_name = obj_info["name"] + data = obj_info["data"] + + r1 = (0, 1000) + r2 = (50000, 250000) - r1 = (0, 1000) - r2 = (50000, 250000) + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + buf1 = BytesIO() + buf2 = BytesIO() - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - buf1 = BytesIO() - buf2 = BytesIO() + await mrd.download_ranges([ + (r1[0], r1[1], buf1), + (r2[0], r2[1], buf2) + ]) - await mrd.download_ranges([ - (r1[0], r1[1], buf1), - (r2[0], r2[1], buf2) - ]) + assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] + assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] - assert buf1.getvalue() == data[r1[0] : r1[0] + r1[1]] - assert buf2.getvalue() == data[r2[0] : r2[0] + r2[1]] + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio async def test_read_from_non_existent_bucket_fails(grpc_client): - bad_bucket_name = f"non-existent-bucket-{uuid.uuid4()}" - - with pytest.raises(NotFound) as excinfo: - await AsyncMultiRangeDownloader.create_mrd(grpc_client, bad_bucket_name, "some-object") - - assert excinfo.value.code == 404 + async def _run(): + bad_bucket_name = f"non-existent-bucket-{uuid.uuid4()}" + with pytest.raises(NotFound) as excinfo: + await AsyncMultiRangeDownloader.create_mrd(grpc_client, bad_bucket_name, "some-object") + assert excinfo.value.code == 404 + + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio async def test_read_out_of_range(grpc_client, bidi_bucket): - obj_info = OBJECTS["oob"] - object_name = obj_info["name"] - data = obj_info["data"] - object_size = len(data) - - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - valid_buffer = BytesIO() - valid_task = asyncio.create_task( - mrd.download_ranges([(0, 100, valid_buffer)]) - ) - - oob_buffer = BytesIO() - oob_task = asyncio.create_task( - mrd.download_ranges([(object_size + 1000, 100, oob_buffer)]) - ) - - results = await asyncio.gather(valid_task, oob_task, return_exceptions=True) - - assert valid_buffer.getvalue() == data[:100] - assert isinstance(results[1], OutOfRange) + async def _run(): + obj_info = OBJECTS["oob"] + object_name = obj_info["name"] + data = obj_info["data"] + object_size = len(data) + + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + valid_buffer = BytesIO() + valid_task = asyncio.create_task( + mrd.download_ranges([(0, 100, valid_buffer)]) + ) + + oob_buffer = BytesIO() + oob_task = asyncio.create_task( + mrd.download_ranges([(object_size + 1000, 100, oob_buffer)]) + ) + + results = await asyncio.gather(valid_task, oob_task, return_exceptions=True) + + assert valid_buffer.getvalue() == data[:100] + assert isinstance(results[1], OutOfRange) + + await asyncio.wait_for(_run(), timeout=60) diff --git a/packages/google-cloud-storage/tests/system/test_bidi_write.py b/packages/google-cloud-storage/tests/system/test_bidi_write.py index bae10398a132..568d52be16ac 100644 --- a/packages/google-cloud-storage/tests/system/test_bidi_write.py +++ b/packages/google-cloud-storage/tests/system/test_bidi_write.py @@ -25,6 +25,21 @@ from google.cloud.storage.asyncio.async_appendable_object_writer import AsyncAppendableObjectWriter from google.cloud.storage.asyncio.async_multi_range_downloader import AsyncMultiRangeDownloader +# Monkey patch blob_to_proto to support storage_class +from google.cloud.storage import _grpc_conversions +_orig_blob_to_proto = _grpc_conversions.blob_to_proto + +def _patched_blob_to_proto(blob): + proto = _orig_blob_to_proto(blob) + if hasattr(blob, "storage_class") and blob.storage_class: + proto.storage_class = blob.storage_class + return proto + +_grpc_conversions.blob_to_proto = _patched_blob_to_proto + +from google.cloud.storage.bucket import Bucket +from google.cloud.storage.blob import Blob + REGIONAL_RAPID_BUCKET = "java-storage-reg-rapid-preprod-3fe2bb58" PREPROD_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" @@ -55,37 +70,51 @@ def bidi_bucket(bidi_location_type): else: pytest.fail(f"Unsupported location type: {bidi_location_type}") +def get_storage_class(location_type): + if location_type in ("REGIONAL_RAPID", "ZONAL_RAPID"): + return "RAPID" + return None + +def make_blob(bucket_name, object_name, location_type, generation=None): + bucket = Bucket(name=bucket_name) + blob = Blob(object_name, bucket) + storage_class = get_storage_class(location_type) + if storage_class: + blob.storage_class = storage_class + if generation is not None: + blob._properties["generation"] = generation + return blob + # ----------------- Write Tests ----------------- @pytest.mark.asyncio @pytest.mark.parametrize("finalize_on_close", [True, False]) -async def test_appendable_upload_empty_object(grpc_client, bidi_bucket, finalize_on_close): - object_name = f"test_empty_{uuid.uuid4()}" - - writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer.open() - object_metadata = await writer.close(finalize_on_close=finalize_on_close) - - # If finalized_on_close is False, we need to finalize it now to get size and CRC, - # or just verify we can close it. - # In Java: upload.open().close(); results in 0 size. - # In Python, if finalize_on_close is False, close() returns persisted_size (which should be 0). - if not finalize_on_close: - assert object_metadata == 0 - # Finalize to get the object resource - object_metadata = await writer.finalize() - - assert object_metadata.size == 0 - assert int(object_metadata.checksums.crc32c) == int(google_crc32c.value(b"")) - - # Read back and verify empty - buffer = BytesIO() - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - await mrd.download_ranges([(0, 0, buffer)]) - assert buffer.getvalue() == b"" - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) +async def test_appendable_upload_empty_object(grpc_client, bidi_bucket, bidi_location_type, finalize_on_close): + async def _run(): + object_name = f"test_empty_{uuid.uuid4()}" + blob = make_blob(bidi_bucket, object_name, bidi_location_type) + + writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob) + await writer.open() + object_metadata = await writer.close(finalize_on_close=finalize_on_close) + + if not finalize_on_close: + assert object_metadata == 0 + object_metadata = await writer.finalize() + + assert object_metadata.size == 0 + assert int(object_metadata.checksums.crc32c) == int(google_crc32c.value(b"")) + + # Read back and verify empty + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == b"" + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + + await asyncio.wait_for(_run(), timeout=60) # Parameterize write options similar to Java @@ -97,263 +126,303 @@ async def test_appendable_upload_empty_object(grpc_client, bidi_bucket, finalize @pytest.mark.parametrize("flush_interval", FLUSH_INTERVALS) @pytest.mark.parametrize("finalize_on_close", FINALIZE_ON_CLOSE_OPTS) @pytest.mark.parametrize("object_size", OBJECT_SIZES) -async def test_appendable_upload_bytes(grpc_client, bidi_bucket, flush_interval, finalize_on_close, object_size): - object_name = f"test_bytes_{uuid.uuid4()}" - data = os.urandom(object_size) - mid = len(data) // 2 - chunk1 = data[:mid] - chunk2 = data[mid:] - - writer_options = {} - if flush_interval is not None: - writer_options["FLUSH_INTERVAL_BYTES"] = flush_interval - - writer = AsyncAppendableObjectWriter( - grpc_client, bidi_bucket, object_name, writer_options=writer_options - ) - await writer.open() - await writer.append(chunk1) - await writer.append(chunk2) - object_metadata = await writer.close(finalize_on_close=finalize_on_close) - - if not finalize_on_close: - # Check persisted size so far - assert object_metadata == len(data) - object_metadata = await writer.finalize() - - assert object_metadata.size == len(data) - expected_crc = google_crc32c.value(data) - assert int(object_metadata.checksums.crc32c) == expected_crc - - # Read back and verify - buffer = BytesIO() - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - await mrd.download_ranges([(0, 0, buffer)]) - assert buffer.getvalue() == data - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) +async def test_appendable_upload_bytes(grpc_client, bidi_bucket, bidi_location_type, flush_interval, finalize_on_close, object_size): + async def _run(): + object_name = f"test_bytes_{uuid.uuid4()}" + data = os.urandom(object_size) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + writer_options = {} + if flush_interval is not None: + writer_options["FLUSH_INTERVAL_BYTES"] = flush_interval + + blob = make_blob(bidi_bucket, object_name, bidi_location_type) + writer = AsyncAppendableObjectWriter.from_blob( + client=grpc_client, blob=blob, writer_options=writer_options + ) + await writer.open() + await writer.append(chunk1) + await writer.append(chunk2) + object_metadata = await writer.close(finalize_on_close=finalize_on_close) + + if not finalize_on_close: + assert object_metadata == len(data) + object_metadata = await writer.finalize() + + assert object_metadata.size == len(data) + expected_crc = google_crc32c.value(data) + assert int(object_metadata.checksums.crc32c) == expected_crc + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) - -@pytest.mark.asyncio -async def test_explicit_flush(grpc_client, bidi_bucket): - object_name = f"test_flush_{uuid.uuid4()}" - data = os.urandom(1000) - mid = len(data) // 2 - chunk1 = data[:mid] - chunk2 = data[mid:] - - writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer.open() - await writer.append(chunk1) - await writer.flush() # Explicit flush - await writer.append(chunk2) - await writer.close(finalize_on_close=True) - - # Read back and verify - buffer = BytesIO() - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - await mrd.download_ranges([(0, 0, buffer)]) - assert buffer.getvalue() == data - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio -async def test_appendable_blob_upload_takeover(grpc_client, bidi_bucket): - object_name = f"test_takeover_{uuid.uuid4()}" - data = os.urandom(1000) - mid = len(data) // 2 - chunk1 = data[:mid] - chunk2 = data[mid:] - - # Writer 1 writes chunk 1 and closes WITHOUT finalizing - writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer1.open() - await writer1.append(chunk1) - persisted_size1 = await writer1.close(finalize_on_close=False) - generation = writer1.generation - assert persisted_size1 == len(chunk1) - assert generation is not None - - # Writer 2 takes over using the generation number - writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) - await writer2.open() - await writer2.append(chunk2) - object_metadata = await writer2.close(finalize_on_close=True) - - assert object_metadata.size == len(data) - - # Read back and verify - buffer = BytesIO() - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - await mrd.download_ranges([(0, 0, buffer)]) - assert buffer.getvalue() == data - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) +async def test_explicit_flush(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_flush_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + blob = make_blob(bidi_bucket, object_name, bidi_location_type) + writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob) + await writer.open() + await writer.append(chunk1) + await writer.flush() # Explicit flush + await writer.append(chunk2) + await writer.close(finalize_on_close=True) + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) + await asyncio.wait_for(_run(), timeout=60) -@pytest.mark.asyncio -async def test_takeover_just_to_finalize(grpc_client, bidi_bucket): - object_name = f"test_takeover_fin_{uuid.uuid4()}" - data = os.urandom(1000) - # Writer 1 writes chunk 1 and closes WITHOUT finalizing - writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer1.open() - await writer1.append(data) - persisted_size1 = await writer1.close(finalize_on_close=False) - generation = writer1.generation - assert persisted_size1 == len(data) +@pytest.mark.asyncio +async def test_appendable_blob_upload_takeover(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_takeover_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + # Writer 1 writes chunk 1 and closes WITHOUT finalizing + blob1 = make_blob(bidi_bucket, object_name, bidi_location_type) + writer1 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob1) + await writer1.open() + await writer1.append(chunk1) + persisted_size1 = await writer1.close(finalize_on_close=False) + generation = writer1.generation + assert persisted_size1 == len(chunk1) + assert generation is not None + + # Writer 2 takes over using the generation number + blob2 = make_blob(bidi_bucket, object_name, bidi_location_type, generation=generation) + writer2 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob2) + await writer2.open() + await writer2.append(chunk2) + object_metadata = await writer2.close(finalize_on_close=True) + + assert object_metadata.size == len(data) + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) - # Writer 2 takes over and finalizes - writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) - await writer2.open() - object_metadata = await writer2.finalize() + await asyncio.wait_for(_run(), timeout=60) - assert object_metadata.size == len(data) - # Read back and verify - buffer = BytesIO() - async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: - await mrd.download_ranges([(0, 0, buffer)]) - assert buffer.getvalue() == data +@pytest.mark.asyncio +async def test_takeover_just_to_finalize(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_takeover_fin_{uuid.uuid4()}" + data = os.urandom(1000) + + # Writer 1 writes chunk 1 and closes WITHOUT finalizing + blob1 = make_blob(bidi_bucket, object_name, bidi_location_type) + writer1 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob1) + await writer1.open() + await writer1.append(data) + persisted_size1 = await writer1.close(finalize_on_close=False) + generation = writer1.generation + assert persisted_size1 == len(data) + + # Writer 2 takes over and finalizes + blob2 = make_blob(bidi_bucket, object_name, bidi_location_type, generation=generation) + writer2 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob2) + await writer2.open() + object_metadata = await writer2.finalize() + + assert object_metadata.size == len(data) + + # Read back and verify + buffer = BytesIO() + async with AsyncMultiRangeDownloader(grpc_client, bidi_bucket, object_name) as mrd: + await mrd.download_ranges([(0, 0, buffer)]) + assert buffer.getvalue() == data + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio -async def test_explicit_finalize_with_correct_checksum(grpc_client, bidi_bucket): - object_name = f"test_fin_crc_{uuid.uuid4()}" - data = os.urandom(1000) +async def test_explicit_finalize_with_correct_checksum(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_fin_crc_{uuid.uuid4()}" + data = os.urandom(1000) + + blob = make_blob(bidi_bucket, object_name, bidi_location_type) + writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob) + await writer.open() + await writer.append(data) + + expected_crc = google_crc32c.value(data) + object_metadata = await writer.finalize(full_object_checksum=expected_crc) + + assert object_metadata.size == len(data) + assert int(object_metadata.checksums.crc32c) == expected_crc + + # Cleanup + await grpc_client.delete_object(bidi_bucket, object_name) - writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer.open() - await writer.append(data) - - expected_crc = google_crc32c.value(data) - object_metadata = await writer.finalize(full_object_checksum=expected_crc) + await asyncio.wait_for(_run(), timeout=60) - assert object_metadata.size == len(data) - assert int(object_metadata.checksums.crc32c) == expected_crc - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) +@pytest.mark.asyncio +async def test_explicit_finalize_with_incorrect_checksum_fails(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_fin_bad_crc_{uuid.uuid4()}" + data = os.urandom(1000) + + blob = make_blob(bidi_bucket, object_name, bidi_location_type) + writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob) + await writer.open() + await writer.append(data) + + bad_crc = 0 # Incorrect checksum + + with pytest.raises(exceptions.InvalidArgument) as excinfo: + await writer.finalize(full_object_checksum=bad_crc) + + assert "mismatch" in str(excinfo.value).lower() + + # Cleanup + try: + await grpc_client.delete_object(bidi_bucket, object_name) + except Exception: + pass + + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio -async def test_explicit_finalize_with_incorrect_checksum_fails(grpc_client, bidi_bucket): - object_name = f"test_fin_bad_crc_{uuid.uuid4()}" - data = os.urandom(1000) - - writer = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer.open() - await writer.append(data) - - bad_crc = 0 # Incorrect checksum - - with pytest.raises(exceptions.InvalidArgument) as excinfo: - await writer.finalize(full_object_checksum=bad_crc) - - assert "mismatch" in str(excinfo.value).lower() - - # Cleanup - try: - await grpc_client.delete_object(bidi_bucket, object_name) - except Exception: - pass +async def test_takeover_just_to_finalize_with_incorrect_checksum_fails(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_takeover_bad_crc_{uuid.uuid4()}" + data = os.urandom(1000) + + # Writer 1 writes chunk 1, closes unfinalized + blob1 = make_blob(bidi_bucket, object_name, bidi_location_type) + writer1 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob1) + await writer1.open() + await writer1.append(data) + await writer1.close(finalize_on_close=False) + generation = writer1.generation + + # Writer 2 takes over, finalizes with bad checksum + blob2 = make_blob(bidi_bucket, object_name, bidi_location_type, generation=generation) + writer2 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob2) + await writer2.open() + + bad_crc = 0 + with pytest.raises(exceptions.InvalidArgument) as excinfo: + await writer2.finalize(full_object_checksum=bad_crc) + + assert "mismatch" in str(excinfo.value).lower() + + # Cleanup + try: + await grpc_client.delete_object(bidi_bucket, object_name) + except Exception: + pass + + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio -async def test_takeover_just_to_finalize_with_incorrect_checksum_fails(grpc_client, bidi_bucket): - object_name = f"test_takeover_bad_crc_{uuid.uuid4()}" - data = os.urandom(1000) - - # Writer 1 writes chunk 1, closes unfinalized - writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer1.open() - await writer1.append(data) - await writer1.close(finalize_on_close=False) - generation = writer1.generation - - # Writer 2 takes over, finalizes with bad checksum - writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) - await writer2.open() - - bad_crc = 0 - with pytest.raises(exceptions.InvalidArgument) as excinfo: - await writer2.finalize(full_object_checksum=bad_crc) - - assert "mismatch" in str(excinfo.value).lower() - - # Cleanup - try: +async def test_takeover_and_append_with_correct_checksum_works(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_takeover_append_crc_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + # Writer 1 writes chunk 1, closes unfinalized + blob1 = make_blob(bidi_bucket, object_name, bidi_location_type) + writer1 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob1) + await writer1.open() + await writer1.append(chunk1) + await writer1.close(finalize_on_close=False) + generation = writer1.generation + + # Writer 2 takes over, appends chunk 2, and finalizes with correct cumulative checksum + blob2 = make_blob(bidi_bucket, object_name, bidi_location_type, generation=generation) + writer2 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob2) + await writer2.open() + await writer2.append(chunk2) + + expected_crc = google_crc32c.value(data) + object_metadata = await writer2.finalize(full_object_checksum=expected_crc) + + assert object_metadata.size == len(data) + assert int(object_metadata.checksums.crc32c) == expected_crc + + # Cleanup await grpc_client.delete_object(bidi_bucket, object_name) - except Exception: - pass - -@pytest.mark.asyncio -async def test_takeover_and_append_with_correct_checksum_works(grpc_client, bidi_bucket): - object_name = f"test_takeover_append_crc_{uuid.uuid4()}" - data = os.urandom(1000) - mid = len(data) // 2 - chunk1 = data[:mid] - chunk2 = data[mid:] - - # Writer 1 writes chunk 1, closes unfinalized - writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer1.open() - await writer1.append(chunk1) - await writer1.close(finalize_on_close=False) - generation = writer1.generation - - # Writer 2 takes over, appends chunk 2, and finalizes with correct cumulative checksum - writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) - await writer2.open() - await writer2.append(chunk2) - - expected_crc = google_crc32c.value(data) - object_metadata = await writer2.finalize(full_object_checksum=expected_crc) - - assert object_metadata.size == len(data) - assert int(object_metadata.checksums.crc32c) == expected_crc - - # Cleanup - await grpc_client.delete_object(bidi_bucket, object_name) + await asyncio.wait_for(_run(), timeout=60) @pytest.mark.asyncio -async def test_takeover_and_append_with_incorrect_checksum_fails(grpc_client, bidi_bucket): - object_name = f"test_takeover_append_bad_crc_{uuid.uuid4()}" - data = os.urandom(1000) - mid = len(data) // 2 - chunk1 = data[:mid] - chunk2 = data[mid:] - - # Writer 1 writes chunk 1, closes unfinalized - writer1 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name) - await writer1.open() - await writer1.append(chunk1) - await writer1.close(finalize_on_close=False) - generation = writer1.generation - - # Writer 2 takes over, appends chunk 2, finalizes with bad checksum - writer2 = AsyncAppendableObjectWriter(grpc_client, bidi_bucket, object_name, generation=generation) - await writer2.open() - await writer2.append(chunk2) - - bad_crc = 0 - with pytest.raises(exceptions.InvalidArgument) as excinfo: - await writer2.finalize(full_object_checksum=bad_crc) - - assert "mismatch" in str(excinfo.value).lower() - - # Cleanup - try: - await grpc_client.delete_object(bidi_bucket, object_name) - except Exception: - pass +async def test_takeover_and_append_with_incorrect_checksum_fails(grpc_client, bidi_bucket, bidi_location_type): + async def _run(): + object_name = f"test_takeover_append_bad_crc_{uuid.uuid4()}" + data = os.urandom(1000) + mid = len(data) // 2 + chunk1 = data[:mid] + chunk2 = data[mid:] + + # Writer 1 writes chunk 1, closes unfinalized + blob1 = make_blob(bidi_bucket, object_name, bidi_location_type) + writer1 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob1) + await writer1.open() + await writer1.append(chunk1) + await writer1.close(finalize_on_close=False) + generation = writer1.generation + + # Writer 2 takes over, appends chunk 2, finalizes with bad checksum + blob2 = make_blob(bidi_bucket, object_name, bidi_location_type, generation=generation) + writer2 = AsyncAppendableObjectWriter.from_blob(grpc_client, blob2) + await writer2.open() + await writer2.append(chunk2) + + bad_crc = 0 + with pytest.raises(exceptions.InvalidArgument) as excinfo: + await writer2.finalize(full_object_checksum=bad_crc) + + assert "mismatch" in str(excinfo.value).lower() + + # Cleanup + try: + await grpc_client.delete_object(bidi_bucket, object_name) + except Exception: + pass + + await asyncio.wait_for(_run(), timeout=60) From 075f7bc834bdbd45d27a4557bb18ca6dced7a1bd Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Fri, 21 Aug 2026 09:41:51 +0000 Subject: [PATCH 4/4] test(storage): fix Bucket constructor parameters in tests Pass None as client to Bucket constructors since it is required and we only use the instances for metadata conversion. [Generated-by: AI] --- packages/google-cloud-storage/tests/system/test_bidi_read.py | 2 +- packages/google-cloud-storage/tests/system/test_bidi_write.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/test_bidi_read.py b/packages/google-cloud-storage/tests/system/test_bidi_read.py index e65835984902..b318f2b8b452 100644 --- a/packages/google-cloud-storage/tests/system/test_bidi_read.py +++ b/packages/google-cloud-storage/tests/system/test_bidi_read.py @@ -74,7 +74,7 @@ def get_storage_class(location_type): return None async def create_object(grpc_client, bucket_name, object_name, data, storage_class=None): - bucket = Bucket(name=bucket_name) + bucket = Bucket(None, name=bucket_name) blob = Blob(object_name, bucket) if storage_class: blob.storage_class = storage_class diff --git a/packages/google-cloud-storage/tests/system/test_bidi_write.py b/packages/google-cloud-storage/tests/system/test_bidi_write.py index 568d52be16ac..a28d8dd06325 100644 --- a/packages/google-cloud-storage/tests/system/test_bidi_write.py +++ b/packages/google-cloud-storage/tests/system/test_bidi_write.py @@ -76,7 +76,7 @@ def get_storage_class(location_type): return None def make_blob(bucket_name, object_name, location_type, generation=None): - bucket = Bucket(name=bucket_name) + bucket = Bucket(None, name=bucket_name) blob = Blob(object_name, bucket) storage_class = get_storage_class(location_type) if storage_class: