docs(storage): add samples

storage/cloud-client/encryption_test.py

@@ -24,11 +24,12 @@
 
 import storage_download_encrypted_file
 import storage_generate_encryption_key
+import storage_object_csek_to_cmek
 import storage_rotate_encryption_key
 import storage_upload_encrypted_file
 
-
 BUCKET = os.environ["CLOUD_STORAGE_BUCKET"]
+KMS_KEY = os.environ["CLOUD_KMS_KEY"]
 
 TEST_ENCRYPTION_KEY = "brtJUWneL92g5q0N2gyDSnlPSYAiIVZ/cWgjyZNeMy0="
 TEST_ENCRYPTION_KEY_DECODED = base64.b64decode(TEST_ENCRYPTION_KEY)
@@ -113,3 +114,12 @@ def test_rotate_encryption_key(test_blob):
 
         downloaded_content = dest_file.read().decode("utf-8")
         assert downloaded_content == test_blob_content
+
+
+def test_object_csek_to_cmek(test_blob):
+    test_blob_name, test_blob_content = test_blob
+    cmek_blob = storage_object_csek_to_cmek.object_csek_to_cmek(
+        BUCKET, test_blob_name, TEST_ENCRYPTION_KEY, KMS_KEY
+    )
+
+    assert cmek_blob.download_as_string(), test_blob_content

storage/cloud-client/iam_test.py

@@ -23,9 +23,9 @@
 import storage_add_bucket_iam_member
 import storage_remove_bucket_conditional_iam_binding
 import storage_remove_bucket_iam_member
+import storage_set_bucket_public_iam
 import storage_view_bucket_iam_members
 
-
 MEMBER = "group:[email protected]"
 ROLE = "roles/storage.legacyBucketReader"
 
@@ -112,3 +112,16 @@ def test_remove_bucket_conditional_iam_binding(bucket):
         (binding["role"] == ROLE and binding.get("condition") == condition)
         for binding in policy.bindings
     )
+
+
+def test_set_bucket_public_iam(bucket):
+    role = "roles/storage.objectViewer"
+    member = "allUsers"
+    storage_set_bucket_public_iam.set_bucket_public_iam(
+        bucket.name, role, member
+    )
+    policy = bucket.get_iam_policy(requested_policy_version=3)
+    assert any(
+        binding["role"] == role and member in binding["members"]
+        for binding in policy.bindings
+    )

storage/cloud-client/snippets_test.py

@@ -23,11 +23,16 @@
 import requests
 
 import storage_add_bucket_label
+import storage_bucket_delete_default_kms_key
+import storage_compose_file
 import storage_copy_file
+import storage_create_bucket_class_location
+import storage_define_bucket_website_configuration
 import storage_delete_file
 import storage_disable_bucket_lifecycle_management
 import storage_disable_versioning
 import storage_download_file
+import storage_download_public_file
 import storage_enable_bucket_lifecycle_management
 import storage_enable_versioning
 import storage_generate_signed_post_policy_v4
@@ -37,11 +42,13 @@
 import storage_get_bucket_labels
 import storage_get_bucket_metadata
 import storage_get_metadata
+import storage_get_service_account
 import storage_list_buckets
 import storage_list_files
 import storage_list_files_with_prefix
 import storage_make_public
 import storage_move_file
+import storage_object_get_kms_key
 import storage_remove_bucket_label
 import storage_set_bucket_default_kms_key
 import storage_set_metadata
@@ -100,6 +107,17 @@ def test_blob(test_bucket):
     yield blob
 
 
+@pytest.fixture
+def test_bucket_create():
+    """Yields a bucket object that is deleted after the test completes."""
+    bucket = None
+    while bucket is None or bucket.exists():
+        bucket_name = "storage-snippets-test-{}".format(uuid.uuid4())
+        bucket = storage.Client().bucket(bucket_name)
+    yield bucket
+    bucket.delete(force=True)
+
+
 def test_list_buckets(test_bucket, capsys):
     storage_list_buckets.list_buckets()
     out, _ = capsys.readouterr()
