feat(zb-experimental): implement open in writer (#1618)

feat(zb-experimental): implement open in writer
---------
Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: chandra-siri

google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py

@@ -126,7 +126,17 @@ async def state_lookup(self) -> int:
 
 asyncdefopen(self) ->None:
 """Opens the underlying bidi-gRPC stream."""
-raiseNotImplementedError("open is not implemented yet.")
+ifself._is_stream_open:
+raiseValueError("Underlying bidi-gRPC stream is already open")
+
+awaitself.write_obj_stream.open()
+self._is_stream_open=True
+ifself.generationisNone:
+self.generation=self.write_obj_stream.generation_number
+self.write_handle=self.write_obj_stream.write_handle
+
+# Update self.persisted_size
+_=awaitself.state_lookup()
 
 asyncdefappend(self, data: bytes):
 raiseNotImplementedError("append is not implemented yet.")

tests/unit/asyncio/test_async_appendable_object_writer.py

@@ -112,13 +112,62 @@ async def test_state_lookup(mock_write_object_stream, mock_client):
 
 
 @pytest.mark.asyncio
-asyncdeftest_unimplemented_methods_raise_error(mock_client):
-"""Test that all currently unimplemented methods raise NotImplementedError."""
+@mock.patch(
+"google.cloud.storage._experimental.asyncio.async_appendable_object_writer._AsyncWriteObjectStream"
+)
+asyncdeftest_open_appendable_object_writer(mock_write_object_stream, mock_client):
+"""Test the open method."""
+# Arrange
 writer=AsyncAppendableObjectWriter(mock_client, BUCKET, OBJECT)
+mock_stream=mock_write_object_stream.return_value
+mock_stream.open=mock.AsyncMock()
+mock_stream.send=mock.AsyncMock()
+mock_stream.recv=mock.AsyncMock()
 
-withpytest.raises(NotImplementedError):
+mock_state_response=mock.MagicMock()
+mock_state_response.persisted_size=1024
+mock_stream.recv.return_value=mock_state_response
+
+mock_stream.generation_number=GENERATION
+mock_stream.write_handle=WRITE_HANDLE
+
+# Act
+awaitwriter.open()
+
+# Assert
+mock_stream.open.assert_awaited_once()
+assertwriter._is_stream_open
+assertwriter.generation==GENERATION
+assertwriter.write_handle==WRITE_HANDLE
+
+expected_request=_storage_v2.BidiWriteObjectRequest(state_lookup=True)
+mock_stream.send.assert_awaited_once_with(expected_request)
+mock_stream.recv.assert_awaited_once()
+assertwriter.persisted_size==1024
+
+
+@pytest.mark.asyncio
+@mock.patch(
+"google.cloud.storage._experimental.asyncio.async_appendable_object_writer._AsyncWriteObjectStream"
+)
+asyncdeftest_open_when_already_open_raises_error(
+mock_write_object_stream, mock_client
+):
+"""Test that opening an already open writer raises a ValueError."""
+# Arrange
+writer=AsyncAppendableObjectWriter(mock_client, BUCKET, OBJECT)
+writer._is_stream_open=True# Manually set to open
+
+# Act & Assert
+withpytest.raises(ValueError, match="Underlying bidi-gRPC stream is already open"):
 awaitwriter.open()
 
+
+@pytest.mark.asyncio
+asyncdeftest_unimplemented_methods_raise_error(mock_client):
+"""Test that all currently unimplemented methods raise NotImplementedError."""
+writer=AsyncAppendableObjectWriter(mock_client, BUCKET, OBJECT)
+
 withpytest.raises(NotImplementedError):
 awaitwriter.append(b"data")
 

tests/unit/test_bucket.py

@@ -2350,7 +2350,7 @@ def test_move_blob_needs_url_encoding(self):
 timeout=30,
 retry=None,
 _target_object=new_blob,
- )
+ )
 
 deftest_move_blob_w_user_project(self):
 source_name="source"