Merge branch 'configurable_flush_interval' of github.com:googleapis/python-storage into helpers

Author: chandra-siri

cloudbuild/zb-system-tests-cloudbuild.yaml

@@ -2,8 +2,25 @@ substitutions:
   _REGION: "us-central1"
   _ZONE: "us-central1-a"
   _SHORT_BUILD_ID: ${BUILD_ID:0:8}
+  _VM_NAME: "py-sdk-sys-test-${_SHORT_BUILD_ID}"
+
+
 
 steps:
+  # Step 0: Generate a persistent SSH key for this build run.
+  # This prevents gcloud from adding a new key to the OS Login profile on every ssh/scp command.
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "generate-ssh-key"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        mkdir -p /workspace/.ssh
+        # Generate the SSH key
+        ssh-keygen -t rsa -f /workspace/.ssh/google_compute_engine -N '' -C gcb
+        # Save the public key content to a file for the cleanup step
+        cat /workspace/.ssh/google_compute_engine.pub > /workspace/gcb_ssh_key.pub
+    waitFor: ["-"]
 
   # Step 1 Create a GCE VM to run the tests.
   # The VM is created in the same zone as the buckets to test rapid storage features.
@@ -15,7 +32,7 @@ steps:
       - "compute"
       - "instances"
       - "create"
-      - "gcsfs-test-vm-${_SHORT_BUILD_ID}"
+      - "${_VM_NAME}"
       - "--project=${PROJECT_ID}"
       - "--zone=${_ZONE}"
       - "--machine-type=e2-medium"
@@ -38,33 +55,46 @@ steps:
         set -e
         # Wait for the VM to be fully initialized and SSH to be ready.
         for i in {1..10}; do
-          if gcloud compute ssh gcsfs-test-vm-${_SHORT_BUILD_ID} --zone=${_ZONE} --internal-ip --command="echo VM is ready"; then
+          if gcloud compute ssh ${_VM_NAME} --zone=${_ZONE} --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine --command="echo VM is ready"; then
             break
           fi
           echo "Waiting for VM to become available... (attempt $i/10)"
           sleep 15
         done
         # copy the script to the VM
-        gcloud compute scp cloudbuild/run_zonal_tests.sh gcsfs-test-vm-${_SHORT_BUILD_ID}:~ --zone=${_ZONE} --internal-ip
+        gcloud compute scp cloudbuild/run_zonal_tests.sh ${_VM_NAME}:~ --zone=${_ZONE} --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine
 
         # Execute the script on the VM via SSH.
         # Capture the exit code to ensure cleanup happens before the build fails.
         set +e
-        gcloud compute ssh gcsfs-test-vm-${_SHORT_BUILD_ID} --zone=${_ZONE} --internal-ip --command="COMMIT_SHA=${COMMIT_SHA} _ZONAL_BUCKET=${_ZONAL_BUCKET} bash run_zonal_tests.sh"
+        gcloud compute ssh ${_VM_NAME} --zone=${_ZONE} --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine --command="COMMIT_SHA=${COMMIT_SHA} _ZONAL_BUCKET=${_ZONAL_BUCKET} bash run_zonal_tests.sh"
         EXIT_CODE=$?
         set -e
 
         echo "--- Deleting GCE VM ---"
-        gcloud compute instances delete "gcsfs-test-vm-${_SHORT_BUILD_ID}" --zone=${_ZONE} --quiet
+        gcloud compute instances delete "${_VM_NAME}" --zone=${_ZONE} --quiet
 
         # Exit with the original exit code from the test script.
         exit $$EXIT_CODE
     waitFor:
       - "create-vm"
+      - "generate-ssh-key"
+
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "cleanup-ssh-key"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        echo "--- Removing SSH key from OS Login profile to prevent accumulation ---"
+        gcloud compute os-login ssh-keys remove \
+          --key-file=/workspace/gcb_ssh_key.pub || true
+    waitFor:
+      - "run-tests-and-delete-vm"
 
 timeout: "3600s" # 60 minutes
 
 options:
   logging: CLOUD_LOGGING_ONLY
   pool:
-    name: "projects/${PROJECT_ID}/locations/us-central1/workerPools/cloud-build-worker-pool"
+    name: "projects/${PROJECT_ID}/locations/us-central1/workerPools/cloud-build-worker-pool"

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

@@ -24,7 +24,6 @@
 from io import BufferedReader
 from typing import Optional, Union
 
-from google.api_core import exceptions
 from google_crc32c import Checksum
 
 from ._utils import raise_if_no_fast_crc32c

tests/system/test_notification.py

@@ -69,6 +69,7 @@ def notification_topic(storage_client, publisher_client, topic_path, no_mtls):
     publisher_client.set_iam_policy(request={"resource": topic_path, "policy": policy})
 
 
+@pytest.mark.skip(reason="until b/470069573 is fixed")
 def test_notification_create_minimal(
     storage_client,
     buckets_to_delete,
@@ -94,6 +95,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,
@@ -128,6 +130,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,
@@ -156,6 +159,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,
@@ -184,6 +188,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/unit/asyncio/test_async_appendable_object_writer.py

@@ -16,6 +16,8 @@
 import pytest
 from unittest import mock
 
+from google_crc32c import Checksum
+
 from google.api_core import exceptions
 from google.cloud.storage._experimental.asyncio.async_appendable_object_writer import (
     AsyncAppendableObjectWriter,
@@ -32,7 +34,7 @@
 GENERATION = 123
 WRITE_HANDLE = b"test-write-handle"
 PERSISTED_SIZE = 456
-_ONE_MIB = 1024 * 1024
+EIGHT_MIB = 8 * 1024 * 1024
 
 
 @pytest.fixture
@@ -103,10 +105,10 @@ def test_init_with_writer_options(mock_write_object_stream, mock_client):
         mock_client,
         BUCKET,
         OBJECT,
-        writer_options={"FLUSH_INTERVAL_BYTES": 8 * _ONE_MIB},
+        writer_options={"FLUSH_INTERVAL_BYTES": EIGHT_MIB},
     )
 
-    assert writer.flush_interval == 8 * _ONE_MIB
+    assert writer.flush_interval == EIGHT_MIB
     assert writer.bytes_appended_since_last_flush == 0
 
     mock_write_object_stream.assert_called_once_with(
@@ -511,10 +513,15 @@ async def test_append_sends_data_in_chunks(mock_write_object_stream, mock_client
     # First chunk
     assert first_call[0][0].write_offset == 100
     assert len(first_call[0][0].checksummed_data.content) == _MAX_CHUNK_SIZE_BYTES
-
+    assert first_call[0][0].checksummed_data.crc32c == int.from_bytes(
+        Checksum(data[:_MAX_CHUNK_SIZE_BYTES]).digest(), byteorder="big"
+    )
     # Second chunk
     assert second_call[0][0].write_offset == 100 + _MAX_CHUNK_SIZE_BYTES
     assert len(second_call[0][0].checksummed_data.content) == 1
+    assert second_call[0][0].checksummed_data.crc32c == int.from_bytes(
+        Checksum(data[_MAX_CHUNK_SIZE_BYTES:]).digest(), byteorder="big"
+    )
 
     assert writer.offset == 100 + len(data)
     writer.simple_flush.assert_not_awaited()