@@ -204,9 +222,7 @@ def test_generate_upload_signed_url_v4(test_bucket, capsys):
     )
 
     requests.put(
-        url,
-        data=content,
-        headers={"content-type": "application/octet-stream"},
+        url, data=content, headers={"content-type": "application/octet-stream"},
     )
 
     bucket = storage.Client().bucket(test_bucket.name)
@@ -217,9 +233,7 @@ def test_generate_upload_signed_url_v4(test_bucket, capsys):
 def test_generate_signed_policy_v4(test_bucket, capsys):
     blob_name = "storage_snippets_test_form"
     short_name = storage_generate_signed_post_policy_v4
-    form = short_name.generate_signed_post_policy_v4(
-        test_bucket.name, blob_name
-    )
+    form = short_name.generate_signed_post_policy_v4(test_bucket.name, blob_name)
     assert "name='key' value='{}'".format(blob_name) in form
     assert "name='x-goog-signature'" in form
     assert "name='x-goog-date'" in form
@@ -238,9 +252,7 @@ def test_rename_blob(test_blob):
     except google.cloud.exceptions.exceptions.NotFound:
         pass
 
-    storage_move_file.rename_blob(
-        bucket.name, test_blob.name, "test_rename_blob"
-    )
+    storage_move_file.rename_blob(bucket.name, test_blob.name, "test_rename_blob")
 
     assert bucket.get_blob("test_rename_blob") is not None
     assert bucket.get_blob(test_blob.name) is None
@@ -275,15 +287,98 @@ def test_versioning(test_bucket, capsys):
 
 
 def test_bucket_lifecycle_management(test_bucket, capsys):
-    bucket = storage_enable_bucket_lifecycle_management.\
-        enable_bucket_lifecycle_management(test_bucket)
+    bucket = storage_enable_bucket_lifecycle_management.enable_bucket_lifecycle_management(
+        test_bucket
+    )
     out, _ = capsys.readouterr()
     assert "[]" in out
     assert "Lifecycle management is enable" in out
     assert len(list(bucket.lifecycle_rules)) > 0
 
-    bucket = storage_disable_bucket_lifecycle_management.\
-        disable_bucket_lifecycle_management(test_bucket)
+    bucket = storage_disable_bucket_lifecycle_management.disable_bucket_lifecycle_management(
+        test_bucket
+    )
     out, _ = capsys.readouterr()
     assert "[]" in out
     assert len(list(bucket.lifecycle_rules)) == 0
+
+
+def test_create_bucket_class_location(test_bucket_create):
+    bucket = storage_create_bucket_class_location.create_bucket_class_location(
+        test_bucket_create.name
+    )
+
+    assert bucket.location == "US"
+    assert bucket.storage_class == "COLDLINE"
+
+
+def test_bucket_delete_default_kms_key(test_bucket, capsys):
+    test_bucket.default_kms_key_name = KMS_KEY
+    test_bucket.patch()
+
+    assert test_bucket.default_kms_key_name == KMS_KEY
+
+    bucket = storage_bucket_delete_default_kms_key.bucket_delete_default_kms_key(
+        test_bucket.name
+    )
+
+    out, _ = capsys.readouterr()
+    assert bucket.default_kms_key_name is None
+    assert bucket.name in out
+
+
+def test_get_service_account(capsys):
+    storage_get_service_account.get_service_account()
+
+    out, _ = capsys.readouterr()
+
+    assert "@gs-project-accounts.iam.gserviceaccount.com" in out
+
+
+def test_download_public_file(test_blob):
+    storage_make_public.make_blob_public(test_blob.bucket.name, test_blob.name)
+    with tempfile.NamedTemporaryFile() as dest_file:
+        storage_download_public_file.download_public_file(
+            test_blob.bucket.name, test_blob.name, dest_file.name
+        )
+
+        assert dest_file.read() == b"Hello, is it me you're looking for?"
+
+
+def test_define_bucket_website_configuration(test_bucket):
+    bucket = storage_define_bucket_website_configuration.define_bucket_website_configuration(
+        test_bucket.name, "index.html", "404.html"
+    )
+
+    website_val = {"mainPageSuffix": "index.html", "notFoundPage": "404.html"}
+
+    assert bucket._properties["website"] == website_val
+
+
+def test_object_get_kms_key(test_bucket):
+    with tempfile.NamedTemporaryFile() as source_file:
+        storage_upload_with_kms_key.upload_blob_with_kms(
+            test_bucket.name, source_file.name, "test_upload_blob_encrypted", KMS_KEY
+        )
+    kms_key = storage_object_get_kms_key.object_get_kms_key(
+        test_bucket.name, "test_upload_blob_encrypted"
+    )
+
+    assert kms_key.startswith(KMS_KEY)
+
+
+def test_storage_compose_file(test_bucket):
+    source_files = ["test_upload_blob_1", "test_upload_blob_2"]
+    blob_list = []
+    for source in source_files:
+        blob = test_bucket.blob(source)
+        blob.upload_from_string(source)
+        blob_list.append(blob)
+
+    with tempfile.NamedTemporaryFile() as dest_file:
+        destination = storage_compose_file.compose_file(
+            test_bucket.name, blob_list, dest_file.name
+        )
+        composed = destination.download_as_string()
+
+        assert composed.decode("utf-8") == source_files[0] + source_files[1]

