def test_s3_multipart_upload_acls(self):
    bucket_name = ('test-bucket-%s' % short_uid())
    self.s3_client.create_bucket(Bucket=bucket_name, ACL='public-read')

    def check_permissions(key, expected_perms):
        grants = self.s3_client.get_object_acl(Bucket=bucket_name, Key=key)['Grants']
        grants = [g for g in grants if ('AllUsers' in g.get('Grantee', {}).get('URI', ''))]
        self.assertEquals(len(grants), 1)
        permissions = grants[0]['Permission']
        permissions = (permissions if isinstance(permissions, list) else [permissions])
        self.assertEquals(len(permissions), expected_perms)
    self.s3_client.put_object(Bucket=bucket_name, Key='acl-key0', Body='something')
    check_permissions('acl-key0', 1)
    self._perform_multipart_upload(bucket=bucket_name, key='acl-key1')
    check_permissions('acl-key1', 1)
    self._perform_multipart_upload(bucket=bucket_name, key='acl-key2', acl='public-read-write')
    check_permissions('acl-key2', 2)
def test_s3_multipart_upload_acls(self):
    bucket_name = ('test-bucket-%s' % short_uid())
    self.s3_client.create_bucket(Bucket=bucket_name, ACL='public-read')

    def check_permissions(key, expected_perms):
        grants = self.s3_client.get_object_acl(Bucket=bucket_name, Key=key)['Grants']
        grants = [g for g in grants if ('AllUsers' in g.get('Grantee', {}).get('URI', ))]
        self.assertEquals(len(grants), 1)
        permissions = grants[0]['Permission']
        permissions = (permissions if isinstance(permissions, list) else [permissions])
        self.assertEquals(len(permissions), expected_perms)
    self.s3_client.put_object(Bucket=bucket_name, Key='acl-key0', Body='something')
    check_permissions('acl-key0', 1)
    self._perform_multipart_upload(bucket=bucket_name, key='acl-key1')
    check_permissions('acl-key1', 1)
    self._perform_multipart_upload(bucket=bucket_name, key='acl-key2', acl='public-read-write')
    check_permissions('acl-key2', 2)