Root Explorer PHP

[ Root System Explorer ]

Editing: test_bucketconfig.py

# -*- coding: utf-8 -*-
# Copyright 2014 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.
"""Integration tests for multiple bucket configuration commands."""

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import json
import gslib.tests.testcase as testcase
from gslib.tests.testcase.integration_testcase import SkipForS3
from gslib.tests.util import ObjectToURI as suri
from gslib.utils.constants import UTF8

class TestBucketConfig(testcase.GsUtilIntegrationTestCase):
  """Integration tests for multiple bucket configuration commands."""

_set_cors_command = ['cors', 'set']
  _get_cors_command = ['cors', 'get']

empty_cors = '[]'

cors_doc = (
      '[{"origin": ["http://origin1.example.com", '
      '"http://origin2.example.com"], '
      '"responseHeader": ["foo", "bar"], "method": ["GET", "PUT", "POST"], '
      '"maxAgeSeconds": 3600},'
      '{"origin": ["http://origin3.example.com"], '
      '"responseHeader": ["foo2", "bar2"], "method": ["GET", "DELETE"]}]\n')
  cors_json_obj = json.loads(cors_doc)

_set_lifecycle_command = ['lifecycle', 'set']
  _get_lifecycle_command = ['lifecycle', 'get']

empty_lifecycle = '{}'

lifecycle_doc = (
      '{"rule": [{"action": {"type": "Delete"}, "condition": {"age": 365}}]}\n')
  lifecycle_json_obj = json.loads(lifecycle_doc)

_set_acl_command = ['acl', 'set']
  _get_acl_command = ['acl', 'get']
  _set_defacl_command = ['defacl', 'set']
  _get_defacl_command = ['defacl', 'get']

@SkipForS3('A number of configs in this test are not supported by S3')
  def test_set_multi_config(self):
    """Tests that bucket config patching affects only the desired config."""
    bucket_uri = self.CreateBucket()
    lifecycle_path = self.CreateTempFile(
        contents=self.lifecycle_doc.encode(UTF8))
    cors_path = self.CreateTempFile(contents=self.cors_doc.encode(UTF8))

self.RunGsUtil(self._set_cors_command + [cors_path, suri(bucket_uri)])
    cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
                              return_stdout=True)
    self.assertEqual(json.loads(cors_out), self.cors_json_obj)

self.RunGsUtil(self._set_lifecycle_command +
                   [lifecycle_path, suri(bucket_uri)])
    cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
                              return_stdout=True)
    lifecycle_out = self.RunGsUtil(self._get_lifecycle_command +
                                   [suri(bucket_uri)],
                                   return_stdout=True)
    self.assertEqual(json.loads(cors_out), self.cors_json_obj)
    self.assertEqual(json.loads(lifecycle_out), self.lifecycle_json_obj)

if not self._ServiceAccountCredentialsPresent():
      # See comments in _ServiceAccountCredentialsPresent
      self.RunGsUtil(
          self._set_acl_command +
          ['authenticated-read', suri(bucket_uri)])

cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
                              return_stdout=True)
    lifecycle_out = self.RunGsUtil(self._get_lifecycle_command +
                                   [suri(bucket_uri)],
                                   return_stdout=True)
    self.assertEqual(json.loads(cors_out), self.cors_json_obj)
    self.assertEqual(json.loads(lifecycle_out), self.lifecycle_json_obj)

if not self._ServiceAccountCredentialsPresent():
      acl_out = self.RunGsUtil(self._get_acl_command + [suri(bucket_uri)],
                               return_stdout=True)
      self.assertIn('allAuthenticatedUsers', acl_out)

self.RunGsUtil(self._set_defacl_command + ['public-read', suri(bucket_uri)])

cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
                              return_stdout=True)
    lifecycle_out = self.RunGsUtil(self._get_lifecycle_command +
                                   [suri(bucket_uri)],
                                   return_stdout=True)
    def_acl_out = self.RunGsUtil(self._get_defacl_command + [suri(bucket_uri)],
                                 return_stdout=True)
    self.assertEqual(json.loads(cors_out), self.cors_json_obj)
    self.assertEqual(json.loads(lifecycle_out), self.lifecycle_json_obj)
    self.assertIn('allUsers', def_acl_out)

[ CANCEL ]

