feat(storage): Sync main and async branch (#1674)

The PR sync the main and async branch.

---------

Co-authored-by: Chandra Shekhar Sirimala <chandrasiri@google.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>
Co-authored-by: Chalmer Lowe <chalmerlowe@google.com>
Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
Co-authored-by: Anthonios Partheniou <partheniou@google.com>

Author: 7 people

tests/system/test_notification.py

@@ -75,6 +75,7 @@ def notification_topic(storage_client, publisher_client, topic_path, no_mtls):
         publisher_client.delete_topic(request={"topic": topic_path})
 
 
+@pytest.mark.skip(reason="until b/470069573 is fixed")
 def test_notification_create_minimal(
     storage_client,
     buckets_to_delete,
@@ -100,6 +101,7 @@ def test_notification_create_minimal(
         notification.delete()
 
 
+@pytest.mark.skip(reason="until b/470069573 is fixed")
 def test_notification_create_explicit(
     storage_client,
     buckets_to_delete,
@@ -134,6 +136,7 @@ def test_notification_create_explicit(
         notification.delete()
 
 
+@pytest.mark.skip(reason="until b/470069573 is fixed")
 def test_notification_create_w_user_project(
     storage_client,
     buckets_to_delete,
@@ -162,6 +165,7 @@ def test_notification_create_w_user_project(
         notification.delete()
 
 
+@pytest.mark.skip(reason="until b/470069573 is fixed")
 def test_notification_create_wo_topic_name(
     storage_client,
     buckets_to_delete,
@@ -190,6 +194,7 @@ def test_notification_create_wo_topic_name(
         notification.create()
 
 
+@pytest.mark.skip(reason="until b/470069573 is fixed")
 def test_bucket_get_notification(
     storage_client,
     buckets_to_delete,

tests/system/test_zonal.py

@@ -1,4 +1,5 @@
 # py standard imports
+import asyncio
 import os
 import uuid
 from io import BytesIO

tests/unit/asyncio/test_async_appendable_object_writer.py

@@ -107,6 +107,7 @@ def mock_appendable_writer():
     mock_stream.persisted_size = 0
     mock_stream.generation_number = GENERATION
     mock_stream.write_handle = WRITE_HANDLE
+    mock_stream.persisted_size = PERSISTED_SIZE
 
     yield {
         "mock_client": mock_client,