From e2eab99ef949bb3be9a672f9c4368e7fa77c85c4 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Fri, 4 Sep 2026 14:26:12 +0000 Subject: [PATCH 1/8] feat(storage): support storage_class in AsyncAppendableObjectWriter Add support for specifying storage_class when writing objects via AsyncAppendableObjectWriter. - Add storage_class parameter to AsyncAppendableObjectWriter.__init__ and propagate it to _AsyncWriteObjectStream on open(). - Add storage_class parameter to _AsyncWriteObjectStream.__init__ and pass it to _storage_v2.Object creation. - Add unit tests covering initialization and open stream requests. --- .../asyncio/async_appendable_object_writer.py | 7 +++ .../asyncio/async_write_object_stream.py | 9 +++- .../test_async_appendable_object_writer.py | 40 +++++++++++++++++ .../asyncio/test_async_write_object_stream.py | 44 +++++++++++++++++++ 4 files changed, 99 insertions(+), 1 deletion(-) diff --git a/packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py b/packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py index 7ef8ef099e6a..46155925b165 100644 --- a/packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py +++ b/packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py @@ -111,6 +111,7 @@ def __init__( generation: Optional[int] = None, write_handle: Optional[_storage_v2.BidiWriteHandle] = None, writer_options: Optional[dict] = None, + storage_class: Optional[str] = None, ): """ Class for appending data to a GCS Appendable Object. @@ -179,6 +180,10 @@ def __init__( The number of bytes to append before "persisting" data in GCS servers. Default is `_DEFAULT_FLUSH_INTERVAL_BYTES`. Must be a multiple of `_MAX_CHUNK_SIZE_BYTES`. + :type storage_class: Optional[str] + :param storage_class: (Optional) Storage class of the object bytes. + If specified, it overrides the bucket's `storage_class`. If not, object storage class + will be the same as bucket's storage_class. """ _utils.raise_if_no_fast_crc32c() self.client = client @@ -186,6 +191,7 @@ def __init__( self.object_name = object_name self.write_handle = write_handle self.generation = generation + self.storage_class = storage_class self.write_obj_stream: Optional[_AsyncWriteObjectStream] = None self._is_stream_open: bool = False @@ -361,6 +367,7 @@ async def _do_open(): generation_number=self.generation, write_handle=self.write_handle, routing_token=self._routing_token, + storage_class=self.storage_class, ) if self._routing_token: diff --git a/packages/google-cloud-storage/google/cloud/storage/asyncio/async_write_object_stream.py b/packages/google-cloud-storage/google/cloud/storage/asyncio/async_write_object_stream.py index 48f786c7654d..2d78a8c20fd1 100644 --- a/packages/google-cloud-storage/google/cloud/storage/asyncio/async_write_object_stream.py +++ b/packages/google-cloud-storage/google/cloud/storage/asyncio/async_write_object_stream.py @@ -58,6 +58,9 @@ class _AsyncWriteObjectStream(_AsyncAbstractObjectStream): :type write_handle: _storage_v2.BidiWriteHandle :param write_handle: (Optional) An existing handle for writing the object. If provided, opening the bidi-gRPC connection will be faster. + + :type storage_class: Optional[str] + :param storage_class: (Optional) The storage class of the object. """ def __init__( @@ -69,6 +72,7 @@ def __init__( write_handle: Optional[_storage_v2.BidiWriteHandle] = None, routing_token: Optional[str] = None, blob: Optional[Blob] = None, + storage_class: Optional[str] = None, ) -> None: if client is None: raise ValueError("client must be provided") @@ -86,6 +90,7 @@ def __init__( self.write_handle: Optional[_storage_v2.BidiWriteHandle] = write_handle self.routing_token: Optional[str] = routing_token self.blob: Optional[Blob] = blob + self.storage_class: Optional[str] = storage_class self._full_bucket_name = f"projects/_/buckets/{self.bucket_name}" self.rpc = self.client._client._transport._wrapped_methods[ @@ -124,7 +129,9 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None: resource = _grpc_conversions.blob_to_proto(self.blob) else: resource = _storage_v2.Object( - name=self.object_name, bucket=self._full_bucket_name + name=self.object_name, + bucket=self._full_bucket_name, + storage_class=self.storage_class, ) self.first_bidi_write_req = _storage_v2.BidiWriteObjectRequest( write_object_spec=_storage_v2.WriteObjectSpec( diff --git a/packages/google-cloud-storage/tests/unit/asyncio/test_async_appendable_object_writer.py b/packages/google-cloud-storage/tests/unit/asyncio/test_async_appendable_object_writer.py index 3775343f8818..87703b5ce5cc 100644 --- a/packages/google-cloud-storage/tests/unit/asyncio/test_async_appendable_object_writer.py +++ b/packages/google-cloud-storage/tests/unit/asyncio/test_async_appendable_object_writer.py @@ -117,6 +117,7 @@ def mock_appendable_writer(): yield { "mock_client": mock_client, "mock_stream": mock_stream, + "mock_stream_cls": mock_stream_cls, } stream_patcher.stop() @@ -137,6 +138,15 @@ def test_init_defaults(self, mock_appendable_writer): assert writer.persisted_size is None assert writer.bytes_appended_since_last_flush == 0 assert writer.flush_interval == _DEFAULT_FLUSH_INTERVAL_BYTES + assert writer.storage_class is None + + @pytest.mark.parametrize("storage_class", ["STANDARD", "RAPID"]) + def test_init_with_storage_class(self, mock_appendable_writer, storage_class): + writer = self._make_one( + mock_appendable_writer["mock_client"], + storage_class=storage_class, + ) + assert writer.storage_class == storage_class def test_init_with_writer_options(self, mock_appendable_writer): writer = self._make_one( @@ -218,6 +228,36 @@ async def test_open_success(self, mock_appendable_writer): assert writer.generation == 456 assert writer.write_handle == b"new-h" mock_appendable_writer["mock_stream"].open.assert_awaited_once() + mock_stream_cls = mock_appendable_writer["mock_stream_cls"] + assert mock_stream_cls.call_args.kwargs["storage_class"] is None + + @pytest.mark.asyncio + @pytest.mark.parametrize("storage_class", ["STANDARD", "RAPID"]) + async def test_open_passes_storage_class( + self, mock_appendable_writer, storage_class + ): + writer = self._make_one( + mock_appendable_writer["mock_client"], + storage_class=storage_class, + ) + mock_appendable_writer["mock_stream"].generation_number = 456 + mock_appendable_writer["mock_stream"].write_handle = b"new-h" + mock_appendable_writer["mock_stream"].persisted_size = 0 + + await writer.open() + + assert writer._is_stream_open + mock_stream_cls = mock_appendable_writer["mock_stream_cls"] + mock_stream_cls.assert_called_once_with( + client=mock_appendable_writer["mock_client"].grpc_client, + bucket_name=BUCKET, + object_name=OBJECT, + blob=None, + generation_number=None, + write_handle=None, + routing_token=None, + storage_class=storage_class, + ) def test_on_open_error_redirection(self, mock_appendable_writer): """Verify redirect info is extracted from helper.""" diff --git a/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py b/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py index 2fcca43b0882..c558d60cab18 100644 --- a/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py +++ b/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py @@ -63,6 +63,14 @@ def test_init_basic(self, mock_client): ("x-goog-request-params", f"bucket={FULL_BUCKET_PATH}"), ) assert not stream.is_stream_open + assert stream.storage_class is None + + @pytest.mark.parametrize("storage_class", ["STANDARD", "RAPID"]) + def test_init_with_storage_class(self, mock_client, storage_class): + stream = _AsyncWriteObjectStream( + mock_client, BUCKET, OBJECT, storage_class=storage_class + ) + assert stream.storage_class == storage_class def test_init_raises_value_error(self, mock_client): with pytest.raises(ValueError, match="client must be provided"): @@ -94,10 +102,46 @@ async def test_open_new_object(self, mock_rpc_cls, mock_client): await stream.open() # Check if BidiRpc was initialized with WriteObjectSpec + call_args = mock_rpc_cls.call_args + initial_request = call_args.kwargs["initial_request"] + # In proto3, string fields default to "" rather than None + resource = initial_request.write_object_spec.resource + assert "storage_class" not in resource + assert resource.storage_class == "" + assert initial_request.write_object_spec.appendable + + assert stream.is_stream_open + assert stream.write_handle == WRITE_HANDLE + assert stream.generation_number == GENERATION + + @mock.patch("google.cloud.storage.asyncio.async_write_object_stream.AsyncBidiRpc") + @pytest.mark.asyncio + @pytest.mark.parametrize("storage_class", ["STANDARD", "RAPID"]) + async def test_open_new_object_with_storage_class( + self, mock_rpc_cls, mock_client, storage_class + ): + mock_rpc = mock_rpc_cls.return_value + mock_rpc.open = AsyncMock() + + mock_response = MagicMock() + mock_response.persisted_size = 0 + mock_response.resource.generation = GENERATION + mock_response.resource.size = 0 + mock_response.write_handle = WRITE_HANDLE + mock_rpc.recv = AsyncMock(return_value=mock_response) + + stream = _AsyncWriteObjectStream( + mock_client, BUCKET, OBJECT, storage_class=storage_class + ) + await stream.open() + call_args = mock_rpc_cls.call_args initial_request = call_args.kwargs["initial_request"] assert initial_request.write_object_spec is not None assert initial_request.write_object_spec.resource.name == OBJECT + assert ( + initial_request.write_object_spec.resource.storage_class == storage_class + ) assert initial_request.write_object_spec.appendable assert stream.is_stream_open From 9797d34b956b16d0721eae9863fb9a6d11efb099 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Tue, 8 Sep 2026 18:54:58 +0000 Subject: [PATCH 2/8] test(storage): support preprod endpoints and RCU system tests --- .../tests/system/conftest.py | 15 +++++- .../tests/system/test_zonal.py | 49 ++++++++++++++----- 2 files changed, 51 insertions(+), 13 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/conftest.py b/packages/google-cloud-storage/tests/system/conftest.py index ee664a75684f..97824b3a268e 100644 --- a/packages/google-cloud-storage/tests/system/conftest.py +++ b/packages/google-cloud-storage/tests/system/conftest.py @@ -36,6 +36,10 @@ from google.cloud.storage.retry import DEFAULT_RETRY # noqa: E402 from . import _helpers # noqa: E402 +from google.api_core.client_options import ClientOptions + +PREPROD_JSON_HOST = "https://storage-preprod-test-unified.googleusercontent.com" +RUN_SYSTEM_TESTS_ON_PREPROD = os.getenv("RUN_SYSTEM_TESTS_ON_PREPROD") == "True" if trace_api is not None: _global_exporter = InMemorySpanExporter() @@ -110,12 +114,19 @@ def _kms_key_name(client, bucket, key_name): key_name, ) +@pytest.fixture(scope="session") +def run_system_tests_on_preprod(): + return os.getenv("RUN_SYSTEM_TESTS_ON_PREPROD") == "True" @pytest.fixture(scope="session") -def storage_client(): +def storage_client(run_system_tests_on_preprod): from google.cloud.storage import Client - client = Client() + client = Client( + client_options=ClientOptions(api_endpoint=PREPROD_JSON_HOST) + if run_system_tests_on_preprod + else None + ) with contextlib.closing(client): yield client diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index 8684e2747475..1791825da8b0 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -27,10 +27,16 @@ ObjectContexts, ObjectCustomContextPayload, ) +from google.api_core.client_options import ClientOptions + +PREPROD_GRPC_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" pytestmark = pytest.mark.skipif( - os.getenv("RUN_ZONAL_SYSTEM_TESTS") != "True", - reason="Zonal system tests need to be explicitly enabled. This helps scheduling tests in Kokoro and Cloud Build.", + not ( + (os.getenv("RUN_ZONAL_SYSTEM_TESTS") == "True") + ^ (os.getenv("RUN_RCU_SYSTEM_TESTS") == "True") + ), + reason="Any one of Zonal or RCU system tests need to be explicitly enabled. This helps scheduling tests in Kokoro and Cloud Build.", ) @@ -40,10 +46,20 @@ _CROSS_REGION_BUCKET = os.getenv("CROSS_REGION_BUCKET") _BYTES_TO_UPLOAD = b"dummy_bytes_to_write_read_and_delete_appendable_object" +RCU_SYSTEM_TESTS = os.getenv("RUN_RCU_SYSTEM_TESTS") == "True" +_RCU_BUCKET = os.getenv("RCU_BUCKET") +bucket_for_testing = ( + _ZONAL_BUCKET if os.getenv("RUN_ZONAL_SYSTEM_TESTS") else _RCU_BUCKET +) -async def create_async_grpc_client(attempt_direct_path=True): +async def create_async_grpc_client(attempt_direct_path=True, preprod=False): """Initializes async client and gets the current event loop.""" - return AsyncGrpcClient(attempt_direct_path=attempt_direct_path) + return AsyncGrpcClient( + attempt_direct_path=attempt_direct_path, + client_options=ClientOptions(api_endpoint=PREPROD_GRPC_ENDPOINT) + if preprod + else None, + ) @pytest.fixture(scope="session") @@ -107,7 +123,7 @@ def event_loop(): @pytest.fixture(scope="session") -def grpc_clients(event_loop): +def grpc_clients(event_loop, run_system_tests_on_preprod): # grpc clients has to be instantiated in the event loop, # otherwise grpc creates it's own event loop and attaches to the client. # Which will lead to deadlock because client running in one event loop and @@ -116,10 +132,14 @@ def grpc_clients(event_loop): # https://github.com/grpc/grpc/blob/61fe9b40a986792ab7d4eb8924027b671faf26ba/src/python/grpcio/grpc/_cython/_cygrpc/aio/common.pyx.pxi#L249 clients = { True: event_loop.run_until_complete( - create_async_grpc_client(attempt_direct_path=True) + create_async_grpc_client( + attempt_direct_path=True, preprod=run_system_tests_on_preprod + ) ), False: event_loop.run_until_complete( - create_async_grpc_client(attempt_direct_path=False) + create_async_grpc_client( + attempt_direct_path=False, preprod=run_system_tests_on_preprod + ) ), } return clients @@ -217,8 +237,13 @@ async def _run(): object_data = os.urandom(object_size) object_checksum = google_crc32c.value(object_data) grpc_client = grpc_clients[attempt_direct_path] - - writer = AsyncAppendableObjectWriter(grpc_client, _ZONAL_BUCKET, object_name) + print("bucket for testing", bucket_for_testing) + writer = AsyncAppendableObjectWriter( + grpc_client, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, + ) await writer.open() await writer.append(object_data) object_metadata = await writer.close(finalize_on_close=True) @@ -227,7 +252,7 @@ async def _run(): buffer = BytesIO() async with AsyncMultiRangeDownloader( - grpc_client, _ZONAL_BUCKET, object_name + grpc_client, bucket_for_testing, object_name ) as mrd: # (0, 0) means read the whole object await mrd.download_ranges([(0, 0, buffer)]) @@ -236,7 +261,9 @@ async def _run(): assert buffer.getvalue() == object_data # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer gc.collect() From 41a8d06186c47ada126f7eab71fef9da54ab8838 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Tue, 8 Sep 2026 19:19:11 +0000 Subject: [PATCH 3/8] test(storage): update zonal system tests for RCU and storage_class --- .../tests/system/test_zonal.py | 200 ++++++++++++------ 1 file changed, 137 insertions(+), 63 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index 1791825da8b0..dc524599310f 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -66,7 +66,7 @@ async def create_async_grpc_client(attempt_direct_path=True, preprod=False): def zonal_kms_key(storage_client, kms_client): """Provisions a KMS key in the same location as of the zonal bucket.""" # Get the zonal bucket and extract its location - bucket = storage_client.get_bucket(_ZONAL_BUCKET) + bucket = storage_client.get_bucket(bucket_for_testing) location = bucket.location.lower() project = storage_client.project @@ -160,7 +160,7 @@ def _get_equal_dist(a: int, b: int) -> tuple[int, int]: step = (b - a) // 3 return a + step, a + 2 * step - +@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='X regions reads/writes for RCU not supported in SDK yet') @pytest.mark.parametrize( "object_size", [ @@ -237,7 +237,6 @@ async def _run(): object_data = os.urandom(object_size) object_checksum = google_crc32c.value(object_data) grpc_client = grpc_clients[attempt_direct_path] - print("bucket for testing", bucket_for_testing) writer = AsyncAppendableObjectWriter( grpc_client, bucket_for_testing, @@ -287,7 +286,12 @@ async def _run(): object_data = os.urandom(object_size) object_checksum = google_crc32c.value(object_data) - writer = AsyncAppendableObjectWriter(grpc_client, _ZONAL_BUCKET, object_name) + writer = AsyncAppendableObjectWriter( + grpc_client, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, + ) await writer.open() mark1, mark2 = _get_equal_dist(0, object_size) await writer.append(object_data[0:mark1]) @@ -297,7 +301,7 @@ async def _run(): assert object_metadata.size == object_size assert int(object_metadata.checksums.crc32c) == object_checksum - mrd = AsyncMultiRangeDownloader(grpc_client, _ZONAL_BUCKET, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) buffer = BytesIO() await mrd.open() # (0, 0) means read the whole object @@ -307,7 +311,7 @@ async def _run(): assert mrd.persisted_size == object_size # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer del mrd gc.collect() @@ -340,8 +344,9 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client, - _ZONAL_BUCKET, + bucket_for_testing, object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, writer_options={"FLUSH_INTERVAL_BYTES": flush_interval}, ) await writer.open() @@ -353,7 +358,7 @@ async def _run(): assert object_metadata.size == object_size assert int(object_metadata.checksums.crc32c) == object_checksum - mrd = AsyncMultiRangeDownloader(grpc_client, _ZONAL_BUCKET, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) buffer = BytesIO() await mrd.open() # (0, 0) means read the whole object @@ -363,14 +368,14 @@ async def _run(): assert mrd.persisted_size == object_size # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer del mrd gc.collect() event_loop.run_until_complete(_run()) - +@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='Write from blob for RCU not supported in SDK yet') def test_write_from_blob( storage_client, blobs_to_delete, @@ -389,7 +394,7 @@ def test_write_from_blob( async def _run(): # 1. Create a Blob instance - blob = storage_client.bucket(_ZONAL_BUCKET).blob(object_name) + blob = storage_client.bucket(bucket_for_testing).blob(object_name) blob.content_type = content_type blob.metadata = metadata blob.cache_control = cache_control @@ -406,7 +411,7 @@ async def _run(): # 3. Verify the object metadata obj = await grpc_client.get_object( - bucket_name=_ZONAL_BUCKET, + bucket_name=bucket_for_testing, object_name=object_name, ) @@ -423,6 +428,7 @@ async def _run(): event_loop.run_until_complete(_run()) +@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='Write from blob with KMS key for RCU not supported in SDK yet') def test_write_from_blob_with_kms_key( storage_client, blobs_to_delete, @@ -437,7 +443,7 @@ def test_write_from_blob_with_kms_key( async def _run(): # Create a local Blob instance with the KMS key - blob = storage_client.bucket(_ZONAL_BUCKET).blob( + blob = storage_client.bucket(bucket_for_testing).blob( object_name, kms_key_name=zonal_kms_key ) @@ -450,7 +456,7 @@ async def _run(): # Verify the encryption metadata obj = await grpc_client.get_object( - bucket_name=_ZONAL_BUCKET, + bucket_name=bucket_for_testing, object_name=object_name, ) @@ -463,12 +469,13 @@ async def _run(): event_loop.run_until_complete(_run()) +@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='Write blob with contexts for RCU not supported in SDK yet') @pytest.mark.asyncio async def test_write_blob_with_contexts(storage_client, blobs_to_delete): async_client = await create_async_grpc_client() blob_name = f"ObjectContextsGrpc-{uuid.uuid4().hex}" - bucket = storage_client.bucket(_ZONAL_BUCKET) + bucket = storage_client.bucket(bucket_for_testing) blob = bucket.blob(blob_name) blob.contexts = ObjectContexts( blob, custom={"foo": ObjectCustomContextPayload(value="bar")} @@ -480,17 +487,17 @@ async def test_write_blob_with_contexts(storage_client, blobs_to_delete): try: blobs = list( - storage_client.list_blobs(_ZONAL_BUCKET, filter_='contexts."foo"="bar"') + storage_client.list_blobs(bucket_for_testing, filter_='contexts."foo"="bar"') ) names = [b.name for b in blobs] assert blob_name in names # Assert contexts via gRPC GetObject - obj_proto = await async_client.get_object(_ZONAL_BUCKET, blob_name) + obj_proto = await async_client.get_object(bucket_for_testing, blob_name) assert "foo" in obj_proto.contexts.custom assert obj_proto.contexts.custom["foo"].value == "bar" finally: - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(blob_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(blob_name)) def test_read_unfinalized_appendable_object( @@ -500,12 +507,17 @@ def test_read_unfinalized_appendable_object( async def _run(): grpc_client = grpc_client_direct - writer = AsyncAppendableObjectWriter(grpc_client, _ZONAL_BUCKET, object_name) + writer = AsyncAppendableObjectWriter( + grpc_client, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, + ) await writer.open() await writer.append(_BYTES_TO_UPLOAD) await writer.flush() - mrd = AsyncMultiRangeDownloader(grpc_client, _ZONAL_BUCKET, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) buffer = BytesIO() await mrd.open() assert mrd.persisted_size == len(_BYTES_TO_UPLOAD) @@ -515,7 +527,7 @@ async def _run(): assert buffer.getvalue() == _BYTES_TO_UPLOAD # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer del mrd gc.collect() @@ -529,20 +541,23 @@ def test_mrd_open_with_read_handle(event_loop, grpc_client_direct): async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(_BYTES_TO_UPLOAD) await writer.close() - mrd = AsyncMultiRangeDownloader(grpc_client_direct, _ZONAL_BUCKET, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client_direct, bucket_for_testing, object_name) await mrd.open() read_handle = mrd.read_handle await mrd.close() # Open a new MRD using the `read_handle` obtained above new_mrd = AsyncMultiRangeDownloader( - grpc_client_direct, _ZONAL_BUCKET, object_name, read_handle=read_handle + grpc_client_direct, bucket_for_testing, object_name, read_handle=read_handle ) await new_mrd.open() # persisted_size not set when opened with read_handle @@ -562,19 +577,24 @@ def test_mrd_open_with_read_handle_over_cloud_path(event_loop, grpc_client): object_name = f"test_read_handl-{str(uuid.uuid4())[:4]}" async def _run(): - writer = AsyncAppendableObjectWriter(grpc_client, _ZONAL_BUCKET, object_name) + writer = AsyncAppendableObjectWriter( + grpc_client, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, + ) await writer.open() await writer.append(_BYTES_TO_UPLOAD) await writer.close() - mrd = AsyncMultiRangeDownloader(grpc_client, _ZONAL_BUCKET, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) await mrd.open() read_handle = mrd.read_handle await mrd.close() # Open a new MRD using the `read_handle` obtained above new_mrd = AsyncMultiRangeDownloader( - grpc_client, _ZONAL_BUCKET, object_name, read_handle=read_handle + grpc_client, bucket_for_testing, object_name, read_handle=read_handle ) await new_mrd.open() # persisted_size is set regardless of whether we use read_handle or not @@ -599,7 +619,10 @@ def test_wrd_open_with_write_handle( async def _run(): # 1. Create an object and get its write_handle writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() write_handle = writer.write_handle @@ -608,10 +631,11 @@ async def _run(): # 2. Open a new writer using the obtained `write_handle` and generation new_writer = AsyncAppendableObjectWriter( grpc_client_direct, - _ZONAL_BUCKET, + bucket_for_testing, object_name, write_handle=write_handle, generation=writer.generation, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await new_writer.open() # Verify that the new writer is open and has the same write_handle @@ -624,7 +648,7 @@ async def _run(): await new_writer.close() # 4. Verify the data was written correctly by reading it back - mrd = AsyncMultiRangeDownloader(grpc_client_direct, _ZONAL_BUCKET, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client_direct, bucket_for_testing, object_name) buffer = BytesIO() await mrd.open() await mrd.download_ranges([(0, 0, buffer)]) @@ -632,7 +656,7 @@ async def _run(): assert buffer.getvalue() == test_data # Clean up - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer del new_writer del mrd @@ -651,7 +675,7 @@ async def _run(): async def _read_and_verify(expected_content, generation=None): """Helper to read object content and verify against expected.""" mrd = AsyncMultiRangeDownloader( - grpc_client, _ZONAL_BUCKET, object_name, generation + grpc_client, bucket_for_testing, object_name, generation ) buffer = BytesIO() await mrd.open() @@ -664,7 +688,12 @@ async def _read_and_verify(expected_content, generation=None): return mrd # First write - writer = AsyncAppendableObjectWriter(grpc_client, _ZONAL_BUCKET, object_name) + writer = AsyncAppendableObjectWriter( + grpc_client, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, + ) await writer.open() await writer.append(_BYTES_TO_UPLOAD) await writer.flush() @@ -675,7 +704,11 @@ async def _read_and_verify(expected_content, generation=None): # Second write, using generation from the first write. writer_2 = AsyncAppendableObjectWriter( - grpc_client, _ZONAL_BUCKET, object_name, generation=generation + grpc_client, + bucket_for_testing, + object_name, + generation=generation, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer_2.open() await writer_2.append(_BYTES_TO_UPLOAD) @@ -685,7 +718,7 @@ async def _read_and_verify(expected_content, generation=None): mrd_2 = await _read_and_verify(_BYTES_TO_UPLOAD + _BYTES_TO_UPLOAD, generation) # Clean up - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer del writer_2 del mrd @@ -710,7 +743,11 @@ def test_open_with_generation_zero( async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client, _ZONAL_BUCKET, object_name, generation=0 + grpc_client, + bucket_for_testing, + object_name, + generation=0, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) # Empty object is created. @@ -722,7 +759,11 @@ async def _run(): with pytest.raises(FailedPrecondition) as exc_info: writer_fail = AsyncAppendableObjectWriter( - grpc_client, _ZONAL_BUCKET, object_name, generation=0 + grpc_client, + bucket_for_testing, + object_name, + generation=0, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer_fail.open() assert exc_info.value.code == 400 @@ -731,7 +772,7 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -746,7 +787,11 @@ def test_open_existing_object_with_gen_None_overrides_existing( async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client, _ZONAL_BUCKET, object_name, generation=0 + grpc_client, + bucket_for_testing, + object_name, + generation=0, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) # Empty object is created. @@ -758,7 +803,11 @@ async def _run(): assert not writer.is_stream_open new_writer = AsyncAppendableObjectWriter( - grpc_client, _ZONAL_BUCKET, object_name, generation=None + grpc_client, + bucket_for_testing, + object_name, + generation=None, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await new_writer.open() assert new_writer.generation != old_gen @@ -768,7 +817,7 @@ async def _run(): del new_writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -781,7 +830,11 @@ def test_delete_object_using_grpc_client(event_loop, grpc_client_direct): async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name, generation=0 + grpc_client_direct, + bucket_for_testing, + object_name, + generation=0, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) # Empty object is created. @@ -789,13 +842,13 @@ async def _run(): await writer.append(b"some_bytes") await writer.close() - await grpc_client_direct.delete_object(_ZONAL_BUCKET, object_name) + await grpc_client_direct.delete_object(bucket_for_testing, object_name) # trying to get raises raises 404. with pytest.raises(NotFound): # TODO: Remove this once GET_OBJECT is exposed in `AsyncGrpcClient` await grpc_client_direct._grpc_client.get_object( - bucket=f"projects/_/buckets/{_ZONAL_BUCKET}", object_=object_name + bucket=f"projects/_/buckets/{bucket_for_testing}", object_=object_name ) # cleanup del writer @@ -833,14 +886,17 @@ async def _run(): object_data = os.urandom(object_size) writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(object_data) await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, bucket_for_testing, object_name ) as mrd: tasks = [] ranges_to_fetch = [] @@ -876,7 +932,7 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -896,14 +952,17 @@ async def _run(): object_data = os.urandom(object_size) writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(object_data) await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, bucket_for_testing, object_name ) as mrd: tasks = [] num_chunks = 40 @@ -937,7 +996,7 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -957,14 +1016,17 @@ async def _run(): object_data = os.urandom(object_size) writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(object_data) await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, bucket_for_testing, object_name ) as mrd: valid_buffer = BytesIO() valid_task = asyncio.create_task( @@ -986,7 +1048,7 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -1018,14 +1080,17 @@ async def _run(): object_data = os.urandom(object_size) writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(object_data) await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, bucket_for_testing, object_name ) as mrd: buffer = BytesIO() await mrd.download_ranges( @@ -1036,7 +1101,7 @@ async def _run(): # cleanup del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -1052,7 +1117,10 @@ def test_mrd_checksum_unfinalized_appendable_skipped( async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(_BYTES_TO_UPLOAD) @@ -1060,7 +1128,7 @@ async def _run(): # Download the unfinalized appendable object with enable_checksum=True async with AsyncMultiRangeDownloader( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, bucket_for_testing, object_name ) as mrd: buffer = BytesIO() # Since it's unfinalized, it should skip the checksum check without raising @@ -1071,7 +1139,7 @@ async def _run(): await writer.close() del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) event_loop.run_until_complete(_run()) @@ -1088,7 +1156,10 @@ def test_finalize_with_correct_checksum( async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(object_data) @@ -1098,7 +1169,7 @@ async def _run(): assert int(object_metadata.checksums.crc32c) == object_checksum # clean up - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer gc.collect() @@ -1118,7 +1189,10 @@ def test_finalize_with_incorrect_checksum_fails( async def _run(): writer = AsyncAppendableObjectWriter( - grpc_client_direct, _ZONAL_BUCKET, object_name + grpc_client_direct, + bucket_for_testing, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) await writer.open() await writer.append(object_data) @@ -1134,7 +1208,7 @@ async def _run(): ) # clean up - blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name)) + blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) del writer gc.collect() From 8a6713056fe1fc9dbc34f2cf7a329af191558e18 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Wed, 9 Sep 2026 08:11:17 +0000 Subject: [PATCH 4/8] style(storage): fix import ordering of ClientOptions in system tests --- packages/google-cloud-storage/tests/system/conftest.py | 3 +-- packages/google-cloud-storage/tests/system/test_zonal.py | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/conftest.py b/packages/google-cloud-storage/tests/system/conftest.py index 97824b3a268e..b1a7c946d094 100644 --- a/packages/google-cloud-storage/tests/system/conftest.py +++ b/packages/google-cloud-storage/tests/system/conftest.py @@ -30,13 +30,12 @@ import pytest # noqa: E402 from google.api_core import exceptions # noqa: E402 - +from google.api_core.client_options import ClientOptions # noqa: E402 from google.cloud import kms # noqa: E402 from google.cloud.storage._helpers import _base64_md5hash # noqa: E402 from google.cloud.storage.retry import DEFAULT_RETRY # noqa: E402 from . import _helpers # noqa: E402 -from google.api_core.client_options import ClientOptions PREPROD_JSON_HOST = "https://storage-preprod-test-unified.googleusercontent.com" RUN_SYSTEM_TESTS_ON_PREPROD = os.getenv("RUN_SYSTEM_TESTS_ON_PREPROD") == "True" diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index dc524599310f..5b73259d03dd 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -11,6 +11,7 @@ import google_crc32c import pytest from google.api_core import exceptions +from google.api_core.client_options import ClientOptions from google.api_core.exceptions import FailedPrecondition, NotFound, OutOfRange # current library imports @@ -27,7 +28,6 @@ ObjectContexts, ObjectCustomContextPayload, ) -from google.api_core.client_options import ClientOptions PREPROD_GRPC_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" From 2840244063d6689316aaadc0be91a97a02da78a4 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Wed, 9 Sep 2026 08:26:38 +0000 Subject: [PATCH 5/8] add doc string in test_zonal.py --- .../google-cloud-storage/tests/system/test_zonal.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index 5b73259d03dd..5cb00295e872 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -1,3 +1,14 @@ +"""System tests for Rapid Buckets (formerly Zonal Buckets) and RCU. + +Usage: + +RUN_RCU_SYSTEM_TESTS=True RCU_BUCKET= RUN_SYSTEM_TESTS_ON_PREPROD=True pytest packages/google-cloud-storage/tests/system/test_zonal.py + +and for Rapid Bucket (formerly Zonal Buckets): + +RUN_ZONAL_SYSTEM_TESTS=True ZONAL_BUCKET=<> CROSS_REGION_BUCKET=<> pytest packages/google-cloud-storage/tests/system/test_zonal.py +""" + # py standard imports import asyncio import datetime From 850d7ef2ec07335c27733fdabc67ac06e28be3cf Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Thu, 10 Sep 2026 09:29:08 +0000 Subject: [PATCH 6/8] test(storage): simplify bucket_for_testing assignment in zonal system tests --- .../tests/system/test_zonal.py | 78 +++++++++++++------ 1 file changed, 56 insertions(+), 22 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index 5cb00295e872..79a47b57f68c 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -41,7 +41,7 @@ ) PREPROD_GRPC_ENDPOINT = "storage-preprod-test-grpc.googleusercontent.com:443" - +# Run system test for either Rapid (formerly zonal) or RCU. But not both => XOR pytestmark = pytest.mark.skipif( not ( (os.getenv("RUN_ZONAL_SYSTEM_TESTS") == "True") @@ -59,9 +59,7 @@ RCU_SYSTEM_TESTS = os.getenv("RUN_RCU_SYSTEM_TESTS") == "True" _RCU_BUCKET = os.getenv("RCU_BUCKET") -bucket_for_testing = ( - _ZONAL_BUCKET if os.getenv("RUN_ZONAL_SYSTEM_TESTS") else _RCU_BUCKET -) +bucket_for_testing = _RCU_BUCKET if RCU_SYSTEM_TESTS else _ZONAL_BUCKET async def create_async_grpc_client(attempt_direct_path=True, preprod=False): """Initializes async client and gets the current event loop.""" @@ -322,7 +320,9 @@ async def _run(): assert mrd.persisted_size == object_size # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer del mrd gc.collect() @@ -379,7 +379,9 @@ async def _run(): assert mrd.persisted_size == object_size # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer del mrd gc.collect() @@ -498,7 +500,9 @@ async def test_write_blob_with_contexts(storage_client, blobs_to_delete): try: blobs = list( - storage_client.list_blobs(bucket_for_testing, filter_='contexts."foo"="bar"') + storage_client.list_blobs( + bucket_for_testing, filter_='contexts."foo"="bar"' + ) ) names = [b.name for b in blobs] assert blob_name in names @@ -508,7 +512,9 @@ async def test_write_blob_with_contexts(storage_client, blobs_to_delete): assert "foo" in obj_proto.contexts.custom assert obj_proto.contexts.custom["foo"].value == "bar" finally: - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(blob_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(blob_name) + ) def test_read_unfinalized_appendable_object( @@ -538,7 +544,9 @@ async def _run(): assert buffer.getvalue() == _BYTES_TO_UPLOAD # Clean up; use json client (i.e. `storage_client` fixture) to delete. - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer del mrd gc.collect() @@ -561,7 +569,9 @@ async def _run(): await writer.append(_BYTES_TO_UPLOAD) await writer.close() - mrd = AsyncMultiRangeDownloader(grpc_client_direct, bucket_for_testing, object_name) + mrd = AsyncMultiRangeDownloader( + grpc_client_direct, bucket_for_testing, object_name + ) await mrd.open() read_handle = mrd.read_handle await mrd.close() @@ -659,7 +669,9 @@ async def _run(): await new_writer.close() # 4. Verify the data was written correctly by reading it back - mrd = AsyncMultiRangeDownloader(grpc_client_direct, bucket_for_testing, object_name) + mrd = AsyncMultiRangeDownloader( + grpc_client_direct, bucket_for_testing, object_name + ) buffer = BytesIO() await mrd.open() await mrd.download_ranges([(0, 0, buffer)]) @@ -667,7 +679,9 @@ async def _run(): assert buffer.getvalue() == test_data # Clean up - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer del new_writer del mrd @@ -729,7 +743,9 @@ async def _read_and_verify(expected_content, generation=None): mrd_2 = await _read_and_verify(_BYTES_TO_UPLOAD + _BYTES_TO_UPLOAD, generation) # Clean up - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer del writer_2 del mrd @@ -783,7 +799,9 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -828,7 +846,9 @@ async def _run(): del new_writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -943,7 +963,9 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -1007,7 +1029,9 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -1059,7 +1083,9 @@ async def _run(): del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -1112,7 +1138,9 @@ async def _run(): # cleanup del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -1150,7 +1178,9 @@ async def _run(): await writer.close() del writer gc.collect() - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) event_loop.run_until_complete(_run()) @@ -1180,7 +1210,9 @@ async def _run(): assert int(object_metadata.checksums.crc32c) == object_checksum # clean up - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer gc.collect() @@ -1219,7 +1251,9 @@ async def _run(): ) # clean up - blobs_to_delete.append(storage_client.bucket(bucket_for_testing).blob(object_name)) + blobs_to_delete.append( + storage_client.bucket(bucket_for_testing).blob(object_name) + ) del writer gc.collect() From 934eddf9e787558344a7baf38dd9b150526c6bb9 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Thu, 10 Sep 2026 09:30:59 +0000 Subject: [PATCH 7/8] rename bucket_for_testing to _BUCKET_UNDER_TEST --- .../tests/system/test_zonal.py | 128 +++++++++--------- 1 file changed, 64 insertions(+), 64 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index 79a47b57f68c..03a6803cb294 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -59,7 +59,7 @@ RCU_SYSTEM_TESTS = os.getenv("RUN_RCU_SYSTEM_TESTS") == "True" _RCU_BUCKET = os.getenv("RCU_BUCKET") -bucket_for_testing = _RCU_BUCKET if RCU_SYSTEM_TESTS else _ZONAL_BUCKET +_BUCKET_UNDER_TEST = _RCU_BUCKET if RCU_SYSTEM_TESTS else _ZONAL_BUCKET async def create_async_grpc_client(attempt_direct_path=True, preprod=False): """Initializes async client and gets the current event loop.""" @@ -75,7 +75,7 @@ async def create_async_grpc_client(attempt_direct_path=True, preprod=False): def zonal_kms_key(storage_client, kms_client): """Provisions a KMS key in the same location as of the zonal bucket.""" # Get the zonal bucket and extract its location - bucket = storage_client.get_bucket(bucket_for_testing) + bucket = storage_client.get_bucket(_BUCKET_UNDER_TEST) location = bucket.location.lower() project = storage_client.project @@ -248,7 +248,7 @@ async def _run(): grpc_client = grpc_clients[attempt_direct_path] writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -260,7 +260,7 @@ async def _run(): buffer = BytesIO() async with AsyncMultiRangeDownloader( - grpc_client, bucket_for_testing, object_name + grpc_client, _BUCKET_UNDER_TEST, object_name ) as mrd: # (0, 0) means read the whole object await mrd.download_ranges([(0, 0, buffer)]) @@ -270,7 +270,7 @@ async def _run(): # Clean up; use json client (i.e. `storage_client` fixture) to delete. blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer gc.collect() @@ -297,7 +297,7 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -310,7 +310,7 @@ async def _run(): assert object_metadata.size == object_size assert int(object_metadata.checksums.crc32c) == object_checksum - mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, _BUCKET_UNDER_TEST, object_name) buffer = BytesIO() await mrd.open() # (0, 0) means read the whole object @@ -321,7 +321,7 @@ async def _run(): # Clean up; use json client (i.e. `storage_client` fixture) to delete. blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer del mrd @@ -355,7 +355,7 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, writer_options={"FLUSH_INTERVAL_BYTES": flush_interval}, @@ -369,7 +369,7 @@ async def _run(): assert object_metadata.size == object_size assert int(object_metadata.checksums.crc32c) == object_checksum - mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, _BUCKET_UNDER_TEST, object_name) buffer = BytesIO() await mrd.open() # (0, 0) means read the whole object @@ -380,7 +380,7 @@ async def _run(): # Clean up; use json client (i.e. `storage_client` fixture) to delete. blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer del mrd @@ -407,7 +407,7 @@ def test_write_from_blob( async def _run(): # 1. Create a Blob instance - blob = storage_client.bucket(bucket_for_testing).blob(object_name) + blob = storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) blob.content_type = content_type blob.metadata = metadata blob.cache_control = cache_control @@ -424,7 +424,7 @@ async def _run(): # 3. Verify the object metadata obj = await grpc_client.get_object( - bucket_name=bucket_for_testing, + bucket_name=_BUCKET_UNDER_TEST, object_name=object_name, ) @@ -456,7 +456,7 @@ def test_write_from_blob_with_kms_key( async def _run(): # Create a local Blob instance with the KMS key - blob = storage_client.bucket(bucket_for_testing).blob( + blob = storage_client.bucket(_BUCKET_UNDER_TEST).blob( object_name, kms_key_name=zonal_kms_key ) @@ -469,7 +469,7 @@ async def _run(): # Verify the encryption metadata obj = await grpc_client.get_object( - bucket_name=bucket_for_testing, + bucket_name=_BUCKET_UNDER_TEST, object_name=object_name, ) @@ -488,7 +488,7 @@ async def test_write_blob_with_contexts(storage_client, blobs_to_delete): async_client = await create_async_grpc_client() blob_name = f"ObjectContextsGrpc-{uuid.uuid4().hex}" - bucket = storage_client.bucket(bucket_for_testing) + bucket = storage_client.bucket(_BUCKET_UNDER_TEST) blob = bucket.blob(blob_name) blob.contexts = ObjectContexts( blob, custom={"foo": ObjectCustomContextPayload(value="bar")} @@ -501,19 +501,19 @@ async def test_write_blob_with_contexts(storage_client, blobs_to_delete): try: blobs = list( storage_client.list_blobs( - bucket_for_testing, filter_='contexts."foo"="bar"' + _BUCKET_UNDER_TEST, filter_='contexts."foo"="bar"' ) ) names = [b.name for b in blobs] assert blob_name in names # Assert contexts via gRPC GetObject - obj_proto = await async_client.get_object(bucket_for_testing, blob_name) + obj_proto = await async_client.get_object(_BUCKET_UNDER_TEST, blob_name) assert "foo" in obj_proto.contexts.custom assert obj_proto.contexts.custom["foo"].value == "bar" finally: blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(blob_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(blob_name) ) @@ -526,7 +526,7 @@ async def _run(): grpc_client = grpc_client_direct writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -534,7 +534,7 @@ async def _run(): await writer.append(_BYTES_TO_UPLOAD) await writer.flush() - mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, _BUCKET_UNDER_TEST, object_name) buffer = BytesIO() await mrd.open() assert mrd.persisted_size == len(_BYTES_TO_UPLOAD) @@ -545,7 +545,7 @@ async def _run(): # Clean up; use json client (i.e. `storage_client` fixture) to delete. blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer del mrd @@ -561,7 +561,7 @@ def test_mrd_open_with_read_handle(event_loop, grpc_client_direct): async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -570,7 +570,7 @@ async def _run(): await writer.close() mrd = AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) await mrd.open() read_handle = mrd.read_handle @@ -578,7 +578,7 @@ async def _run(): # Open a new MRD using the `read_handle` obtained above new_mrd = AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name, read_handle=read_handle + grpc_client_direct, _BUCKET_UNDER_TEST, object_name, read_handle=read_handle ) await new_mrd.open() # persisted_size not set when opened with read_handle @@ -600,7 +600,7 @@ def test_mrd_open_with_read_handle_over_cloud_path(event_loop, grpc_client): async def _run(): writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -608,14 +608,14 @@ async def _run(): await writer.append(_BYTES_TO_UPLOAD) await writer.close() - mrd = AsyncMultiRangeDownloader(grpc_client, bucket_for_testing, object_name) + mrd = AsyncMultiRangeDownloader(grpc_client, _BUCKET_UNDER_TEST, object_name) await mrd.open() read_handle = mrd.read_handle await mrd.close() # Open a new MRD using the `read_handle` obtained above new_mrd = AsyncMultiRangeDownloader( - grpc_client, bucket_for_testing, object_name, read_handle=read_handle + grpc_client, _BUCKET_UNDER_TEST, object_name, read_handle=read_handle ) await new_mrd.open() # persisted_size is set regardless of whether we use read_handle or not @@ -641,7 +641,7 @@ async def _run(): # 1. Create an object and get its write_handle writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -652,7 +652,7 @@ async def _run(): # 2. Open a new writer using the obtained `write_handle` and generation new_writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, write_handle=write_handle, generation=writer.generation, @@ -670,7 +670,7 @@ async def _run(): # 4. Verify the data was written correctly by reading it back mrd = AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) buffer = BytesIO() await mrd.open() @@ -680,7 +680,7 @@ async def _run(): # Clean up blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer del new_writer @@ -700,7 +700,7 @@ async def _run(): async def _read_and_verify(expected_content, generation=None): """Helper to read object content and verify against expected.""" mrd = AsyncMultiRangeDownloader( - grpc_client, bucket_for_testing, object_name, generation + grpc_client, _BUCKET_UNDER_TEST, object_name, generation ) buffer = BytesIO() await mrd.open() @@ -715,7 +715,7 @@ async def _read_and_verify(expected_content, generation=None): # First write writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -730,7 +730,7 @@ async def _read_and_verify(expected_content, generation=None): # Second write, using generation from the first write. writer_2 = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, generation=generation, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, @@ -744,7 +744,7 @@ async def _read_and_verify(expected_content, generation=None): # Clean up blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer del writer_2 @@ -771,7 +771,7 @@ def test_open_with_generation_zero( async def _run(): writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, generation=0, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, @@ -787,7 +787,7 @@ async def _run(): with pytest.raises(FailedPrecondition) as exc_info: writer_fail = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, generation=0, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, @@ -800,7 +800,7 @@ async def _run(): gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -817,7 +817,7 @@ def test_open_existing_object_with_gen_None_overrides_existing( async def _run(): writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, generation=0, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, @@ -833,7 +833,7 @@ async def _run(): new_writer = AsyncAppendableObjectWriter( grpc_client, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, generation=None, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, @@ -847,7 +847,7 @@ async def _run(): gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -862,7 +862,7 @@ def test_delete_object_using_grpc_client(event_loop, grpc_client_direct): async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, generation=0, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, @@ -873,13 +873,13 @@ async def _run(): await writer.append(b"some_bytes") await writer.close() - await grpc_client_direct.delete_object(bucket_for_testing, object_name) + await grpc_client_direct.delete_object(_BUCKET_UNDER_TEST, object_name) # trying to get raises raises 404. with pytest.raises(NotFound): # TODO: Remove this once GET_OBJECT is exposed in `AsyncGrpcClient` await grpc_client_direct._grpc_client.get_object( - bucket=f"projects/_/buckets/{bucket_for_testing}", object_=object_name + bucket=f"projects/_/buckets/{_BUCKET_UNDER_TEST}", object_=object_name ) # cleanup del writer @@ -918,7 +918,7 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -927,7 +927,7 @@ async def _run(): await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) as mrd: tasks = [] ranges_to_fetch = [] @@ -964,7 +964,7 @@ async def _run(): del writer gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -986,7 +986,7 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -995,7 +995,7 @@ async def _run(): await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) as mrd: tasks = [] num_chunks = 40 @@ -1030,7 +1030,7 @@ async def _run(): del writer gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -1052,7 +1052,7 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -1061,7 +1061,7 @@ async def _run(): await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) as mrd: valid_buffer = BytesIO() valid_task = asyncio.create_task( @@ -1084,7 +1084,7 @@ async def _run(): del writer gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -1118,7 +1118,7 @@ async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -1127,7 +1127,7 @@ async def _run(): await writer.close(finalize_on_close=True) async with AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) as mrd: buffer = BytesIO() await mrd.download_ranges( @@ -1139,7 +1139,7 @@ async def _run(): del writer gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -1157,7 +1157,7 @@ def test_mrd_checksum_unfinalized_appendable_skipped( async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -1167,7 +1167,7 @@ async def _run(): # Download the unfinalized appendable object with enable_checksum=True async with AsyncMultiRangeDownloader( - grpc_client_direct, bucket_for_testing, object_name + grpc_client_direct, _BUCKET_UNDER_TEST, object_name ) as mrd: buffer = BytesIO() # Since it's unfinalized, it should skip the checksum check without raising @@ -1179,7 +1179,7 @@ async def _run(): del writer gc.collect() blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) event_loop.run_until_complete(_run()) @@ -1198,7 +1198,7 @@ def test_finalize_with_correct_checksum( async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -1211,7 +1211,7 @@ async def _run(): # clean up blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer gc.collect() @@ -1233,7 +1233,7 @@ def test_finalize_with_incorrect_checksum_fails( async def _run(): writer = AsyncAppendableObjectWriter( grpc_client_direct, - bucket_for_testing, + _BUCKET_UNDER_TEST, object_name, storage_class="RAPID" if RCU_SYSTEM_TESTS else None, ) @@ -1252,7 +1252,7 @@ async def _run(): # clean up blobs_to_delete.append( - storage_client.bucket(bucket_for_testing).blob(object_name) + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) ) del writer gc.collect() From ee44da785190c24c9e14182b4d8cfc092e0cad56 Mon Sep 17 00:00:00 2001 From: Chandra Shekhar Sirimala Date: Thu, 10 Sep 2026 13:31:02 +0000 Subject: [PATCH 8/8] style(storage): fix import ordering and formatting in system and unit tests --- .../tests/system/conftest.py | 3 +++ .../tests/system/test_zonal.py | 20 +++++++++++++++---- .../asyncio/test_async_write_object_stream.py | 4 +--- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/conftest.py b/packages/google-cloud-storage/tests/system/conftest.py index b1a7c946d094..a74fd74f9652 100644 --- a/packages/google-cloud-storage/tests/system/conftest.py +++ b/packages/google-cloud-storage/tests/system/conftest.py @@ -31,6 +31,7 @@ import pytest # noqa: E402 from google.api_core import exceptions # noqa: E402 from google.api_core.client_options import ClientOptions # noqa: E402 + from google.cloud import kms # noqa: E402 from google.cloud.storage._helpers import _base64_md5hash # noqa: E402 from google.cloud.storage.retry import DEFAULT_RETRY # noqa: E402 @@ -113,10 +114,12 @@ def _kms_key_name(client, bucket, key_name): key_name, ) + @pytest.fixture(scope="session") def run_system_tests_on_preprod(): return os.getenv("RUN_SYSTEM_TESTS_ON_PREPROD") == "True" + @pytest.fixture(scope="session") def storage_client(run_system_tests_on_preprod): from google.cloud.storage import Client diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py index 03a6803cb294..d5c5360bd676 100644 --- a/packages/google-cloud-storage/tests/system/test_zonal.py +++ b/packages/google-cloud-storage/tests/system/test_zonal.py @@ -61,6 +61,7 @@ _RCU_BUCKET = os.getenv("RCU_BUCKET") _BUCKET_UNDER_TEST = _RCU_BUCKET if RCU_SYSTEM_TESTS else _ZONAL_BUCKET + async def create_async_grpc_client(attempt_direct_path=True, preprod=False): """Initializes async client and gets the current event loop.""" return AsyncGrpcClient( @@ -169,7 +170,10 @@ def _get_equal_dist(a: int, b: int) -> tuple[int, int]: step = (b - a) // 3 return a + step, a + 2 * step -@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='X regions reads/writes for RCU not supported in SDK yet') + +@pytest.mark.skipif( + RCU_SYSTEM_TESTS, reason="X regions reads/writes for RCU not supported in SDK yet" +) @pytest.mark.parametrize( "object_size", [ @@ -388,7 +392,10 @@ async def _run(): event_loop.run_until_complete(_run()) -@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='Write from blob for RCU not supported in SDK yet') + +@pytest.mark.skipif( + RCU_SYSTEM_TESTS, reason="Write from blob for RCU not supported in SDK yet" +) def test_write_from_blob( storage_client, blobs_to_delete, @@ -441,7 +448,10 @@ async def _run(): event_loop.run_until_complete(_run()) -@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='Write from blob with KMS key for RCU not supported in SDK yet') +@pytest.mark.skipif( + RCU_SYSTEM_TESTS, + reason="Write from blob with KMS key for RCU not supported in SDK yet", +) def test_write_from_blob_with_kms_key( storage_client, blobs_to_delete, @@ -482,7 +492,9 @@ async def _run(): event_loop.run_until_complete(_run()) -@pytest.mark.skipif(RCU_SYSTEM_TESTS, reason='Write blob with contexts for RCU not supported in SDK yet') +@pytest.mark.skipif( + RCU_SYSTEM_TESTS, reason="Write blob with contexts for RCU not supported in SDK yet" +) @pytest.mark.asyncio async def test_write_blob_with_contexts(storage_client, blobs_to_delete): async_client = await create_async_grpc_client() diff --git a/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py b/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py index c558d60cab18..e87b63a1759d 100644 --- a/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py +++ b/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py @@ -139,9 +139,7 @@ async def test_open_new_object_with_storage_class( initial_request = call_args.kwargs["initial_request"] assert initial_request.write_object_spec is not None assert initial_request.write_object_spec.resource.name == OBJECT - assert ( - initial_request.write_object_spec.resource.storage_class == storage_class - ) + assert initial_request.write_object_spec.resource.storage_class == storage_class assert initial_request.write_object_spec.appendable assert stream.is_stream_open