551 lines
24 KiB
Python
551 lines
24 KiB
Python
# Copyright (c) 2014 OpenStack Foundation
|
|
#
|
|
# 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 unittest
|
|
import functools
|
|
import sys
|
|
import traceback
|
|
from mock import patch, MagicMock
|
|
|
|
from swift.common import swob
|
|
from swift.common.swob import Request
|
|
from swift.common.utils import json
|
|
|
|
from swift.common.middleware.s3api.etree import tostring, Element, SubElement
|
|
from swift.common.middleware.s3api.subresource import ACL, ACLPrivate, User, \
|
|
encode_acl, AuthenticatedUsers, AllUsers, Owner, Grant, PERMISSIONS
|
|
from test.unit.common.middleware.s3api.test_s3api import S3ApiTestCase
|
|
from test.unit.common.middleware.s3api.exceptions import NotMethodException
|
|
from test.unit.common.middleware.s3api import FakeSwift
|
|
|
|
|
|
XMLNS_XSI = 'http://www.w3.org/2001/XMLSchema-instance'
|
|
|
|
|
|
def s3acl(func=None, s3acl_only=False, versioning_enabled=True):
|
|
"""
|
|
NOTE: s3acl decorator needs an instance of s3api testing framework.
|
|
(i.e. An instance for first argument is necessary)
|
|
"""
|
|
if func is None:
|
|
return functools.partial(
|
|
s3acl,
|
|
s3acl_only=s3acl_only,
|
|
versioning_enabled=versioning_enabled)
|
|
|
|
@functools.wraps(func)
|
|
def s3acl_decorator(*args, **kwargs):
|
|
if not args and not kwargs:
|
|
raise NotMethodException('Use s3acl decorator for a method')
|
|
|
|
def call_func(failing_point=''):
|
|
try:
|
|
# For maintainability, we patch 204 status for every
|
|
# get_container_info. if you want, we can rewrite the
|
|
# statement easily with nested decorator like as:
|
|
#
|
|
# @s3acl
|
|
# @patch(xxx)
|
|
# def test_xxxx(self)
|
|
|
|
fake_info = {'status': 204}
|
|
if versioning_enabled:
|
|
fake_info['sysmeta'] = {
|
|
'versions-container': '\x00versions\x00bucket',
|
|
}
|
|
|
|
with patch('swift.common.middleware.s3api.s3request.'
|
|
'get_container_info', return_value=fake_info):
|
|
func(*args, **kwargs)
|
|
except AssertionError:
|
|
# Make traceback message to clarify the assertion
|
|
exc_type, exc_instance, exc_traceback = sys.exc_info()
|
|
formatted_traceback = ''.join(traceback.format_tb(
|
|
exc_traceback))
|
|
message = '\n%s\n%s' % (formatted_traceback,
|
|
exc_type.__name__)
|
|
if exc_instance.args:
|
|
message += ':\n%s' % (exc_instance.args[0],)
|
|
message += failing_point
|
|
raise exc_type(message)
|
|
|
|
instance = args[0]
|
|
|
|
if not s3acl_only:
|
|
call_func()
|
|
instance.swift._calls = []
|
|
|
|
instance.s3api.conf.s3_acl = True
|
|
instance.swift.s3_acl = True
|
|
owner = Owner('test:tester', 'test:tester')
|
|
generate_s3acl_environ('test', instance.swift, owner)
|
|
call_func(' (fail at s3_acl)')
|
|
|
|
return s3acl_decorator
|
|
|
|
|
|
def _gen_test_headers(owner, grants=[], resource='container'):
|
|
if not grants:
|
|
grants = [Grant(User('test:tester'), 'FULL_CONTROL')]
|
|
return encode_acl(resource, ACL(owner, grants))
|
|
|
|
|
|
def _make_xml(grantee):
|
|
owner = 'test:tester'
|
|
permission = 'READ'
|
|
elem = Element('AccessControlPolicy')
|
|
elem_owner = SubElement(elem, 'Owner')
|
|
SubElement(elem_owner, 'ID').text = owner
|
|
SubElement(elem_owner, 'DisplayName').text = owner
|
|
acl_list_elem = SubElement(elem, 'AccessControlList')
|
|
elem_grant = SubElement(acl_list_elem, 'Grant')
|
|
elem_grant.append(grantee)
|
|
SubElement(elem_grant, 'Permission').text = permission
|
|
return tostring(elem)
|
|
|
|
|
|
def generate_s3acl_environ(account, swift, owner):
|
|
|
|
def gen_grant(permission):
|
|
# generate Grant with a grantee named by "permission"
|
|
account_name = '%s:%s' % (account, permission.lower())
|
|
return Grant(User(account_name), permission)
|
|
|
|
grants = [gen_grant(perm) for perm in PERMISSIONS]
|
|
container_headers = _gen_test_headers(owner, grants)
|
|
object_headers = _gen_test_headers(owner, grants, 'object')
|
|
object_body = 'hello'
|
|
object_headers['Content-Length'] = len(object_body)
|
|
|
|
# TEST method is used to resolve a tenant name
|
|
swift.register('TEST', '/v1/AUTH_test', swob.HTTPMethodNotAllowed,
|
|
{}, None)
|
|
swift.register('TEST', '/v1/AUTH_X', swob.HTTPMethodNotAllowed,
|
|
{}, None)
|
|
|
|
# for bucket
|
|
swift.register('HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
|
|
container_headers, None)
|
|
swift.register('HEAD', '/v1/AUTH_test/bucket+segments', swob.HTTPNoContent,
|
|
container_headers, None)
|
|
swift.register('PUT', '/v1/AUTH_test/bucket',
|
|
swob.HTTPCreated, {}, None)
|
|
swift.register('GET', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
|
|
container_headers, json.dumps([]))
|
|
swift.register('POST', '/v1/AUTH_test/bucket',
|
|
swob.HTTPNoContent, {}, None)
|
|
swift.register('DELETE', '/v1/AUTH_test/bucket',
|
|
swob.HTTPNoContent, {}, None)
|
|
|
|
# necessary for canned-acl tests
|
|
public_headers = _gen_test_headers(owner, [Grant(AllUsers(), 'READ')])
|
|
swift.register('GET', '/v1/AUTH_test/public', swob.HTTPNoContent,
|
|
public_headers, json.dumps([]))
|
|
authenticated_headers = _gen_test_headers(
|
|
owner, [Grant(AuthenticatedUsers(), 'READ')], 'bucket')
|
|
swift.register('GET', '/v1/AUTH_test/authenticated',
|
|
swob.HTTPNoContent, authenticated_headers,
|
|
json.dumps([]))
|
|
|
|
# for object
|
|
swift.register('HEAD', '/v1/AUTH_test/bucket/object', swob.HTTPOk,
|
|
object_headers, None)
|
|
|
|
|
|
class TestS3ApiS3Acl(S3ApiTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestS3ApiS3Acl, self).setUp()
|
|
|
|
self.s3api.conf.s3_acl = True
|
|
self.swift.s3_acl = True
|
|
|
|
account = 'test'
|
|
owner_name = '%s:tester' % account
|
|
self.default_owner = Owner(owner_name, owner_name)
|
|
generate_s3acl_environ(account, self.swift, self.default_owner)
|
|
|
|
def tearDown(self):
|
|
self.s3api.conf.s3_acl = False
|
|
|
|
def test_bucket_acl_PUT_with_other_owner(self):
|
|
req = Request.blank('/bucket?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header()},
|
|
body=tostring(
|
|
ACLPrivate(
|
|
Owner(id='test:other',
|
|
name='test:other')).elem()))
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'AccessDenied')
|
|
|
|
def test_object_acl_PUT_xml_error(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header()},
|
|
body="invalid xml")
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'MalformedACLError')
|
|
|
|
def test_canned_acl_private(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'private'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_canned_acl_public_read(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'public-read'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_canned_acl_public_read_write(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'public-read-write'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_canned_acl_authenticated_read(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'authenticated-read'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_canned_acl_bucket_owner_read(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'bucket-owner-read'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_canned_acl_bucket_owner_full_control(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'bucket-owner-full-control'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_invalid_canned_acl(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-acl': 'invalid'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'InvalidArgument')
|
|
|
|
def _test_grant_header(self, permission):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-' + permission:
|
|
'id=test:tester'})
|
|
return self.call_s3api(req)
|
|
|
|
def test_grant_read(self):
|
|
status, headers, body = self._test_grant_header('read')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_write(self):
|
|
status, headers, body = self._test_grant_header('write')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_read_acp(self):
|
|
status, headers, body = self._test_grant_header('read-acp')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_write_acp(self):
|
|
status, headers, body = self._test_grant_header('write-acp')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_full_control(self):
|
|
status, headers, body = self._test_grant_header('full-control')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_invalid_permission(self):
|
|
status, headers, body = self._test_grant_header('invalid')
|
|
self.assertEqual(self._get_error_code(body), 'MissingSecurityHeader')
|
|
|
|
def test_grant_with_both_header_and_xml(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-full-control':
|
|
'id=test:tester'},
|
|
body=tostring(
|
|
ACLPrivate(
|
|
Owner(id='test:tester',
|
|
name='test:tester')).elem()))
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'UnexpectedContent')
|
|
|
|
def test_grant_with_both_header_and_canned_acl(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-full-control':
|
|
'id=test:tester',
|
|
'x-amz-acl': 'public-read'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'InvalidRequest')
|
|
|
|
def test_grant_email(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-read': 'emailAddress=a@b.c'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'NotImplemented')
|
|
|
|
def test_grant_email_xml(self):
|
|
grantee = Element('Grantee', nsmap={'xsi': XMLNS_XSI})
|
|
grantee.set('{%s}type' % XMLNS_XSI, 'AmazonCustomerByEmail')
|
|
SubElement(grantee, 'EmailAddress').text = 'Grantees@email.com'
|
|
xml = _make_xml(grantee=grantee)
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header()},
|
|
body=xml)
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'NotImplemented')
|
|
|
|
def test_grant_invalid_group_xml(self):
|
|
grantee = Element('Grantee', nsmap={'xsi': XMLNS_XSI})
|
|
grantee.set('{%s}type' % XMLNS_XSI, 'Invalid')
|
|
xml = _make_xml(grantee=grantee)
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header()},
|
|
body=xml)
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'MalformedACLError')
|
|
|
|
def test_grant_authenticated_users(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-read':
|
|
'uri="http://acs.amazonaws.com/groups/'
|
|
'global/AuthenticatedUsers"'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_all_users(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-read':
|
|
'uri="http://acs.amazonaws.com/groups/'
|
|
'global/AllUsers"'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_grant_invalid_uri(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-read':
|
|
'uri="http://localhost/"'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'InvalidArgument')
|
|
|
|
def test_grant_invalid_uri_xml(self):
|
|
grantee = Element('Grantee', nsmap={'xsi': XMLNS_XSI})
|
|
grantee.set('{%s}type' % XMLNS_XSI, 'Group')
|
|
SubElement(grantee, 'URI').text = 'invalid'
|
|
xml = _make_xml(grantee)
|
|
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header()},
|
|
body=xml)
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'InvalidArgument')
|
|
|
|
def test_grant_invalid_target(self):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS test:tester:hmac',
|
|
'Date': self.get_date_header(),
|
|
'x-amz-grant-read': 'key=value'})
|
|
status, headers, body = self.call_s3api(req)
|
|
self.assertEqual(self._get_error_code(body), 'InvalidArgument')
|
|
|
|
def _test_bucket_acl_GET(self, account):
|
|
req = Request.blank('/bucket?acl',
|
|
environ={'REQUEST_METHOD': 'GET'},
|
|
headers={'Authorization': 'AWS %s:hmac' % account,
|
|
'Date': self.get_date_header()})
|
|
return self.call_s3api(req)
|
|
|
|
def test_bucket_acl_GET_without_permission(self):
|
|
status, headers, body = self._test_bucket_acl_GET('test:other')
|
|
self.assertEqual(self._get_error_code(body), 'AccessDenied')
|
|
|
|
def test_bucket_acl_GET_with_read_acp_permission(self):
|
|
status, headers, body = self._test_bucket_acl_GET('test:read_acp')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_bucket_acl_GET_with_fullcontrol_permission(self):
|
|
status, headers, body = self._test_bucket_acl_GET('test:full_control')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_bucket_acl_GET_with_owner_permission(self):
|
|
status, headers, body = self._test_bucket_acl_GET('test:tester')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def _test_bucket_acl_PUT(self, account, permission='FULL_CONTROL'):
|
|
acl = ACL(self.default_owner, [Grant(User(account), permission)])
|
|
req = Request.blank('/bucket?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS %s:hmac' % account,
|
|
'Date': self.get_date_header()},
|
|
body=tostring(acl.elem()))
|
|
|
|
return self.call_s3api(req)
|
|
|
|
def test_bucket_acl_PUT_without_permission(self):
|
|
status, headers, body = self._test_bucket_acl_PUT('test:other')
|
|
self.assertEqual(self._get_error_code(body), 'AccessDenied')
|
|
|
|
def test_bucket_acl_PUT_with_write_acp_permission(self):
|
|
status, headers, body = self._test_bucket_acl_PUT('test:write_acp')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_bucket_acl_PUT_with_fullcontrol_permission(self):
|
|
status, headers, body = self._test_bucket_acl_PUT('test:full_control')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_bucket_acl_PUT_with_owner_permission(self):
|
|
status, headers, body = self._test_bucket_acl_PUT('test:tester')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def _test_object_acl_GET(self, account):
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'GET'},
|
|
headers={'Authorization': 'AWS %s:hmac' % account,
|
|
'Date': self.get_date_header()})
|
|
return self.call_s3api(req)
|
|
|
|
def test_object_acl_GET_without_permission(self):
|
|
status, headers, body = self._test_object_acl_GET('test:other')
|
|
self.assertEqual(self._get_error_code(body), 'AccessDenied')
|
|
|
|
def test_object_acl_GET_with_read_acp_permission(self):
|
|
status, headers, body = self._test_object_acl_GET('test:read_acp')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_object_acl_GET_with_fullcontrol_permission(self):
|
|
status, headers, body = self._test_object_acl_GET('test:full_control')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_object_acl_GET_with_owner_permission(self):
|
|
status, headers, body = self._test_object_acl_GET('test:tester')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def _test_object_acl_PUT(self, account, permission='FULL_CONTROL'):
|
|
acl = ACL(self.default_owner, [Grant(User(account), permission)])
|
|
req = Request.blank('/bucket/object?acl',
|
|
environ={'REQUEST_METHOD': 'PUT'},
|
|
headers={'Authorization': 'AWS %s:hmac' % account,
|
|
'Date': self.get_date_header()},
|
|
body=tostring(acl.elem()))
|
|
|
|
return self.call_s3api(req)
|
|
|
|
def test_object_acl_PUT_without_permission(self):
|
|
status, headers, body = self._test_object_acl_PUT('test:other')
|
|
self.assertEqual(self._get_error_code(body), 'AccessDenied')
|
|
|
|
def test_object_acl_PUT_with_write_acp_permission(self):
|
|
status, headers, body = self._test_object_acl_PUT('test:write_acp')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_object_acl_PUT_with_fullcontrol_permission(self):
|
|
status, headers, body = self._test_object_acl_PUT('test:full_control')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_object_acl_PUT_with_owner_permission(self):
|
|
status, headers, body = self._test_object_acl_PUT('test:tester')
|
|
self.assertEqual(status.split()[0], '200')
|
|
|
|
def test_s3acl_decorator(self):
|
|
@s3acl
|
|
def non_class_s3acl_error():
|
|
raise TypeError()
|
|
|
|
class FakeClass(object):
|
|
def __init__(self):
|
|
self.s3api = MagicMock()
|
|
self.swift = FakeSwift()
|
|
|
|
@s3acl
|
|
def s3acl_error(self):
|
|
raise TypeError()
|
|
|
|
@s3acl
|
|
def s3acl_assert_fail(self):
|
|
assert False
|
|
|
|
@s3acl(s3acl_only=True)
|
|
def s3acl_s3only_error(self):
|
|
if self.s3api.conf.s3_acl:
|
|
raise TypeError()
|
|
|
|
@s3acl(s3acl_only=True)
|
|
def s3acl_s3only_no_error(self):
|
|
if not self.s3api.conf.s3_acl:
|
|
raise TypeError()
|
|
|
|
fake_class = FakeClass()
|
|
|
|
self.assertRaises(NotMethodException, non_class_s3acl_error)
|
|
self.assertRaises(TypeError, fake_class.s3acl_error)
|
|
self.assertRaises(AssertionError, fake_class.s3acl_assert_fail)
|
|
self.assertRaises(TypeError, fake_class.s3acl_s3only_error)
|
|
self.assertIsNone(fake_class.s3acl_s3only_no_error())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|