chore: add system and unit tests

Author: nidhiii-27

packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py

@@ -0,0 +1,40 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from google.cloud import _storage_v2
+
+# Map Python Blob attributes to GCS V2 Object proto field names.
+_BLOB_ATTR_TO_PROTO_FIELD = {
+    "content_type": "content_type",
+    "metadata": "metadata",
+    "kms_key_name": "kms_key",
+}
+
+
+def blob_to_proto(blob):
+    """Converts a Blob instance to a GCS V2 Object proto message."""
+
+    resource_params = {
+        "name": blob.name,
+    }
+
+    if blob.bucket:
+        resource_params["bucket"] = f"projects/_/buckets/{blob.bucket.name}"
+
+    for attr_name, proto_field in _BLOB_ATTR_TO_PROTO_FIELD.items():
+        value = getattr(blob, attr_name, None)
+        if value is not None:
+            resource_params[proto_field] = value
+
+    return _storage_v2.Object(**resource_params)

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

@@ -212,6 +212,7 @@ def __init__(
         self._routing_token: Optional[str] = None
         self.object_resource: Optional[_storage_v2.Object] = None
         self._flush_count = 0
+        self.blob: Optional[Blob] = None
 
     @classmethod
     def from_blob(

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

@@ -19,13 +19,13 @@
 
 from google.cloud import _storage_v2
 from google.cloud.storage import Blob
+from google.cloud.storage import _grpc_conversions
 from google.cloud.storage.asyncio import _utils
 from google.cloud.storage.asyncio.async_abstract_object_stream import (
     _AsyncAbstractObjectStream,
 )
 from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
 
-_BLOB_SYNC_ATTRS = ("content_type", "metadata")
 
 class _AsyncWriteObjectStream(_AsyncAbstractObjectStream):
     """Class representing a gRPC bidi-stream for writing data from a GCS
@@ -121,21 +121,15 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
         # if `generation_number` == 0 new object will be created only if there
         # isn't any existing object.
         if self.generation_number is None or self.generation_number == 0:
-            resource_params = {
-            "name": self.object_name,
-            "bucket": self._full_bucket_name,
-            }
             if self.blob:
-                resource_params.update({
-                    attr: getattr(self.blob, attr)
-                    for attr in _BLOB_SYNC_ATTRS
-                    if getattr(self.blob, attr, None) is not None
-                })
+                resource = _grpc_conversions.blob_to_proto(self.blob)
+            else:
+                resource = _storage_v2.Object(
+                    name=self.object_name, bucket=self._full_bucket_name
+                )
             self.first_bidi_write_req = _storage_v2.BidiWriteObjectRequest(
                 write_object_spec=_storage_v2.WriteObjectSpec(
-                    resource=_storage_v2.Object(
-                        **resource_params
-                    ),
+                    resource=resource,
                     appendable=True,
                     if_generation_match=self.generation_number,
                 ),

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

@@ -18,6 +18,7 @@
 from google.cloud.storage.asyncio.async_appendable_object_writer import (
     AsyncAppendableObjectWriter,
 )
+from google.cloud import storage
 from google.cloud.storage import Blob
 from google.cloud.storage._experimental.asyncio.async_grpc_client import AsyncGrpcClient
 
@@ -32,7 +33,11 @@ async def storage_create_and_write_appendable_object(
 
     if grpc_client is None:
         grpc_client = AsyncGrpcClient()
-    blob = Blob.from_uri("gs://{}/{}".format(bucket_name, object_name))
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+    bucket_location = bucket.location.lower()
+    kms_key_name = f"projects/{storage_client.project}/locations/{bucket_location}/keyRings/gcs-test/cryptoKeys/gcs-test"
+    blob = Blob(bucket=bucket, name=object_name, kms_key_name=kms_key_name)
     blob.content_type = "text/plain"
     blob.metadata = {"environment": "dev"}
     writer = AsyncAppendableObjectWriter.from_blob(
@@ -53,6 +58,8 @@ async def storage_create_and_write_appendable_object(
     print(new_object)
     print(new_object.size)
     print(new_object.content_type)
+    print(new_object.metadata)
+    print(new_object.kms_key)
     print(
         f"Appended object {object_name} created of size {writer.persisted_size} bytes."
     )
@@ -72,4 +79,4 @@ async def storage_create_and_write_appendable_object(
             bucket_name=args.bucket_name,
             object_name=args.object_name,
         )
-    )
+    )

packages/google-cloud-storage/tests/system/test_zonal.py

@@ -28,7 +28,7 @@
 
 
 # TODO: replace this with a fixture once zonal bucket creation / deletion
-# is supported in grpc client or json client client.
+# is supported in grpc client or json client.
 _ZONAL_BUCKET = os.getenv("ZONAL_BUCKET")
 _CROSS_REGION_BUCKET = os.getenv("CROSS_REGION_BUCKET")
 _BYTES_TO_UPLOAD = b"dummy_bytes_to_write_read_and_delete_appendable_object"
@@ -285,6 +285,84 @@ async def _run():
     event_loop.run_until_complete(_run())
 
 
+def test_write_from_blob(
+    storage_client,
+    blobs_to_delete,
+    event_loop,
+    grpc_client,
+):
+    object_name = f"test_from_blob-{str(uuid.uuid4())[:4]}"
+    content_type = "text/plain"
+    metadata = {"environment": "system-test"}
+    test_data = b"system-test-data"
+
+    async def _run():
+        # 1. Create a Blob instance
+        blob = storage_client.bucket(_ZONAL_BUCKET).blob(object_name)
+        blob.content_type = content_type
+        blob.metadata = metadata
+
+        # 2. Use from_blob to create the writer
+        writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob)
+        await writer.open()
+        await writer.append(test_data)
+        await writer.close(finalize_on_close=True)
+
+        # 3. Verify the object metadata
+        obj = await grpc_client.get_object(
+            bucket_name=_ZONAL_BUCKET,
+            object_name=object_name,
+        )
+
+        assert obj.content_type == content_type
+        assert obj.metadata["environment"] == "system-test"
+
+        blobs_to_delete.append(blob)
+
+    event_loop.run_until_complete(_run())
+
+
+def test_write_from_blob_with_kms_key(
+    storage_client,
+    blobs_to_delete,
+    event_loop,
+    grpc_client,
+):
+    """Verifies AsyncAppendableObjectWriter.from_blob correctly applies KMS encryption."""
+
+    object_name = f"test_from_blob_kms-{str(uuid.uuid4())[:4]}"
+    test_data = b"kms-protected-data"
+    test_bucket = storage_client.bucket(_ZONAL_BUCKET)
+    bucket_location = test_bucket.location.lower()
+    # TODO: Use a fixture for a zonal kms key once we have fixture for zonal bucket
+    kms_key_name = f"projects/{storage_client.project}/locations/{bucket_location}/keyRings/gcs-test/cryptoKeys/gcs-test"
+
+    async def _run():
+        # Create a local Blob instance with the KMS key
+        blob = test_bucket.blob(object_name, kms_key_name=kms_key_name)
+
+        writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob)
+
+        await writer.open()
+        await writer.append(test_data)
+
+        await writer.close(finalize_on_close=True)
+
+        # Verify the encryption metadata
+        obj = await grpc_client.get_object(
+            bucket_name=_ZONAL_BUCKET,
+            object_name=object_name,
+        )
+
+        # Assert that the object was encrypted with the correct key
+        # GCS appends a version suffix, so we use startswith()
+        assert obj.kms_key_name.startswith(kms_key_name)
+
+        blobs_to_delete.append(blob)
+
+    event_loop.run_until_complete(_run())
+
+
 def test_read_unfinalized_appendable_object(
     storage_client, blobs_to_delete, event_loop, grpc_client_direct
 ):

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

@@ -19,6 +19,7 @@
 import pytest
 from google.api_core import exceptions
 from google.rpc import status_pb2
+from google.cloud.storage import Blob
 
 from google.cloud._storage_v2.types import storage as storage_type
 from google.cloud._storage_v2.types.storage import BidiWriteObjectRedirectedError
@@ -166,6 +167,24 @@ def test_init_raises_if_crc32c_missing(self, mock_appendable_writer):
             with pytest.raises(exceptions.FailedPrecondition):
                 self._make_one(mock_appendable_writer["mock_client"])
 
+    def test_from_blob(self, mock_appendable_writer):
+        mock_blob = mock.Mock(spec=Blob)
+        mock_blob.name = OBJECT
+        mock_blob.bucket.name = BUCKET
+
+        writer = AsyncAppendableObjectWriter.from_blob(
+            mock_appendable_writer["mock_client"],
+            mock_blob,
+            generation=GENERATION,
+            writer_options={"FLUSH_INTERVAL_BYTES": EIGHT_MIB},
+        )
+
+        assert writer.bucket_name == BUCKET
+        assert writer.object_name == OBJECT
+        assert writer.generation == GENERATION
+        assert writer.flush_interval == EIGHT_MIB
+        assert writer.blob == mock_blob
+
     # -------------------------------------------------------------------------
     # Stream Lifecycle Tests
     # -------------------------------------------------------------------------
@@ -176,9 +195,9 @@ async def test_state_lookup(self, mock_appendable_writer):
         writer._is_stream_open = True
         writer.write_obj_stream = mock_appendable_writer["mock_stream"]
 
-        mock_appendable_writer[
-            "mock_stream"
-        ].recv.return_value = storage_type.BidiWriteObjectResponse(persisted_size=100)
+        mock_appendable_writer["mock_stream"].recv.return_value = (
+            storage_type.BidiWriteObjectResponse(persisted_size=100)
+        )
 
         size = await writer.state_lookup()
 
@@ -389,9 +408,9 @@ async def test_flush_resets_counters(self, mock_appendable_writer):
         writer.write_obj_stream = mock_appendable_writer["mock_stream"]
         writer.bytes_appended_since_last_flush = 100
 
-        mock_appendable_writer[
-            "mock_stream"
-        ].recv.return_value = storage_type.BidiWriteObjectResponse(persisted_size=200)
+        mock_appendable_writer["mock_stream"].recv.return_value = (
+            storage_type.BidiWriteObjectResponse(persisted_size=200)
+        )
 
         await writer.flush()
 
@@ -432,9 +451,9 @@ async def test_finalize_lifecycle(self, mock_appendable_writer):
         writer.write_obj_stream = mock_appendable_writer["mock_stream"]
 
         resource = storage_type.Object(size=999)
-        mock_appendable_writer[
-            "mock_stream"
-        ].recv.return_value = storage_type.BidiWriteObjectResponse(resource=resource)
+        mock_appendable_writer["mock_stream"].recv.return_value = (
+            storage_type.BidiWriteObjectResponse(resource=resource)
+        )
 
         res = await writer.finalize()
 

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

@@ -19,6 +19,7 @@
 import pytest
 
 from google.cloud import _storage_v2
+from google.cloud.storage import Blob
 from google.cloud.storage.asyncio.async_write_object_stream import (
     _AsyncWriteObjectStream,
 )
@@ -151,6 +152,29 @@ async def test_open_metadata_merging(self, mock_rpc_cls, mock_client):
         assert f"bucket={FULL_BUCKET_PATH}" in params
         assert "extra=param" in params
 
+    @mock.patch("google.cloud.storage.asyncio.async_write_object_stream.AsyncBidiRpc")
+    @pytest.mark.asyncio
+    async def test_open_new_object_with_blob_sync_attrs(
+        self, mock_rpc_cls, mock_client
+    ):
+        mock_rpc = mock_rpc_cls.return_value
+        mock_rpc.open = AsyncMock()
+        mock_rpc.recv = AsyncMock(return_value=MagicMock(resource=None))
+
+        mock_blob = mock.Mock(spec=Blob)
+        mock_blob.content_type = "text/plain"
+        mock_blob.metadata = {"test-key": "test-value"}
+
+        stream = _AsyncWriteObjectStream(mock_client, BUCKET, OBJECT, blob=mock_blob)
+        await stream.open()
+
+        # Verify initial request contains synced attributes from blob
+        initial_request = mock_rpc_cls.call_args.kwargs["initial_request"]
+        resource = initial_request.write_object_spec.resource
+
+        assert resource.content_type == "text/plain"
+        assert resource.metadata == {"test-key": "test-value"}
+
     @pytest.mark.asyncio
     async def test_open_already_open_raises(self, mock_client):
         stream = _AsyncWriteObjectStream(mock_client, BUCKET, OBJECT)