samples: add cloud client samples from python-docs-samples

samples/snippets/README.md

@@ -0,0 +1,10 @@
+
+For requester_pays_test.py, we need to use a different Storage bucket.
+
+The test looks for an environment variable `REQUESTER_PAYS_TEST_BUCKET`.
+
+Also, the service account for the test needs to have `Billing Project
+Manager` role in order to make changes on buckets with requester pays
+enabled.
+
+We added that role to the test service account.

samples/snippets/acl_test.py

@@ -0,0 +1,172 @@
+# Copyright 2016 Google, Inc.
+#
+# 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 os
+import uuid
+
+import backoff
+from google.cloud import storage
+from googleapiclient.errors import HttpError
+import pytest
+
+import storage_add_bucket_default_owner
+import storage_add_bucket_owner
+import storage_add_file_owner
+import storage_print_bucket_acl
+import storage_print_bucket_acl_for_user
+import storage_print_file_acl
+import storage_print_file_acl_for_user
+import storage_remove_bucket_default_owner
+import storage_remove_bucket_owner
+import storage_remove_file_owner
+
+# Typically we'd use a @example.com address, but GCS requires a real Google
+# account. Retrieve a service account email with storage admin permissions.
+TEST_EMAIL = (
+    "py38-storage-test"
+    "@python-docs-samples-tests.iam.gserviceaccount.com"
+)
+
+
+@pytest.fixture(scope="module")
+def test_bucket():
+    """Yields a bucket that is deleted after the test completes."""
+
+    # The new projects have uniform bucket-level access and our tests don't
+    # pass with those buckets. We need to use the old main project for now.
+    original_value = os.environ['GOOGLE_CLOUD_PROJECT']
+    os.environ['GOOGLE_CLOUD_PROJECT'] = os.environ['MAIN_GOOGLE_CLOUD_PROJECT']
+    bucket = None
+    while bucket is None or bucket.exists():
+        bucket_name = "acl-test-{}".format(uuid.uuid4())
+        bucket = storage.Client().bucket(bucket_name)
+    bucket.create()
+    yield bucket
+    bucket.delete(force=True)
+    # Set the value back.
+    os.environ['GOOGLE_CLOUD_PROJECT'] = original_value
+
+
+@pytest.fixture
+def test_blob(test_bucket):
+    """Yields a blob that is deleted after the test completes."""
+    bucket = test_bucket
+    blob = bucket.blob("storage_acl_test_sigil-{}".format(uuid.uuid4()))
+    blob.upload_from_string("Hello, is it me you're looking for?")
+    yield blob
+
+
+def test_print_bucket_acl(test_bucket, capsys):
+    storage_print_bucket_acl.print_bucket_acl(test_bucket.name)
+    out, _ = capsys.readouterr()
+    assert out
+
+
+def test_print_bucket_acl_for_user(test_bucket, capsys):
+    test_bucket.acl.user(TEST_EMAIL).grant_owner()
+    test_bucket.acl.save()
+
+    storage_print_bucket_acl_for_user.print_bucket_acl_for_user(
+        test_bucket.name, TEST_EMAIL
+    )
+
+    out, _ = capsys.readouterr()
+    assert "OWNER" in out
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_add_bucket_owner(test_bucket):
+    storage_add_bucket_owner.add_bucket_owner(test_bucket.name, TEST_EMAIL)
+
+    test_bucket.acl.reload()
+    assert "OWNER" in test_bucket.acl.user(TEST_EMAIL).get_roles()
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_remove_bucket_owner(test_bucket):
+    test_bucket.acl.user(TEST_EMAIL).grant_owner()
+    test_bucket.acl.save()
+
+    storage_remove_bucket_owner.remove_bucket_owner(
+        test_bucket.name, TEST_EMAIL)
+
+    test_bucket.acl.reload()
+    assert "OWNER" not in test_bucket.acl.user(TEST_EMAIL).get_roles()
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_add_bucket_default_owner(test_bucket):
+    storage_add_bucket_default_owner.add_bucket_default_owner(
+        test_bucket.name, TEST_EMAIL
+    )
+
+    test_bucket.default_object_acl.reload()
+    roles = test_bucket.default_object_acl.user(TEST_EMAIL).get_roles()
+    assert "OWNER" in roles
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_remove_bucket_default_owner(test_bucket):
+    test_bucket.acl.user(TEST_EMAIL).grant_owner()
+    test_bucket.acl.save()
+
+    storage_remove_bucket_default_owner.remove_bucket_default_owner(
+        test_bucket.name, TEST_EMAIL
+    )
+
+    test_bucket.default_object_acl.reload()
+    roles = test_bucket.default_object_acl.user(TEST_EMAIL).get_roles()
+    assert "OWNER" not in roles
+
+
+def test_print_blob_acl(test_blob, capsys):
+    storage_print_file_acl.print_blob_acl(
+        test_blob.bucket.name, test_blob.name)
+    out, _ = capsys.readouterr()
+    assert out
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_print_blob_acl_for_user(test_blob, capsys):
+    test_blob.acl.user(TEST_EMAIL).grant_owner()
+    test_blob.acl.save()
+
+    storage_print_file_acl_for_user.print_blob_acl_for_user(
+        test_blob.bucket.name, test_blob.name, TEST_EMAIL
+    )
+
+    out, _ = capsys.readouterr()
+    assert "OWNER" in out
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_add_blob_owner(test_blob):
+    storage_add_file_owner.add_blob_owner(
+        test_blob.bucket.name, test_blob.name, TEST_EMAIL)
+
+    test_blob.acl.reload()
+    assert "OWNER" in test_blob.acl.user(TEST_EMAIL).get_roles()
+
+
+@backoff.on_exception(backoff.expo, HttpError, max_time=60)
+def test_remove_blob_owner(test_blob):
+    test_blob.acl.user(TEST_EMAIL).grant_owner()
+    test_blob.acl.save()
+
+    storage_remove_file_owner.remove_blob_owner(
+        test_blob.bucket.name, test_blob.name, TEST_EMAIL
+    )
+
+    test_blob.acl.reload()
+    assert "OWNER" not in test_blob.acl.user(TEST_EMAIL).get_roles()