storage/cloud-client/storage_bucket_delete_default_kms_key.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+# Copyright 2020 Google LLC. All Rights Reserved.
+#
+# 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.
+
+import sys
+
+# [START storage_bucket_delete_default_kms_key]
+from google.cloud import storage
+
+
+def bucket_delete_default_kms_key(bucket_name):
+    """Delete a default KMS key of bucket"""
+    # bucket_name = "your-bucket-name"
+
+    storage_client = storage.Client()
+
+    bucket = storage_client.get_bucket(bucket_name)
+    bucket.default_kms_key_name = None
+    bucket.patch()
+
+    print("Default KMS key was removed from {}".format(bucket.name))
+    return bucket
+
+
+# [END storage_bucket_delete_default_kms_key]
+
+if __name__ == "__main__":
+    bucket_delete_default_kms_key(bucket_name=sys.argv[1])

storage/cloud-client/storage_compose_file.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+# Copyright 2020 Google LLC. All Rights Reserved.
+#
+# 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.
+
+import sys
+
+# [START storage_compose_file]
+from google.cloud import storage
+
+
+def compose_file(bucket_name, sources, destination_blob_name):
+    """Concatenate source blobs into destination blob."""
+    # bucket_name = "your-bucket-name"
+    # sources = [blob_1, blob_2]
+    # destination_blob_name = "destination-object-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    destination = bucket.blob(destination_blob_name)
+    destination.content_type = "text/plain"
+    destination.compose(sources)
+
+    print(
+        "Composed new object {} in the bucket {}".format(
+            destination_blob_name, bucket.name
+        )
+    )
+    return destination
+
+
+# [END storage_compose_file]
+
+if __name__ == "__main__":
+    compose_file(
+        bucket_name=sys.argv[1], sources=sys.argv[2], destination_blob_name=sys.argv[3],
+    )

storage/cloud-client/storage_create_bucket_class_location.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+
+# Copyright 2020 Google LLC. All Rights Reserved.
+#
+# 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.
+
+import sys
+
+# [START storage_create_bucket_class_location]
+from google.cloud import storage
+
+
+def create_bucket_class_location(bucket_name):
+    """Create a new bucket in specific location with storage class"""
+    # bucket_name = "your-new-bucket-name"
+
+    storage_client = storage.Client()
+
+    bucket = storage_client.bucket(bucket_name)
+    bucket.storage_class = "COLDLINE"
+    new_bucket = storage_client.create_bucket(bucket, location="us")
+
+    print(
+        "Created bucket {} in {} with storage class {}".format(
+            new_bucket.name, new_bucket.location, new_bucket.storage_class
+        )
+    )
+    return new_bucket
+
+
+# [END storage_create_bucket_class_location]
+
+if __name__ == "__main__":
+    create_bucket_class_location(bucket_name=sys.argv[1])