tools/telemetry/third_party/gsutilz/gslib/tests/test_bucketconfig.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright 2014 Google Inc. All Rights Reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 """Integration tests for multiple bucket configuration commands."""
  16
  17 import json
  18 import gslib.tests.testcase as testcase
  19 from gslib.tests.testcase.integration_testcase import SkipForS3
  20 from gslib.tests.util import ObjectToURI as suri
  21
  22
  23 class TestBucketConfig(testcase.GsUtilIntegrationTestCase):
  24   """Integration tests for multiple bucket configuration commands."""
  25
  26   _set_cors_command = ['cors', 'set']
  27   _get_cors_command = ['cors', 'get']
  28
  29   empty_cors = '[]'
  30
  31   cors_doc = (
  32       '[{"origin": ["http://origin1.example.com", '
  33       '"http://origin2.example.com"], '
  34       '"responseHeader": ["foo", "bar"], "method": ["GET", "PUT", "POST"], '
  35       '"maxAgeSeconds": 3600},'
  36       '{"origin": ["http://origin3.example.com"], '
  37       '"responseHeader": ["foo2", "bar2"], "method": ["GET", "DELETE"]}]\n')
  38   cors_json_obj = json.loads(cors_doc)
  39
  40   _set_lifecycle_command = ['lifecycle', 'set']
  41   _get_lifecycle_command = ['lifecycle', 'get']
  42
  43   empty_lifecycle = '{}'
  44
  45   lifecycle_doc = (
  46       '{"rule": [{"action": {"type": "Delete"}, "condition": {"age": 365}}]}\n')
  47   lifecycle_json_obj = json.loads(lifecycle_doc)
  48
  49   _set_acl_command = ['acl', 'set']
  50   _get_acl_command = ['acl', 'get']
  51   _set_defacl_command = ['defacl', 'set']
  52   _get_defacl_command = ['defacl', 'get']
  53
  54   @SkipForS3('A number of configs in this test are not supported by S3')
  55   def test_set_multi_config(self):
  56     """Tests that bucket config patching affects only the desired config."""
  57     bucket_uri = self.CreateBucket()
  58     lifecycle_path = self.CreateTempFile(contents=self.lifecycle_doc)
  59     cors_path = self.CreateTempFile(contents=self.cors_doc)
  60
  61     self.RunGsUtil(self._set_cors_command + [cors_path, suri(bucket_uri)])
  62     cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
  63                               return_stdout=True)
  64     self.assertEqual(json.loads(cors_out), self.cors_json_obj)
  65
  66     self.RunGsUtil(self._set_lifecycle_command + [lifecycle_path,
  67                                                   suri(bucket_uri)])
  68     cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
  69                               return_stdout=True)
  70     lifecycle_out = self.RunGsUtil(self._get_lifecycle_command +
  71                                    [suri(bucket_uri)], return_stdout=True)
  72     self.assertEqual(json.loads(cors_out), self.cors_json_obj)
  73     self.assertEqual(json.loads(lifecycle_out), self.lifecycle_json_obj)
  74
  75     self.RunGsUtil(
  76         self._set_acl_command + ['authenticated-read', suri(bucket_uri)])
  77
  78     cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
  79                               return_stdout=True)
  80     lifecycle_out = self.RunGsUtil(self._get_lifecycle_command +
  81                                    [suri(bucket_uri)], return_stdout=True)
  82     acl_out = self.RunGsUtil(self._get_acl_command + [suri(bucket_uri)],
  83                              return_stdout=True)
  84     self.assertEqual(json.loads(cors_out), self.cors_json_obj)
  85     self.assertEqual(json.loads(lifecycle_out), self.lifecycle_json_obj)
  86     self.assertIn('allAuthenticatedUsers', acl_out)
  87
  88     self.RunGsUtil(
  89         self._set_defacl_command + ['public-read', suri(bucket_uri)])
  90
  91     cors_out = self.RunGsUtil(self._get_cors_command + [suri(bucket_uri)],
  92                               return_stdout=True)
  93     lifecycle_out = self.RunGsUtil(self._get_lifecycle_command +
  94                                    [suri(bucket_uri)], return_stdout=True)
  95     acl_out = self.RunGsUtil(self._get_acl_command + [suri(bucket_uri)],
  96                              return_stdout=True)
  97     def_acl_out = self.RunGsUtil(self._get_defacl_command + [suri(bucket_uri)],
  98                                  return_stdout=True)
  99     self.assertEqual(json.loads(cors_out), self.cors_json_obj)
 100     self.assertEqual(json.loads(lifecycle_out), self.lifecycle_json_obj)
 101     self.assertIn('allAuthenticatedUsers', acl_out)
 102     self.assertIn('allUsers', def_acl_out)