samples/snippets/bucket_lock_test.py

@@ -0,0 +1,176 @@
+# Copyright 2018 Google Inc. 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 time
+import uuid
+
+from google.cloud import storage
+import pytest
+
+import storage_disable_default_event_based_hold
+import storage_enable_default_event_based_hold
+import storage_get_default_event_based_hold
+import storage_get_retention_policy
+import storage_lock_retention_policy
+import storage_release_event_based_hold
+import storage_release_temporary_hold
+import storage_remove_retention_policy
+import storage_set_event_based_hold
+import storage_set_retention_policy
+import storage_set_temporary_hold
+
+
+BLOB_NAME = "storage_snippets_test_sigil"
+BLOB_CONTENT = "Hello, is it me you're looking for?"
+# Retention policy for 5 seconds
+RETENTION_POLICY = 5
+
+
+@pytest.fixture
+def bucket():
+    """Yields a bucket that is deleted after the test completes."""
+    bucket = None
+    while bucket is None or bucket.exists():
+        bucket_name = "bucket-lock-{}".format(uuid.uuid4())
+        bucket = storage.Client().bucket(bucket_name)
+    bucket.create()
+    yield bucket
+    bucket.delete(force=True)
+
+
+def test_retention_policy_no_lock(bucket, capsys):
+    storage_set_retention_policy.set_retention_policy(
+        bucket.name, RETENTION_POLICY
+    )
+    bucket.reload()
+
+    assert bucket.retention_period is RETENTION_POLICY
+    assert bucket.retention_policy_effective_time is not None
+    assert bucket.retention_policy_locked is None
+
+    storage_get_retention_policy.get_retention_policy(bucket.name)
+    out, _ = capsys.readouterr()
+    assert "Retention Policy for {}".format(bucket.name) in out
+    assert "Retention Period: 5" in out
+    assert "Effective Time: " in out
+    assert "Retention Policy is locked" not in out
+
+    blob = bucket.blob(BLOB_NAME)
+    blob.upload_from_string(BLOB_CONTENT)
+
+    assert blob.retention_expiration_time is not None
+
+    storage_remove_retention_policy.remove_retention_policy(bucket.name)
+    bucket.reload()
+    assert bucket.retention_period is None
+
+    time.sleep(RETENTION_POLICY)
+
+
+def test_retention_policy_lock(bucket, capsys):
+    storage_set_retention_policy.set_retention_policy(
+        bucket.name, RETENTION_POLICY
+    )
+    bucket.reload()
+    assert bucket.retention_policy_locked is None
+
+    storage_lock_retention_policy.lock_retention_policy(bucket.name)
+    bucket.reload()
+    assert bucket.retention_policy_locked is True
+
+    storage_get_retention_policy.get_retention_policy(bucket.name)
+    out, _ = capsys.readouterr()
+    assert "Retention Policy is locked" in out
+
+
+def test_enable_disable_bucket_default_event_based_hold(bucket, capsys):
+    storage_get_default_event_based_hold.get_default_event_based_hold(
+        bucket.name
+    )
+    out, _ = capsys.readouterr()
+    assert (
+        "Default event-based hold is not enabled for {}".format(bucket.name)
+        in out
+    )
+    assert (
+        "Default event-based hold is enabled for {}".format(bucket.name)
+        not in out
+    )
+
+    storage_enable_default_event_based_hold.enable_default_event_based_hold(
+        bucket.name
+    )
+    bucket.reload()
+
+    assert bucket.default_event_based_hold is True
+
+    storage_get_default_event_based_hold.get_default_event_based_hold(
+        bucket.name
+    )
+    out, _ = capsys.readouterr()
+    assert (
+        "Default event-based hold is enabled for {}".format(bucket.name) in out
+    )
+
+    # Changes to the bucket will be readable immediately after writing,
+    # but configuration changes may take time to propagate.
+    time.sleep(10)
+
+    blob = bucket.blob(BLOB_NAME)
+    blob.upload_from_string(BLOB_CONTENT)
+    assert blob.event_based_hold is True
+
+    storage_release_event_based_hold.release_event_based_hold(
+        bucket.name, blob.name
+    )
+    blob.reload()
+    assert blob.event_based_hold is False
+
+    storage_disable_default_event_based_hold.disable_default_event_based_hold(
+        bucket.name
+    )
+    bucket.reload()
+    assert bucket.default_event_based_hold is False
+
+
+def test_enable_disable_temporary_hold(bucket):
+    blob = bucket.blob(BLOB_NAME)
+    blob.upload_from_string(BLOB_CONTENT)
+    assert blob.temporary_hold is None
+
+    storage_set_temporary_hold.set_temporary_hold(bucket.name, blob.name)
+    blob.reload()
+    assert blob.temporary_hold is True
+
+    storage_release_temporary_hold.release_temporary_hold(
+        bucket.name, blob.name
+    )
+    blob.reload()
+    assert blob.temporary_hold is False
+
+
+def test_enable_disable_event_based_hold(bucket):
+    blob = bucket.blob(BLOB_NAME)
+    blob.upload_from_string(BLOB_CONTENT)
+    assert blob.event_based_hold is None
+
+    storage_set_event_based_hold.set_event_based_hold(bucket.name, blob.name)
+    blob.reload()
+    assert blob.event_based_hold is True
+
+    storage_release_event_based_hold.release_event_based_hold(
+        bucket.name, blob.name
+    )
+    blob.reload()
+    assert blob.event_based_hold is False