Name	Type	Actions
.. (Parent Directory)
📄 __init__.py	FILE	[EDIT]
📁 __pycache__/	DIR
📄 mock_cloud_api.py	FILE	[EDIT]
📄 mock_logging_handler.py	FILE	[EDIT]
📄 rewrite_helper.py	FILE	[EDIT]
📄 signurl_signatures.py	FILE	[EDIT]
📄 test_Doption.py	FILE	[EDIT]
📄 test_acl.py	FILE	[EDIT]
📄 test_autoclass.py	FILE	[EDIT]
📄 test_boto_util.py	FILE	[EDIT]
📄 test_bucketconfig.py	FILE	[EDIT]
📄 test_bucketpolicyonly.py	FILE	[EDIT]
📄 test_cat.py	FILE	[EDIT]
📄 test_cloud_api_delegator.py	FILE	[EDIT]
📄 test_command.py	FILE	[EDIT]
📄 test_command_runner.py	FILE	[EDIT]
📄 test_compose.py	FILE	[EDIT]
📄 test_context_config.py	FILE	[EDIT]
📄 test_copy_helper_funcs.py	FILE	[EDIT]
📄 test_copy_objects_iterator.py	FILE	[EDIT]
📄 test_cors.py	FILE	[EDIT]
📄 test_cp.py	FILE	[EDIT]
📄 test_creds_config.py	FILE	[EDIT]
📄 test_daisy_chain_wrapper.py	FILE	[EDIT]
📁 test_data/	DIR
📄 test_defacl.py	FILE	[EDIT]
📄 test_defstorageclass.py	FILE	[EDIT]
📄 test_du.py	FILE	[EDIT]
📄 test_encryption_helper.py	FILE	[EDIT]
📄 test_execution_util.py	FILE	[EDIT]
📄 test_file_part.py	FILE	[EDIT]
📄 test_gcs_json_api.py	FILE	[EDIT]
📄 test_gcs_json_credentials.py	FILE	[EDIT]
📄 test_gcs_json_media.py	FILE	[EDIT]
📄 test_gsutil.py	FILE	[EDIT]
📄 test_hash.py	FILE	[EDIT]
📄 test_hashing_helper.py	FILE	[EDIT]
📄 test_help.py	FILE	[EDIT]
📄 test_hmac.py	FILE	[EDIT]
📄 test_iam.py	FILE	[EDIT]
📄 test_kms.py	FILE	[EDIT]
📄 test_label.py	FILE	[EDIT]
📄 test_lifecycle.py	FILE	[EDIT]
📄 test_logging.py	FILE	[EDIT]
📄 test_ls.py	FILE	[EDIT]
📄 test_mb.py	FILE	[EDIT]
📄 test_metrics.py	FILE	[EDIT]
📄 test_mtls.py	FILE	[EDIT]
📄 test_mv.py	FILE	[EDIT]
📄 test_naming.py	FILE	[EDIT]
📄 test_notification.py	FILE	[EDIT]
📄 test_notification_pubsub.py	FILE	[EDIT]
📄 test_pap.py	FILE	[EDIT]
📄 test_parallel_cp.py	FILE	[EDIT]
📄 test_parallelism_framework.py	FILE	[EDIT]
📄 test_perfdiag.py	FILE	[EDIT]
📄 test_plurality_checkable_iterator.py	FILE	[EDIT]
📄 test_posix_util.py	FILE	[EDIT]
📄 test_psc.py	FILE	[EDIT]
📄 test_rb.py	FILE	[EDIT]
📄 test_requester_pays.py	FILE	[EDIT]
📄 test_resumable_streaming.py	FILE	[EDIT]
📄 test_retention.py	FILE	[EDIT]
📄 test_retention_util.py	FILE	[EDIT]
📄 test_rewrite.py	FILE	[EDIT]
📄 test_rm.py	FILE	[EDIT]
📄 test_rpo.py	FILE	[EDIT]
📄 test_rsync.py	FILE	[EDIT]
📄 test_rsync_funcs.py	FILE	[EDIT]
📄 test_seek_ahead_thread.py	FILE	[EDIT]
📄 test_setmeta.py	FILE	[EDIT]
📄 test_shim_util.py	FILE	[EDIT]
📄 test_signurl.py	FILE	[EDIT]
📄 test_stat.py	FILE	[EDIT]
📄 test_stet_cp.py	FILE	[EDIT]
📄 test_stet_util.py	FILE	[EDIT]
📄 test_storage_url.py	FILE	[EDIT]
📄 test_tabcomplete.py	FILE	[EDIT]
📄 test_temporary_file_util.py	FILE	[EDIT]
📄 test_trace.py	FILE	[EDIT]
📄 test_tracker_file.py	FILE	[EDIT]
📄 test_ubla.py	FILE	[EDIT]
📄 test_ui.py	FILE	[EDIT]
📄 test_update.py	FILE	[EDIT]
📄 test_user_agent_helper.py	FILE	[EDIT]
📄 test_util.py	FILE	[EDIT]
📄 test_versioning.py	FILE	[EDIT]
📄 test_web.py	FILE	[EDIT]
📄 test_wildcard_iterator.py	FILE	[EDIT]
📄 test_wrapped_credentials.py	FILE	[EDIT]
📁 testcase/	DIR
📄 util.py	FILE	[EDIT]