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

   1 # -*- coding: utf-8 -*-
   2 # Copyright 2013 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 the defacl command."""
  16
  17 from __future__ import absolute_import
  18
  19 import os
  20 import re
  21 import gslib.tests.testcase as case
  22 from gslib.tests.testcase.integration_testcase import SkipForS3
  23 from gslib.tests.util import ObjectToURI as suri
  24
  25 PUBLIC_READ_JSON_ACL_TEXT = '"entity":"allUsers","role":"READER"'
  26
  27
  28 @SkipForS3('S3 does not support default object ACLs.')
  29 class TestDefacl(case.GsUtilIntegrationTestCase):
  30   """Integration tests for the defacl command."""
  31
  32   _defacl_ch_prefix = ['defacl', 'ch']
  33   _defacl_get_prefix = ['defacl', 'get']
  34   _defacl_set_prefix = ['defacl', 'set']
  35
  36   def _MakeScopeRegex(self, role, entity_type, email_address):
  37     template_regex = (r'\{.*"entity":\s*"%s-%s".*"role":\s*"%s".*\}' %
  38                       (entity_type, email_address, role))
  39     return re.compile(template_regex, flags=re.DOTALL)
  40
  41   def testChangeDefaultAcl(self):
  42     """Tests defacl ch."""
  43     bucket = self.CreateBucket()
  44
  45     test_regex = self._MakeScopeRegex(
  46         'OWNER', 'group', self.GROUP_TEST_ADDRESS)
  47     test_regex2 = self._MakeScopeRegex(
  48         'READER', 'group', self.GROUP_TEST_ADDRESS)
  49     json_text = self.RunGsUtil(self._defacl_get_prefix +
  50                                [suri(bucket)], return_stdout=True)
  51     self.assertNotRegexpMatches(json_text, test_regex)
  52
  53     self.RunGsUtil(self._defacl_ch_prefix +
  54                    ['-g', self.GROUP_TEST_ADDRESS+':FC', suri(bucket)])
  55     json_text2 = self.RunGsUtil(self._defacl_get_prefix +
  56                                 [suri(bucket)], return_stdout=True)
  57     self.assertRegexpMatches(json_text2, test_regex)
  58
  59     self.RunGsUtil(self._defacl_ch_prefix +
  60                    ['-g', self.GROUP_TEST_ADDRESS+':READ', suri(bucket)])
  61     json_text3 = self.RunGsUtil(self._defacl_get_prefix +
  62                                 [suri(bucket)], return_stdout=True)
  63     self.assertRegexpMatches(json_text3, test_regex2)
  64
  65     stderr = self.RunGsUtil(self._defacl_ch_prefix +
  66                             ['-g', self.GROUP_TEST_ADDRESS+':WRITE',
  67                              suri(bucket)],
  68                             return_stderr=True, expected_status=1)
  69     self.assertIn('WRITER cannot be set as a default object ACL', stderr)
  70
  71   def testChangeDefaultAclPrivate(self):
  72     bucket = self.CreateBucket()
  73     test_regex = self._MakeScopeRegex(
  74         'READER', 'group', self.GROUP_TEST_ADDRESS)
  75     self.RunGsUtil(self._defacl_set_prefix + ['private', suri(bucket)])
  76     json_text = self.RunGsUtil(self._defacl_get_prefix +
  77                                [suri(bucket)], return_stdout=True)
  78     self.assertRegexpMatches(json_text, r'\[\]\s*')
  79
  80     self.RunGsUtil(self._defacl_ch_prefix +
  81                    ['-g', self.GROUP_TEST_ADDRESS+':READ', suri(bucket)])
  82     json_text2 = self.RunGsUtil(self._defacl_get_prefix +
  83                                 [suri(bucket)], return_stdout=True)
  84     self.assertRegexpMatches(json_text2, test_regex)
  85
  86   def testChangeMultipleBuckets(self):
  87     """Tests defacl ch on multiple buckets."""
  88     bucket1 = self.CreateBucket()
  89     bucket2 = self.CreateBucket()
  90
  91     test_regex = self._MakeScopeRegex(
  92         'READER', 'group', self.GROUP_TEST_ADDRESS)
  93     json_text = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket1)],
  94                                return_stdout=True)
  95     self.assertNotRegexpMatches(json_text, test_regex)
  96     json_text = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket2)],
  97                                return_stdout=True)
  98     self.assertNotRegexpMatches(json_text, test_regex)
  99
 100     self.RunGsUtil(self._defacl_ch_prefix +
 101                    ['-g', self.GROUP_TEST_ADDRESS+':READ',
 102                     suri(bucket1), suri(bucket2)])
 103     json_text = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket1)],
 104                                return_stdout=True)
 105     self.assertRegexpMatches(json_text, test_regex)
 106     json_text = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket2)],
 107                                return_stdout=True)
 108     self.assertRegexpMatches(json_text, test_regex)
 109
 110   def testChangeMultipleAcls(self):
 111     """Tests defacl ch with multiple ACL entries."""
 112     bucket = self.CreateBucket()
 113
 114     test_regex_group = self._MakeScopeRegex(
 115         'READER', 'group', self.GROUP_TEST_ADDRESS)
 116     test_regex_user = self._MakeScopeRegex(
 117         'OWNER', 'user', self.USER_TEST_ADDRESS)
 118     json_text = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket)],
 119                                return_stdout=True)
 120     self.assertNotRegexpMatches(json_text, test_regex_group)
 121     self.assertNotRegexpMatches(json_text, test_regex_user)
 122
 123     self.RunGsUtil(self._defacl_ch_prefix +
 124                    ['-g', self.GROUP_TEST_ADDRESS+':READ',
 125                     '-u', self.USER_TEST_ADDRESS+':fc', suri(bucket)])
 126     json_text = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket)],
 127                                return_stdout=True)
 128     self.assertRegexpMatches(json_text, test_regex_group)
 129     self.assertRegexpMatches(json_text, test_regex_user)
 130
 131   def testEmptyDefAcl(self):
 132     bucket = self.CreateBucket()
 133     self.RunGsUtil(self._defacl_set_prefix + ['private', suri(bucket)])
 134     stdout = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket)],
 135                             return_stdout=True)
 136     self.assertEquals(stdout.rstrip(), '[]')
 137     self.RunGsUtil(self._defacl_ch_prefix +
 138                    ['-u', self.USER_TEST_ADDRESS+':fc', suri(bucket)])
 139
 140   def testDeletePermissionsWithCh(self):
 141     """Tests removing permissions with defacl ch."""
 142     bucket = self.CreateBucket()
 143
 144     test_regex = self._MakeScopeRegex(
 145         'OWNER', 'user', self.USER_TEST_ADDRESS)
 146     json_text = self.RunGsUtil(
 147         self._defacl_get_prefix + [suri(bucket)], return_stdout=True)
 148     self.assertNotRegexpMatches(json_text, test_regex)
 149
 150     self.RunGsUtil(self._defacl_ch_prefix +
 151                    ['-u', self.USER_TEST_ADDRESS+':fc', suri(bucket)])
 152     json_text = self.RunGsUtil(
 153         self._defacl_get_prefix + [suri(bucket)], return_stdout=True)
 154     self.assertRegexpMatches(json_text, test_regex)
 155
 156     self.RunGsUtil(self._defacl_ch_prefix +
 157                    ['-d', self.USER_TEST_ADDRESS, suri(bucket)])
 158     json_text = self.RunGsUtil(
 159         self._defacl_get_prefix + [suri(bucket)], return_stdout=True)
 160     self.assertNotRegexpMatches(json_text, test_regex)
 161
 162   def testTooFewArgumentsFails(self):
 163     """Tests calling defacl with insufficient number of arguments."""
 164     # No arguments for get, but valid subcommand.
 165     stderr = self.RunGsUtil(self._defacl_get_prefix, return_stderr=True,
 166                             expected_status=1)
 167     self.assertIn('command requires at least', stderr)
 168
 169     # No arguments for set, but valid subcommand.
 170     stderr = self.RunGsUtil(self._defacl_set_prefix, return_stderr=True,
 171                             expected_status=1)
 172     self.assertIn('command requires at least', stderr)
 173
 174     # No arguments for ch, but valid subcommand.
 175     stderr = self.RunGsUtil(self._defacl_ch_prefix, return_stderr=True,
 176                             expected_status=1)
 177     self.assertIn('command requires at least', stderr)
 178
 179     # Neither arguments nor subcommand.
 180     stderr = self.RunGsUtil(['defacl'], return_stderr=True, expected_status=1)
 181     self.assertIn('command requires at least', stderr)
 182
 183
 184 class TestDefaclOldAlias(TestDefacl):
 185   _defacl_ch_prefix = ['chdefacl']
 186   _defacl_get_prefix = ['getdefacl']
 187   _defacl_set_prefix = ['setdefacl']