"Fossies" - the Fresh Open Source Software Archive

Member "swift-2.21.0/test/unit/common/middleware/s3api/test_service.py" (25 Mar 2019, 9281 Bytes) of package /linux/misc/openstack/swift-2.21.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_service.py": 2.19.1_vs_2.21.0.

    1 # Copyright (c) 2014 OpenStack Foundation
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License");
    4 # you may not use this file except in compliance with the License.
    5 # You may obtain a copy of the License at
    6 #
    7 #    http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS,
   11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
   12 # implied.
   13 # See the License for the specific language governing permissions and
   14 # limitations under the License.
   15 
   16 import unittest
   17 
   18 from swift.common import swob
   19 from swift.common.swob import Request
   20 from swift.common.utils import json
   21 
   22 from test.unit.common.middleware.s3api.test_s3_acl import s3acl
   23 from test.unit.common.middleware.s3api import S3ApiTestCase
   24 from swift.common.middleware.s3api.etree import fromstring
   25 from swift.common.middleware.s3api.subresource import ACL, Owner, encode_acl
   26 
   27 
   28 def create_bucket_list_json(buckets):
   29     """
   30     Create a json from bucket list
   31     :param buckets: a list of tuples (or lists) consist of elements orderd as
   32                     name, count, bytes
   33     """
   34     bucket_list = [{'name': item[0], 'count': item[1], 'bytes': item[2]}
   35                    for item in buckets]
   36     return json.dumps(bucket_list)
   37 
   38 
   39 class TestS3ApiService(S3ApiTestCase):
   40     def setup_buckets(self):
   41         self.buckets = (('apple', 1, 200), ('orange', 3, 430))
   42         bucket_list = create_bucket_list_json(self.buckets)
   43         self.swift.register('GET', '/v1/AUTH_test', swob.HTTPOk, {},
   44                             bucket_list)
   45 
   46     def setUp(self):
   47         super(TestS3ApiService, self).setUp()
   48 
   49         self.setup_buckets()
   50 
   51     def test_service_GET_error(self):
   52         code = self._test_method_error(
   53             'GET', '', swob.HTTPUnauthorized, expected_xml_tags=(
   54                 'Code', 'Message', 'AWSAccessKeyId', 'StringToSign',
   55                 'StringToSignBytes', 'SignatureProvided'))
   56         self.assertEqual(code, 'SignatureDoesNotMatch')
   57         code = self._test_method_error('GET', '', swob.HTTPForbidden)
   58         self.assertEqual(code, 'AccessDenied')
   59         code = self._test_method_error('GET', '', swob.HTTPServerError)
   60         self.assertEqual(code, 'InternalError')
   61 
   62     @s3acl
   63     def test_service_GET(self):
   64         req = Request.blank('/',
   65                             environ={'REQUEST_METHOD': 'GET'},
   66                             headers={'Authorization': 'AWS test:tester:hmac',
   67                                      'Date': self.get_date_header()})
   68         status, headers, body = self.call_s3api(req)
   69         self.assertEqual(status.split()[0], '200')
   70 
   71         elem = fromstring(body, 'ListAllMyBucketsResult')
   72 
   73         all_buckets = elem.find('./Buckets')
   74         buckets = all_buckets.iterchildren('Bucket')
   75         listing = list(list(buckets)[0])
   76         self.assertEqual(len(listing), 2)
   77 
   78         names = []
   79         for b in all_buckets.iterchildren('Bucket'):
   80             names.append(b.find('./Name').text)
   81 
   82         self.assertEqual(len(names), len(self.buckets))
   83         for i in self.buckets:
   84             self.assertTrue(i[0] in names)
   85 
   86     @s3acl
   87     def test_service_GET_subresource(self):
   88         req = Request.blank('/?acl',
   89                             environ={'REQUEST_METHOD': 'GET'},
   90                             headers={'Authorization': 'AWS test:tester:hmac',
   91                                      'Date': self.get_date_header()})
   92         status, headers, body = self.call_s3api(req)
   93         self.assertEqual(status.split()[0], '200')
   94 
   95         elem = fromstring(body, 'ListAllMyBucketsResult')
   96 
   97         all_buckets = elem.find('./Buckets')
   98         buckets = all_buckets.iterchildren('Bucket')
   99         listing = list(list(buckets)[0])
  100         self.assertEqual(len(listing), 2)
  101 
  102         names = []
  103         for b in all_buckets.iterchildren('Bucket'):
  104             names.append(b.find('./Name').text)
  105 
  106         self.assertEqual(len(names), len(self.buckets))
  107         for i in self.buckets:
  108             self.assertTrue(i[0] in names)
  109 
  110     def test_service_GET_with_blind_resource(self):
  111         buckets = (('apple', 1, 200), ('orange', 3, 430),
  112                    ('apple+segment', 1, 200))
  113         expected = buckets[:-1]
  114         bucket_list = create_bucket_list_json(buckets)
  115         self.swift.register('GET', '/v1/AUTH_test', swob.HTTPOk, {},
  116                             bucket_list)
  117 
  118         req = Request.blank('/',
  119                             environ={'REQUEST_METHOD': 'GET'},
  120                             headers={'Authorization': 'AWS test:tester:hmac',
  121                                      'Date': self.get_date_header()})
  122 
  123         status, headers, body = self.call_s3api(req)
  124         self.assertEqual(status.split()[0], '200')
  125 
  126         elem = fromstring(body, 'ListAllMyBucketsResult')
  127         all_buckets = elem.find('./Buckets')
  128         buckets = all_buckets.iterchildren('Bucket')
  129         listing = list(list(buckets)[0])
  130         self.assertEqual(len(listing), 2)
  131 
  132         names = []
  133         for b in all_buckets.iterchildren('Bucket'):
  134             names.append(b.find('./Name').text)
  135 
  136         self.assertEqual(len(names), len(expected))
  137         for i in expected:
  138             self.assertTrue(i[0] in names)
  139 
  140     def _test_service_GET_for_check_bucket_owner(self, buckets):
  141         self.s3api.conf.check_bucket_owner = True
  142         bucket_list = create_bucket_list_json(buckets)
  143         self.swift.register('GET', '/v1/AUTH_test', swob.HTTPOk, {},
  144                             bucket_list)
  145 
  146         req = Request.blank('/',
  147                             environ={'REQUEST_METHOD': 'GET'},
  148                             headers={'Authorization': 'AWS test:tester:hmac',
  149                                      'Date': self.get_date_header()})
  150         return self.call_s3api(req)
  151 
  152     @s3acl(s3acl_only=True)
  153     def test_service_GET_without_bucket(self):
  154         bucket_list = []
  155         for var in range(0, 10):
  156             bucket = 'bucket%s' % var
  157             self.swift.register('HEAD', '/v1/AUTH_test/%s' % bucket,
  158                                 swob.HTTPNotFound, {}, None)
  159             bucket_list.append((bucket, var, 300 + var))
  160 
  161         status, headers, body = \
  162             self._test_service_GET_for_check_bucket_owner(bucket_list)
  163         self.assertEqual(status.split()[0], '200')
  164 
  165         elem = fromstring(body, 'ListAllMyBucketsResult')
  166 
  167         resp_buckets = elem.find('./Buckets')
  168         buckets = resp_buckets.iterchildren('Bucket')
  169         self.assertEqual(len(list(buckets)), 0)
  170 
  171     @s3acl(s3acl_only=True)
  172     def test_service_GET_without_owner_bucket(self):
  173         bucket_list = []
  174         for var in range(0, 10):
  175             user_id = 'test:other'
  176             bucket = 'bucket%s' % var
  177             owner = Owner(user_id, user_id)
  178             headers = encode_acl('container', ACL(owner, []))
  179             self.swift.register('HEAD', '/v1/AUTH_test/%s' % bucket,
  180                                 swob.HTTPNoContent, headers, None)
  181             bucket_list.append((bucket, var, 300 + var))
  182 
  183         status, headers, body = \
  184             self._test_service_GET_for_check_bucket_owner(bucket_list)
  185         self.assertEqual(status.split()[0], '200')
  186 
  187         elem = fromstring(body, 'ListAllMyBucketsResult')
  188 
  189         resp_buckets = elem.find('./Buckets')
  190         buckets = resp_buckets.iterchildren('Bucket')
  191         self.assertEqual(len(list(buckets)), 0)
  192 
  193     @s3acl(s3acl_only=True)
  194     def test_service_GET_bucket_list(self):
  195         bucket_list = []
  196         for var in range(0, 10):
  197             if var % 3 == 0:
  198                 user_id = 'test:tester'
  199             else:
  200                 user_id = 'test:other'
  201             bucket = 'bucket%s' % var
  202             owner = Owner(user_id, user_id)
  203             headers = encode_acl('container', ACL(owner, []))
  204             # set register to get owner of buckets
  205             if var % 3 == 2:
  206                 self.swift.register('HEAD', '/v1/AUTH_test/%s' % bucket,
  207                                     swob.HTTPNotFound, {}, None)
  208             else:
  209                 self.swift.register('HEAD', '/v1/AUTH_test/%s' % bucket,
  210                                     swob.HTTPNoContent, headers, None)
  211             bucket_list.append((bucket, var, 300 + var))
  212 
  213         status, headers, body = \
  214             self._test_service_GET_for_check_bucket_owner(bucket_list)
  215         self.assertEqual(status.split()[0], '200')
  216 
  217         elem = fromstring(body, 'ListAllMyBucketsResult')
  218         resp_buckets = elem.find('./Buckets')
  219         buckets = resp_buckets.iterchildren('Bucket')
  220         listing = list(list(buckets)[0])
  221         self.assertEqual(len(listing), 2)
  222 
  223         names = []
  224         for b in resp_buckets.iterchildren('Bucket'):
  225             names.append(b.find('./Name').text)
  226 
  227         # Check whether getting bucket only locate in multiples of 3 in
  228         # bucket_list which mean requested user is owner.
  229         expected_buckets = [b for i, b in enumerate(bucket_list)
  230                             if i % 3 == 0]
  231         self.assertEqual(len(names), len(expected_buckets))
  232         for i in expected_buckets:
  233             self.assertTrue(i[0] in names)
  234         self.assertEqual(len(self.swift.calls_with_headers), 11)
  235 
  236 if __name__ == '__main__':
  237     unittest.main()