"Fossies" - the Fresh Open Source Software Archive

Member "swift-2.21.0/test/unit/common/middleware/s3api/test_obj.py" (25 Mar 2019, 55623 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_obj.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 from datetime import datetime
   18 import hashlib
   19 import os
   20 from os.path import join
   21 import time
   22 from mock import patch
   23 
   24 from swift.common import swob
   25 from swift.common.swob import Request
   26 
   27 from test.unit.common.middleware.s3api import S3ApiTestCase
   28 from test.unit.common.middleware.s3api.test_s3_acl import s3acl
   29 from swift.common.middleware.s3api.subresource import ACL, User, encode_acl, \
   30     Owner, Grant
   31 from swift.common.middleware.s3api.etree import fromstring
   32 from swift.common.middleware.s3api.utils import mktime, S3Timestamp
   33 
   34 
   35 class TestS3ApiObj(S3ApiTestCase):
   36 
   37     def setUp(self):
   38         super(TestS3ApiObj, self).setUp()
   39 
   40         self.object_body = 'hello'
   41         self.etag = hashlib.md5(self.object_body).hexdigest()
   42         self.last_modified = 'Fri, 01 Apr 2014 12:00:00 GMT'
   43 
   44         self.response_headers = {'Content-Type': 'text/html',
   45                                  'Content-Length': len(self.object_body),
   46                                  'Content-Disposition': 'inline',
   47                                  'Content-Language': 'en',
   48                                  'x-object-meta-test': 'swift',
   49                                  'etag': self.etag,
   50                                  'last-modified': self.last_modified,
   51                                  'expires': 'Mon, 21 Sep 2015 12:00:00 GMT',
   52                                  'x-robots-tag': 'nofollow',
   53                                  'cache-control': 'private'}
   54 
   55         self.swift.register('GET', '/v1/AUTH_test/bucket/object',
   56                             swob.HTTPOk, self.response_headers,
   57                             self.object_body)
   58         self.swift.register('PUT', '/v1/AUTH_test/bucket/object',
   59                             swob.HTTPCreated,
   60                             {'etag': self.etag,
   61                              'last-modified': self.last_modified,
   62                              'x-object-meta-something': 'oh hai'},
   63                             None)
   64 
   65     def _test_object_GETorHEAD(self, method):
   66         req = Request.blank('/bucket/object',
   67                             environ={'REQUEST_METHOD': method},
   68                             headers={'Authorization': 'AWS test:tester:hmac',
   69                                      'Date': self.get_date_header()})
   70         status, headers, body = self.call_s3api(req)
   71         self.assertEqual(status.split()[0], '200')
   72 
   73         unexpected_headers = []
   74         for key, val in self.response_headers.items():
   75             if key in ('Content-Length', 'Content-Type', 'content-encoding',
   76                        'last-modified', 'cache-control', 'Content-Disposition',
   77                        'Content-Language', 'expires', 'x-robots-tag'):
   78                 self.assertIn(key, headers)
   79                 self.assertEqual(headers[key], str(val))
   80 
   81             elif key == 'etag':
   82                 self.assertEqual(headers[key], '"%s"' % val)
   83 
   84             elif key.startswith('x-object-meta-'):
   85                 self.assertIn('x-amz-meta-' + key[14:], headers)
   86                 self.assertEqual(headers['x-amz-meta-' + key[14:]], val)
   87 
   88             else:
   89                 unexpected_headers.append((key, val))
   90 
   91         if unexpected_headers:
   92                 self.fail('unexpected headers: %r' % unexpected_headers)
   93 
   94         self.assertEqual(headers['etag'],
   95                          '"%s"' % self.response_headers['etag'])
   96 
   97         if method == 'GET':
   98             self.assertEqual(body, self.object_body)
   99 
  100     @s3acl
  101     def test_object_HEAD_error(self):
  102         # HEAD does not return the body even an error response in the
  103         # specifications of the REST API.
  104         # So, check the response code for error test of HEAD.
  105         req = Request.blank('/bucket/object',
  106                             environ={'REQUEST_METHOD': 'HEAD'},
  107                             headers={'Authorization': 'AWS test:tester:hmac',
  108                                      'Date': self.get_date_header()})
  109         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  110                             swob.HTTPUnauthorized, {}, None)
  111         status, headers, body = self.call_s3api(req)
  112         self.assertEqual(status.split()[0], '403')
  113         self.assertEqual(body, '')  # sanity
  114 
  115         req = Request.blank('/bucket/object',
  116                             environ={'REQUEST_METHOD': 'HEAD'},
  117                             headers={'Authorization': 'AWS test:tester:hmac',
  118                                      'Date': self.get_date_header()})
  119         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  120                             swob.HTTPForbidden, {}, None)
  121         status, headers, body = self.call_s3api(req)
  122         self.assertEqual(status.split()[0], '403')
  123         self.assertEqual(body, '')  # sanity
  124 
  125         req = Request.blank('/bucket/object',
  126                             environ={'REQUEST_METHOD': 'HEAD'},
  127                             headers={'Authorization': 'AWS test:tester:hmac',
  128                                      'Date': self.get_date_header()})
  129         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  130                             swob.HTTPNotFound, {}, None)
  131         status, headers, body = self.call_s3api(req)
  132         self.assertEqual(status.split()[0], '404')
  133         self.assertEqual(body, '')  # sanity
  134 
  135         req = Request.blank('/bucket/object',
  136                             environ={'REQUEST_METHOD': 'HEAD'},
  137                             headers={'Authorization': 'AWS test:tester:hmac',
  138                                      'Date': self.get_date_header()})
  139         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  140                             swob.HTTPPreconditionFailed, {}, None)
  141         status, headers, body = self.call_s3api(req)
  142         self.assertEqual(status.split()[0], '412')
  143         self.assertEqual(body, '')  # sanity
  144 
  145         req = Request.blank('/bucket/object',
  146                             environ={'REQUEST_METHOD': 'HEAD'},
  147                             headers={'Authorization': 'AWS test:tester:hmac',
  148                                      'Date': self.get_date_header()})
  149         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  150                             swob.HTTPServerError, {}, None)
  151         status, headers, body = self.call_s3api(req)
  152         self.assertEqual(status.split()[0], '500')
  153         self.assertEqual(body, '')  # sanity
  154 
  155         req = Request.blank('/bucket/object',
  156                             environ={'REQUEST_METHOD': 'HEAD'},
  157                             headers={'Authorization': 'AWS test:tester:hmac',
  158                                      'Date': self.get_date_header()})
  159         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  160                             swob.HTTPServiceUnavailable, {}, None)
  161         status, headers, body = self.call_s3api(req)
  162         self.assertEqual(status.split()[0], '500')
  163         self.assertEqual(body, '')  # sanity
  164 
  165     def test_object_HEAD(self):
  166         self._test_object_GETorHEAD('HEAD')
  167 
  168     def _test_object_HEAD_Range(self, range_value):
  169         req = Request.blank('/bucket/object',
  170                             environ={'REQUEST_METHOD': 'HEAD'},
  171                             headers={'Authorization': 'AWS test:tester:hmac',
  172                                      'Range': range_value,
  173                                      'Date': self.get_date_header()})
  174         return self.call_s3api(req)
  175 
  176     @s3acl
  177     def test_object_HEAD_Range_with_invalid_value(self):
  178         range_value = ''
  179         status, headers, body = self._test_object_HEAD_Range(range_value)
  180         self.assertEqual(status.split()[0], '200')
  181         self.assertTrue('content-length' in headers)
  182         self.assertEqual(headers['content-length'], '5')
  183         self.assertTrue('content-range' not in headers)
  184 
  185         range_value = 'hoge'
  186         status, headers, body = self._test_object_HEAD_Range(range_value)
  187         self.assertEqual(status.split()[0], '200')
  188         self.assertTrue('content-length' in headers)
  189         self.assertEqual(headers['content-length'], '5')
  190         self.assertTrue('content-range' not in headers)
  191 
  192         range_value = 'bytes='
  193         status, headers, body = self._test_object_HEAD_Range(range_value)
  194         self.assertEqual(status.split()[0], '200')
  195         self.assertTrue('content-length' in headers)
  196         self.assertEqual(headers['content-length'], '5')
  197         self.assertTrue('content-range' not in headers)
  198 
  199         range_value = 'bytes=1'
  200         status, headers, body = self._test_object_HEAD_Range(range_value)
  201         self.assertEqual(status.split()[0], '200')
  202         self.assertTrue('content-length' in headers)
  203         self.assertEqual(headers['content-length'], '5')
  204         self.assertTrue('content-range' not in headers)
  205 
  206         range_value = 'bytes=5-1'
  207         status, headers, body = self._test_object_HEAD_Range(range_value)
  208         self.assertEqual(status.split()[0], '200')
  209         self.assertTrue('content-length' in headers)
  210         self.assertEqual(headers['content-length'], '5')
  211         self.assertTrue('content-range' not in headers)
  212 
  213         range_value = 'bytes=5-10'
  214         status, headers, body = self._test_object_HEAD_Range(range_value)
  215         self.assertEqual(status.split()[0], '416')
  216 
  217     @s3acl
  218     def test_object_HEAD_Range(self):
  219         # update response headers
  220         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  221                             swob.HTTPOk, self.response_headers,
  222                             self.object_body)
  223         range_value = 'bytes=0-3'
  224         status, headers, body = self._test_object_HEAD_Range(range_value)
  225         self.assertEqual(status.split()[0], '206')
  226         self.assertTrue('content-length' in headers)
  227         self.assertEqual(headers['content-length'], '4')
  228         self.assertTrue('content-range' in headers)
  229         self.assertTrue(headers['content-range'].startswith('bytes 0-3'))
  230         self.assertTrue('x-amz-meta-test' in headers)
  231         self.assertEqual('swift', headers['x-amz-meta-test'])
  232 
  233         range_value = 'bytes=3-3'
  234         status, headers, body = self._test_object_HEAD_Range(range_value)
  235         self.assertEqual(status.split()[0], '206')
  236         self.assertTrue('content-length' in headers)
  237         self.assertEqual(headers['content-length'], '1')
  238         self.assertTrue('content-range' in headers)
  239         self.assertTrue(headers['content-range'].startswith('bytes 3-3'))
  240         self.assertTrue('x-amz-meta-test' in headers)
  241         self.assertEqual('swift', headers['x-amz-meta-test'])
  242 
  243         range_value = 'bytes=1-'
  244         status, headers, body = self._test_object_HEAD_Range(range_value)
  245         self.assertEqual(status.split()[0], '206')
  246         self.assertTrue('content-length' in headers)
  247         self.assertEqual(headers['content-length'], '4')
  248         self.assertTrue('content-range' in headers)
  249         self.assertTrue(headers['content-range'].startswith('bytes 1-4'))
  250         self.assertTrue('x-amz-meta-test' in headers)
  251         self.assertEqual('swift', headers['x-amz-meta-test'])
  252 
  253         range_value = 'bytes=-3'
  254         status, headers, body = self._test_object_HEAD_Range(range_value)
  255         self.assertEqual(status.split()[0], '206')
  256         self.assertTrue('content-length' in headers)
  257         self.assertEqual(headers['content-length'], '3')
  258         self.assertTrue('content-range' in headers)
  259         self.assertTrue(headers['content-range'].startswith('bytes 2-4'))
  260         self.assertTrue('x-amz-meta-test' in headers)
  261         self.assertEqual('swift', headers['x-amz-meta-test'])
  262 
  263     @s3acl
  264     def test_object_GET_error(self):
  265         code = self._test_method_error('GET', '/bucket/object',
  266                                        swob.HTTPUnauthorized)
  267         self.assertEqual(code, 'SignatureDoesNotMatch')
  268         code = self._test_method_error('GET', '/bucket/object',
  269                                        swob.HTTPForbidden)
  270         self.assertEqual(code, 'AccessDenied')
  271         code = self._test_method_error('GET', '/bucket/object',
  272                                        swob.HTTPNotFound)
  273         self.assertEqual(code, 'NoSuchKey')
  274         code = self._test_method_error('GET', '/bucket/object',
  275                                        swob.HTTPServerError)
  276         self.assertEqual(code, 'InternalError')
  277         code = self._test_method_error('GET', '/bucket/object',
  278                                        swob.HTTPPreconditionFailed)
  279         self.assertEqual(code, 'PreconditionFailed')
  280         code = self._test_method_error('GET', '/bucket/object',
  281                                        swob.HTTPServiceUnavailable)
  282         self.assertEqual(code, 'InternalError')
  283 
  284     @s3acl
  285     def test_object_GET(self):
  286         self._test_object_GETorHEAD('GET')
  287 
  288     @s3acl(s3acl_only=True)
  289     def test_object_GET_with_s3acl_and_unknown_user(self):
  290         self.swift.remote_user = None
  291         req = Request.blank('/bucket/object',
  292                             environ={'REQUEST_METHOD': 'GET'},
  293                             headers={'Authorization': 'AWS test:tester:hmac',
  294                                      'Date': self.get_date_header()})
  295         status, headers, body = self.call_s3api(req)
  296         self.assertEqual(status, '403 Forbidden')
  297         self.assertEqual(self._get_error_code(body), 'SignatureDoesNotMatch')
  298 
  299     @s3acl(s3acl_only=True)
  300     def test_object_GET_with_s3acl_and_keystone(self):
  301         # for passing keystone authentication root
  302         orig_auth = self.swift._fake_auth_middleware
  303         calls = []
  304 
  305         def wrapped_auth(env):
  306             calls.append((env['REQUEST_METHOD'], 's3api.auth_details' in env))
  307             orig_auth(env)
  308 
  309         with patch.object(self.swift, '_fake_auth_middleware', wrapped_auth):
  310             self._test_object_GETorHEAD('GET')
  311         self.assertEqual(calls, [
  312             ('TEST', True),
  313             ('HEAD', False),
  314             ('GET', False),
  315         ])
  316 
  317     @s3acl
  318     def test_object_GET_Range(self):
  319         req = Request.blank('/bucket/object',
  320                             environ={'REQUEST_METHOD': 'GET'},
  321                             headers={'Authorization': 'AWS test:tester:hmac',
  322                                      'Range': 'bytes=0-3',
  323                                      'Date': self.get_date_header()})
  324         status, headers, body = self.call_s3api(req)
  325         self.assertEqual(status.split()[0], '206')
  326 
  327         self.assertTrue('content-range' in headers)
  328         self.assertTrue(headers['content-range'].startswith('bytes 0-3'))
  329 
  330     @s3acl
  331     def test_object_GET_Range_error(self):
  332         code = self._test_method_error('GET', '/bucket/object',
  333                                        swob.HTTPRequestedRangeNotSatisfiable)
  334         self.assertEqual(code, 'InvalidRange')
  335 
  336     @s3acl
  337     def test_object_GET_Response(self):
  338         req = Request.blank('/bucket/object',
  339                             environ={'REQUEST_METHOD': 'GET',
  340                                      'QUERY_STRING':
  341                                      'response-content-type=%s&'
  342                                      'response-content-language=%s&'
  343                                      'response-expires=%s&'
  344                                      'response-cache-control=%s&'
  345                                      'response-content-disposition=%s&'
  346                                      'response-content-encoding=%s&'
  347                                      % ('text/plain', 'en',
  348                                         'Fri, 01 Apr 2014 12:00:00 GMT',
  349                                         'no-cache',
  350                                         'attachment',
  351                                         'gzip')},
  352                             headers={'Authorization': 'AWS test:tester:hmac',
  353                                      'Date': self.get_date_header()})
  354         status, headers, body = self.call_s3api(req)
  355         self.assertEqual(status.split()[0], '200')
  356 
  357         self.assertTrue('content-type' in headers)
  358         self.assertEqual(headers['content-type'], 'text/plain')
  359         self.assertTrue('content-language' in headers)
  360         self.assertEqual(headers['content-language'], 'en')
  361         self.assertTrue('expires' in headers)
  362         self.assertEqual(headers['expires'], 'Fri, 01 Apr 2014 12:00:00 GMT')
  363         self.assertTrue('cache-control' in headers)
  364         self.assertEqual(headers['cache-control'], 'no-cache')
  365         self.assertTrue('content-disposition' in headers)
  366         self.assertEqual(headers['content-disposition'],
  367                          'attachment')
  368         self.assertTrue('content-encoding' in headers)
  369         self.assertEqual(headers['content-encoding'], 'gzip')
  370 
  371     @s3acl
  372     def test_object_PUT_error(self):
  373         code = self._test_method_error('PUT', '/bucket/object',
  374                                        swob.HTTPUnauthorized)
  375         self.assertEqual(code, 'SignatureDoesNotMatch')
  376         code = self._test_method_error('PUT', '/bucket/object',
  377                                        swob.HTTPForbidden)
  378         self.assertEqual(code, 'AccessDenied')
  379         code = self._test_method_error('PUT', '/bucket/object',
  380                                        swob.HTTPNotFound)
  381         self.assertEqual(code, 'NoSuchBucket')
  382         code = self._test_method_error('PUT', '/bucket/object',
  383                                        swob.HTTPRequestEntityTooLarge)
  384         self.assertEqual(code, 'EntityTooLarge')
  385         code = self._test_method_error('PUT', '/bucket/object',
  386                                        swob.HTTPServerError)
  387         self.assertEqual(code, 'InternalError')
  388         code = self._test_method_error('PUT', '/bucket/object',
  389                                        swob.HTTPUnprocessableEntity)
  390         self.assertEqual(code, 'BadDigest')
  391         code = self._test_method_error('PUT', '/bucket/object',
  392                                        swob.HTTPLengthRequired)
  393         self.assertEqual(code, 'MissingContentLength')
  394         code = self._test_method_error('PUT', '/bucket/object',
  395                                        swob.HTTPPreconditionFailed)
  396         self.assertEqual(code, 'InternalError')
  397         code = self._test_method_error('PUT', '/bucket/object',
  398                                        swob.HTTPServiceUnavailable)
  399         self.assertEqual(code, 'InternalError')
  400         code = self._test_method_error('PUT', '/bucket/object',
  401                                        swob.HTTPCreated,
  402                                        {'X-Amz-Copy-Source': ''})
  403         self.assertEqual(code, 'InvalidArgument')
  404         code = self._test_method_error('PUT', '/bucket/object',
  405                                        swob.HTTPCreated,
  406                                        {'X-Amz-Copy-Source': '/'})
  407         self.assertEqual(code, 'InvalidArgument')
  408         code = self._test_method_error('PUT', '/bucket/object',
  409                                        swob.HTTPCreated,
  410                                        {'X-Amz-Copy-Source': '/bucket'})
  411         self.assertEqual(code, 'InvalidArgument')
  412         code = self._test_method_error('PUT', '/bucket/object',
  413                                        swob.HTTPCreated,
  414                                        {'X-Amz-Copy-Source': '/bucket/'})
  415         self.assertEqual(code, 'InvalidArgument')
  416         code = self._test_method_error(
  417             'PUT', '/bucket/object',
  418             swob.HTTPCreated,
  419             {'X-Amz-Copy-Source': '/bucket/src_obj?foo=bar'})
  420         self.assertEqual(code, 'InvalidArgument')
  421         # adding other query paramerters will cause an error
  422         code = self._test_method_error(
  423             'PUT', '/bucket/object',
  424             swob.HTTPCreated,
  425             {'X-Amz-Copy-Source': '/bucket/src_obj?versionId=foo&bar=baz'})
  426         self.assertEqual(code, 'InvalidArgument')
  427         # ...even versionId appears in the last
  428         code = self._test_method_error(
  429             'PUT', '/bucket/object',
  430             swob.HTTPCreated,
  431             {'X-Amz-Copy-Source': '/bucket/src_obj?bar=baz&versionId=foo'})
  432         self.assertEqual(code, 'InvalidArgument')
  433         code = self._test_method_error(
  434             'PUT', '/bucket/object',
  435             swob.HTTPCreated,
  436             {'X-Amz-Copy-Source': '/bucket/src_obj?versionId=foo'})
  437         self.assertEqual(code, 'NotImplemented')
  438         code = self._test_method_error(
  439             'PUT', '/bucket/object',
  440             swob.HTTPCreated,
  441             {'X-Amz-Copy-Source': '/src_bucket/src_object',
  442              'X-Amz-Copy-Source-Range': 'bytes=0-0'})
  443         self.assertEqual(code, 'InvalidArgument')
  444         code = self._test_method_error('PUT', '/bucket/object',
  445                                        swob.HTTPRequestTimeout)
  446         self.assertEqual(code, 'RequestTimeout')
  447 
  448     @s3acl
  449     def test_object_PUT(self):
  450         etag = self.response_headers['etag']
  451         content_md5 = etag.decode('hex').encode('base64').strip()
  452 
  453         req = Request.blank(
  454             '/bucket/object',
  455             environ={'REQUEST_METHOD': 'PUT'},
  456             headers={'Authorization': 'AWS test:tester:hmac',
  457                      'x-amz-storage-class': 'STANDARD',
  458                      'Content-MD5': content_md5,
  459                      'Date': self.get_date_header()},
  460             body=self.object_body)
  461         req.date = datetime.now()
  462         req.content_type = 'text/plain'
  463         status, headers, body = self.call_s3api(req)
  464         self.assertEqual(status.split()[0], '200')
  465         # Check that s3api returns an etag header.
  466         self.assertEqual(headers['etag'], '"%s"' % etag)
  467 
  468         _, _, headers = self.swift.calls_with_headers[-1]
  469         # Check that s3api converts a Content-MD5 header into an etag.
  470         self.assertEqual(headers['etag'], etag)
  471 
  472     @s3acl
  473     def test_object_PUT_v4(self):
  474         body_sha = hashlib.sha256(self.object_body).hexdigest()
  475         req = Request.blank(
  476             '/bucket/object',
  477             environ={'REQUEST_METHOD': 'PUT'},
  478             headers={
  479                 'Authorization':
  480                     'AWS4-HMAC-SHA256 '
  481                     'Credential=test:tester/%s/us-east-1/s3/aws4_request, '
  482                     'SignedHeaders=host;x-amz-date, '
  483                     'Signature=hmac' % (
  484                         self.get_v4_amz_date_header().split('T', 1)[0]),
  485                 'x-amz-date': self.get_v4_amz_date_header(),
  486                 'x-amz-storage-class': 'STANDARD',
  487                 'x-amz-content-sha256': body_sha,
  488                 'Date': self.get_date_header()},
  489             body=self.object_body)
  490         req.date = datetime.now()
  491         req.content_type = 'text/plain'
  492         status, headers, body = self.call_s3api(req)
  493         self.assertEqual(status.split()[0], '200')
  494         # Check that s3api returns an etag header.
  495         self.assertEqual(headers['etag'],
  496                          '"%s"' % self.response_headers['etag'])
  497 
  498         _, _, headers = self.swift.calls_with_headers[-1]
  499         # No way to determine ETag to send
  500         self.assertNotIn('etag', headers)
  501 
  502     @s3acl
  503     def test_object_PUT_v4_bad_hash(self):
  504         req = Request.blank(
  505             '/bucket/object',
  506             environ={'REQUEST_METHOD': 'PUT'},
  507             headers={
  508                 'Authorization':
  509                     'AWS4-HMAC-SHA256 '
  510                     'Credential=test:tester/%s/us-east-1/s3/aws4_request, '
  511                     'SignedHeaders=host;x-amz-date, '
  512                     'Signature=hmac' % (
  513                         self.get_v4_amz_date_header().split('T', 1)[0]),
  514                 'x-amz-date': self.get_v4_amz_date_header(),
  515                 'x-amz-storage-class': 'STANDARD',
  516                 'x-amz-content-sha256': 'not the hash',
  517                 'Date': self.get_date_header()},
  518             body=self.object_body)
  519         req.date = datetime.now()
  520         req.content_type = 'text/plain'
  521         status, headers, body = self.call_s3api(req)
  522         self.assertEqual(status.split()[0], '400')
  523         print(body)
  524         self.assertEqual(self._get_error_code(body), 'BadDigest')
  525 
  526     def test_object_PUT_headers(self):
  527         content_md5 = self.etag.decode('hex').encode('base64').strip()
  528 
  529         self.swift.register('HEAD', '/v1/AUTH_test/some/source',
  530                             swob.HTTPOk, {'last-modified': self.last_modified},
  531                             None)
  532         req = Request.blank(
  533             '/bucket/object',
  534             environ={'REQUEST_METHOD': 'PUT'},
  535             headers={'Authorization': 'AWS test:tester:hmac',
  536                      'X-Amz-Storage-Class': 'STANDARD',
  537                      'X-Amz-Meta-Something': 'oh hai',
  538                      'X-Amz-Meta-Unreadable-Prefix': '\x04w',
  539                      'X-Amz-Meta-Unreadable-Suffix': 'h\x04',
  540                      'X-Amz-Meta-Lots-Of-Unprintable': 5 * '\x04',
  541                      'X-Amz-Copy-Source': '/some/source',
  542                      'Content-MD5': content_md5,
  543                      'Date': self.get_date_header()})
  544         req.date = datetime.now()
  545         req.content_type = 'text/plain'
  546         status, headers, body = self.call_s3api(req)
  547         # Check that s3api does not return an etag header,
  548         # specified copy source.
  549         self.assertTrue(headers.get('etag') is None)
  550         # Check that s3api does not return custom metadata in response
  551         self.assertTrue(headers.get('x-amz-meta-something') is None)
  552 
  553         _, _, headers = self.swift.calls_with_headers[-1]
  554         # Check that s3api converts a Content-MD5 header into an etag.
  555         self.assertEqual(headers['ETag'], self.etag)
  556         # Check that metadata is omited if no directive is specified
  557         self.assertIsNone(headers.get('X-Object-Meta-Something'))
  558         self.assertIsNone(headers.get('X-Object-Meta-Unreadable-Prefix'))
  559         self.assertIsNone(headers.get('X-Object-Meta-Unreadable-Suffix'))
  560         self.assertIsNone(headers.get('X-Object-Meta-Lots-Of-Unprintable'))
  561 
  562         self.assertEqual(headers['X-Copy-From'], '/some/source')
  563         self.assertEqual(headers['Content-Length'], '0')
  564 
  565     def _test_object_PUT_copy(self, head_resp, put_header=None,
  566                               src_path='/some/source', timestamp=None):
  567         account = 'test:tester'
  568         grants = [Grant(User(account), 'FULL_CONTROL')]
  569         head_headers = \
  570             encode_acl('object',
  571                        ACL(Owner(account, account), grants))
  572         head_headers.update({'last-modified': self.last_modified})
  573         self.swift.register('HEAD', '/v1/AUTH_test/some/source',
  574                             head_resp, head_headers, None)
  575         put_header = put_header or {}
  576         return self._call_object_copy(src_path, put_header, timestamp)
  577 
  578     def _test_object_PUT_copy_self(self, head_resp,
  579                                    put_header=None, timestamp=None):
  580         account = 'test:tester'
  581         grants = [Grant(User(account), 'FULL_CONTROL')]
  582         head_headers = \
  583             encode_acl('object',
  584                        ACL(Owner(account, account), grants))
  585         head_headers.update({'last-modified': self.last_modified})
  586         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  587                             head_resp, head_headers, None)
  588         put_header = put_header or {}
  589         return self._call_object_copy('/bucket/object', put_header, timestamp)
  590 
  591     def _call_object_copy(self, src_path, put_header, timestamp=None):
  592         put_headers = {'Authorization': 'AWS test:tester:hmac',
  593                        'X-Amz-Copy-Source': src_path,
  594                        'Date': self.get_date_header()}
  595         put_headers.update(put_header)
  596 
  597         req = Request.blank('/bucket/object',
  598                             environ={'REQUEST_METHOD': 'PUT'},
  599                             headers=put_headers)
  600 
  601         req.date = datetime.now()
  602         req.content_type = 'text/plain'
  603         timestamp = timestamp or time.time()
  604         with patch('swift.common.middleware.s3api.utils.time.time',
  605                    return_value=timestamp):
  606             return self.call_s3api(req)
  607 
  608     @s3acl
  609     def test_object_PUT_copy(self):
  610         def do_test(src_path=None):
  611             date_header = self.get_date_header()
  612             timestamp = mktime(date_header)
  613             allowed_last_modified = [S3Timestamp(timestamp).s3xmlformat]
  614             status, headers, body = self._test_object_PUT_copy(
  615                 swob.HTTPOk, put_header={'Date': date_header},
  616                 timestamp=timestamp, src_path=src_path)
  617             # may have gotten unlucky and had the clock roll over
  618             date_header = self.get_date_header()
  619             timestamp = mktime(date_header)
  620             allowed_last_modified.append(S3Timestamp(timestamp).s3xmlformat)
  621 
  622             self.assertEqual(status.split()[0], '200')
  623             self.assertEqual(headers['Content-Type'], 'application/xml')
  624 
  625             self.assertTrue(headers.get('etag') is None)
  626             self.assertTrue(headers.get('x-amz-meta-something') is None)
  627             elem = fromstring(body, 'CopyObjectResult')
  628             self.assertIn(elem.find('LastModified').text,
  629                           allowed_last_modified)
  630             self.assertEqual(elem.find('ETag').text, '"%s"' % self.etag)
  631 
  632             _, _, headers = self.swift.calls_with_headers[-1]
  633             self.assertEqual(headers['X-Copy-From'], '/some/source')
  634             self.assertTrue(headers.get('X-Fresh-Metadata') is None)
  635             self.assertEqual(headers['Content-Length'], '0')
  636 
  637         do_test('/some/source')
  638         do_test('/some/source?')
  639         do_test('/some/source?versionId=null')
  640         # Some clients (like Boto) don't include the leading slash;
  641         # AWS seems to tolerate this so we should, too
  642         do_test('some/source')
  643 
  644     @s3acl
  645     def test_object_PUT_copy_metadata_replace(self):
  646         date_header = self.get_date_header()
  647         timestamp = mktime(date_header)
  648         allowed_last_modified = [S3Timestamp(timestamp).s3xmlformat]
  649         status, headers, body = \
  650             self._test_object_PUT_copy(
  651                 swob.HTTPOk,
  652                 {'X-Amz-Metadata-Directive': 'REPLACE',
  653                  'X-Amz-Meta-Something': 'oh hai',
  654                  'X-Amz-Meta-Unreadable-Prefix': '\x04w',
  655                  'X-Amz-Meta-Unreadable-Suffix': 'h\x04',
  656                  'X-Amz-Meta-Lots-Of-Unprintable': 5 * '\x04',
  657                  'Cache-Control': 'hello',
  658                  'content-disposition': 'how are you',
  659                  'content-encoding': 'good and you',
  660                  'content-language': 'great',
  661                  'content-type': 'so',
  662                  'expires': 'yeah',
  663                  'x-robots-tag': 'bye'})
  664         date_header = self.get_date_header()
  665         timestamp = mktime(date_header)
  666         allowed_last_modified.append(S3Timestamp(timestamp).s3xmlformat)
  667 
  668         self.assertEqual(status.split()[0], '200')
  669         self.assertEqual(headers['Content-Type'], 'application/xml')
  670         self.assertIsNone(headers.get('etag'))
  671         elem = fromstring(body, 'CopyObjectResult')
  672         self.assertIn(elem.find('LastModified').text, allowed_last_modified)
  673         self.assertEqual(elem.find('ETag').text, '"%s"' % self.etag)
  674 
  675         _, _, headers = self.swift.calls_with_headers[-1]
  676         self.assertEqual(headers['X-Copy-From'], '/some/source')
  677         # Check that metadata is included if replace directive is specified
  678         # and that Fresh Metadata is set
  679         self.assertTrue(headers.get('X-Fresh-Metadata') == 'True')
  680         self.assertEqual(headers['X-Object-Meta-Something'], 'oh hai')
  681         self.assertEqual(headers['X-Object-Meta-Unreadable-Prefix'],
  682                          '=?UTF-8?Q?=04w?=')
  683         self.assertEqual(headers['X-Object-Meta-Unreadable-Suffix'],
  684                          '=?UTF-8?Q?h=04?=')
  685         self.assertEqual(headers['X-Object-Meta-Lots-Of-Unprintable'],
  686                          '=?UTF-8?B?BAQEBAQ=?=')
  687         # Check other metadata is set
  688         self.assertEqual(headers['Cache-Control'], 'hello')
  689         self.assertEqual(headers['Content-Disposition'], 'how are you')
  690         self.assertEqual(headers['Content-Encoding'], 'good and you')
  691         self.assertEqual(headers['Content-Language'], 'great')
  692         # Content-Type can't be set during an S3 copy operation
  693         self.assertIsNone(headers.get('Content-Type'))
  694         self.assertEqual(headers['Expires'], 'yeah')
  695         self.assertEqual(headers['X-Robots-Tag'], 'bye')
  696 
  697         self.assertEqual(headers['Content-Length'], '0')
  698 
  699     @s3acl
  700     def test_object_PUT_copy_metadata_copy(self):
  701         date_header = self.get_date_header()
  702         timestamp = mktime(date_header)
  703         allowed_last_modified = [S3Timestamp(timestamp).s3xmlformat]
  704         status, headers, body = \
  705             self._test_object_PUT_copy(
  706                 swob.HTTPOk,
  707                 {'X-Amz-Metadata-Directive': 'COPY',
  708                  'X-Amz-Meta-Something': 'oh hai',
  709                  'X-Amz-Meta-Unreadable-Prefix': '\x04w',
  710                  'X-Amz-Meta-Unreadable-Suffix': 'h\x04',
  711                  'X-Amz-Meta-Lots-Of-Unprintable': 5 * '\x04',
  712                  'Cache-Control': 'hello',
  713                  'content-disposition': 'how are you',
  714                  'content-encoding': 'good and you',
  715                  'content-language': 'great',
  716                  'content-type': 'so',
  717                  'expires': 'yeah',
  718                  'x-robots-tag': 'bye'})
  719         date_header = self.get_date_header()
  720         timestamp = mktime(date_header)
  721         allowed_last_modified.append(S3Timestamp(timestamp).s3xmlformat)
  722 
  723         self.assertEqual(status.split()[0], '200')
  724         self.assertEqual(headers['Content-Type'], 'application/xml')
  725         self.assertIsNone(headers.get('etag'))
  726 
  727         elem = fromstring(body, 'CopyObjectResult')
  728         self.assertIn(elem.find('LastModified').text, allowed_last_modified)
  729         self.assertEqual(elem.find('ETag').text, '"%s"' % self.etag)
  730 
  731         _, _, headers = self.swift.calls_with_headers[-1]
  732         self.assertEqual(headers['X-Copy-From'], '/some/source')
  733         # Check that metadata is omited if COPY directive is specified
  734         self.assertIsNone(headers.get('X-Fresh-Metadata'))
  735         self.assertIsNone(headers.get('X-Object-Meta-Something'))
  736         self.assertIsNone(headers.get('X-Object-Meta-Unreadable-Prefix'))
  737         self.assertIsNone(headers.get('X-Object-Meta-Unreadable-Suffix'))
  738         self.assertIsNone(headers.get('X-Object-Meta-Lots-Of-Unprintable'))
  739         self.assertIsNone(headers.get('Cache-Control'))
  740         self.assertIsNone(headers.get('Content-Disposition'))
  741         self.assertIsNone(headers.get('Content-Encoding'))
  742         self.assertIsNone(headers.get('Content-Language'))
  743         self.assertIsNone(headers.get('Content-Type'))
  744         self.assertIsNone(headers.get('Expires'))
  745         self.assertIsNone(headers.get('X-Robots-Tag'))
  746 
  747         self.assertEqual(headers['Content-Length'], '0')
  748 
  749     @s3acl
  750     def test_object_PUT_copy_self(self):
  751         status, headers, body = \
  752             self._test_object_PUT_copy_self(swob.HTTPOk)
  753         self.assertEqual(status.split()[0], '400')
  754         elem = fromstring(body, 'Error')
  755         err_msg = ("This copy request is illegal because it is trying to copy "
  756                    "an object to itself without changing the object's "
  757                    "metadata, storage class, website redirect location or "
  758                    "encryption attributes.")
  759         self.assertEqual(elem.find('Code').text, 'InvalidRequest')
  760         self.assertEqual(elem.find('Message').text, err_msg)
  761 
  762     @s3acl
  763     def test_object_PUT_copy_self_metadata_copy(self):
  764         header = {'x-amz-metadata-directive': 'COPY'}
  765         status, headers, body = \
  766             self._test_object_PUT_copy_self(swob.HTTPOk, header)
  767         self.assertEqual(status.split()[0], '400')
  768         elem = fromstring(body, 'Error')
  769         err_msg = ("This copy request is illegal because it is trying to copy "
  770                    "an object to itself without changing the object's "
  771                    "metadata, storage class, website redirect location or "
  772                    "encryption attributes.")
  773         self.assertEqual(elem.find('Code').text, 'InvalidRequest')
  774         self.assertEqual(elem.find('Message').text, err_msg)
  775 
  776     @s3acl
  777     def test_object_PUT_copy_self_metadata_replace(self):
  778         date_header = self.get_date_header()
  779         timestamp = mktime(date_header)
  780         allowed_last_modified = [S3Timestamp(timestamp).s3xmlformat]
  781         header = {'x-amz-metadata-directive': 'REPLACE',
  782                   'Date': date_header}
  783         status, headers, body = self._test_object_PUT_copy_self(
  784             swob.HTTPOk, header, timestamp=timestamp)
  785         date_header = self.get_date_header()
  786         timestamp = mktime(date_header)
  787         allowed_last_modified.append(S3Timestamp(timestamp).s3xmlformat)
  788 
  789         self.assertEqual(status.split()[0], '200')
  790         self.assertEqual(headers['Content-Type'], 'application/xml')
  791         self.assertTrue(headers.get('etag') is None)
  792         elem = fromstring(body, 'CopyObjectResult')
  793         self.assertIn(elem.find('LastModified').text, allowed_last_modified)
  794         self.assertEqual(elem.find('ETag').text, '"%s"' % self.etag)
  795 
  796         _, _, headers = self.swift.calls_with_headers[-1]
  797         self.assertEqual(headers['X-Copy-From'], '/bucket/object')
  798         self.assertEqual(headers['Content-Length'], '0')
  799 
  800     @s3acl
  801     def test_object_PUT_copy_headers_error(self):
  802         etag = '7dfa07a8e59ddbcd1dc84d4c4f82aea1'
  803         last_modified_since = 'Fri, 01 Apr 2014 12:00:00 GMT'
  804 
  805         header = {'X-Amz-Copy-Source-If-Match': etag,
  806                   'Date': self.get_date_header()}
  807         status, header, body = \
  808             self._test_object_PUT_copy(swob.HTTPPreconditionFailed,
  809                                        header)
  810         self.assertEqual(self._get_error_code(body), 'PreconditionFailed')
  811 
  812         header = {'X-Amz-Copy-Source-If-None-Match': etag}
  813         status, header, body = \
  814             self._test_object_PUT_copy(swob.HTTPNotModified,
  815                                        header)
  816         self.assertEqual(self._get_error_code(body), 'PreconditionFailed')
  817 
  818         header = {'X-Amz-Copy-Source-If-Modified-Since': last_modified_since}
  819         status, header, body = \
  820             self._test_object_PUT_copy(swob.HTTPNotModified,
  821                                        header)
  822         self.assertEqual(self._get_error_code(body), 'PreconditionFailed')
  823 
  824         header = \
  825             {'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since}
  826         status, header, body = \
  827             self._test_object_PUT_copy(swob.HTTPPreconditionFailed,
  828                                        header)
  829         self.assertEqual(self._get_error_code(body), 'PreconditionFailed')
  830 
  831     def test_object_PUT_copy_headers_with_match(self):
  832         etag = '7dfa07a8e59ddbcd1dc84d4c4f82aea1'
  833         last_modified_since = 'Fri, 01 Apr 2014 11:00:00 GMT'
  834 
  835         header = {'X-Amz-Copy-Source-If-Match': etag,
  836                   'X-Amz-Copy-Source-If-Modified-Since': last_modified_since,
  837                   'Date': self.get_date_header()}
  838         status, header, body = \
  839             self._test_object_PUT_copy(swob.HTTPOk, header)
  840         self.assertEqual(status.split()[0], '200')
  841         self.assertEqual(len(self.swift.calls_with_headers), 2)
  842         _, _, headers = self.swift.calls_with_headers[-1]
  843         self.assertTrue(headers.get('If-Match') is None)
  844         self.assertTrue(headers.get('If-Modified-Since') is None)
  845         _, _, headers = self.swift.calls_with_headers[0]
  846         self.assertEqual(headers['If-Match'], etag)
  847         self.assertEqual(headers['If-Modified-Since'], last_modified_since)
  848 
  849     @s3acl(s3acl_only=True)
  850     def test_object_PUT_copy_headers_with_match_and_s3acl(self):
  851         etag = '7dfa07a8e59ddbcd1dc84d4c4f82aea1'
  852         last_modified_since = 'Fri, 01 Apr 2014 11:00:00 GMT'
  853 
  854         header = {'X-Amz-Copy-Source-If-Match': etag,
  855                   'X-Amz-Copy-Source-If-Modified-Since': last_modified_since,
  856                   'Date': self.get_date_header()}
  857         status, header, body = \
  858             self._test_object_PUT_copy(swob.HTTPOk, header)
  859 
  860         self.assertEqual(status.split()[0], '200')
  861         self.assertEqual(len(self.swift.calls_with_headers), 3)
  862         # After the check of the copy source in the case of s3acl is valid,
  863         # s3api check the bucket write permissions of the destination.
  864         _, _, headers = self.swift.calls_with_headers[-2]
  865         self.assertTrue(headers.get('If-Match') is None)
  866         self.assertTrue(headers.get('If-Modified-Since') is None)
  867         _, _, headers = self.swift.calls_with_headers[-1]
  868         self.assertTrue(headers.get('If-Match') is None)
  869         self.assertTrue(headers.get('If-Modified-Since') is None)
  870         _, _, headers = self.swift.calls_with_headers[0]
  871         self.assertEqual(headers['If-Match'], etag)
  872         self.assertEqual(headers['If-Modified-Since'], last_modified_since)
  873 
  874     def test_object_PUT_copy_headers_with_not_match(self):
  875         etag = '7dfa07a8e59ddbcd1dc84d4c4f82aea1'
  876         last_modified_since = 'Fri, 01 Apr 2014 12:00:00 GMT'
  877 
  878         header = {'X-Amz-Copy-Source-If-None-Match': etag,
  879                   'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since,
  880                   'Date': self.get_date_header()}
  881         status, header, body = \
  882             self._test_object_PUT_copy(swob.HTTPOk, header)
  883 
  884         self.assertEqual(status.split()[0], '200')
  885         self.assertEqual(len(self.swift.calls_with_headers), 2)
  886         _, _, headers = self.swift.calls_with_headers[-1]
  887         self.assertTrue(headers.get('If-None-Match') is None)
  888         self.assertTrue(headers.get('If-Unmodified-Since') is None)
  889         _, _, headers = self.swift.calls_with_headers[0]
  890         self.assertEqual(headers['If-None-Match'], etag)
  891         self.assertEqual(headers['If-Unmodified-Since'], last_modified_since)
  892 
  893     @s3acl(s3acl_only=True)
  894     def test_object_PUT_copy_headers_with_not_match_and_s3acl(self):
  895         etag = '7dfa07a8e59ddbcd1dc84d4c4f82aea1'
  896         last_modified_since = 'Fri, 01 Apr 2014 12:00:00 GMT'
  897 
  898         header = {'X-Amz-Copy-Source-If-None-Match': etag,
  899                   'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since,
  900                   'Date': self.get_date_header()}
  901         status, header, body = \
  902             self._test_object_PUT_copy(swob.HTTPOk, header)
  903         self.assertEqual(status.split()[0], '200')
  904         # After the check of the copy source in the case of s3acl is valid,
  905         # s3api check the bucket write permissions of the destination.
  906         self.assertEqual(len(self.swift.calls_with_headers), 3)
  907         _, _, headers = self.swift.calls_with_headers[-1]
  908         self.assertTrue(headers.get('If-None-Match') is None)
  909         self.assertTrue(headers.get('If-Unmodified-Since') is None)
  910         _, _, headers = self.swift.calls_with_headers[0]
  911         self.assertEqual(headers['If-None-Match'], etag)
  912         self.assertEqual(headers['If-Unmodified-Since'], last_modified_since)
  913 
  914     @s3acl
  915     def test_object_POST_error(self):
  916         code = self._test_method_error('POST', '/bucket/object', None)
  917         self.assertEqual(code, 'NotImplemented')
  918 
  919     @s3acl
  920     def test_object_DELETE_error(self):
  921         code = self._test_method_error('DELETE', '/bucket/object',
  922                                        swob.HTTPUnauthorized)
  923         self.assertEqual(code, 'SignatureDoesNotMatch')
  924         code = self._test_method_error('DELETE', '/bucket/object',
  925                                        swob.HTTPForbidden)
  926         self.assertEqual(code, 'AccessDenied')
  927         code = self._test_method_error('DELETE', '/bucket/object',
  928                                        swob.HTTPServerError)
  929         self.assertEqual(code, 'InternalError')
  930         code = self._test_method_error('DELETE', '/bucket/object',
  931                                        swob.HTTPServiceUnavailable)
  932         self.assertEqual(code, 'InternalError')
  933 
  934         with patch(
  935                 'swift.common.middleware.s3api.s3request.get_container_info',
  936                 return_value={'status': 404}):
  937             code = self._test_method_error('DELETE', '/bucket/object',
  938                                            swob.HTTPNotFound)
  939             self.assertEqual(code, 'NoSuchBucket')
  940 
  941     @s3acl
  942     def test_object_DELETE_no_multipart(self):
  943         self.s3api.conf.allow_multipart_uploads = False
  944         req = Request.blank('/bucket/object',
  945                             environ={'REQUEST_METHOD': 'DELETE'},
  946                             headers={'Authorization': 'AWS test:tester:hmac',
  947                                      'Date': self.get_date_header()})
  948         status, headers, body = self.call_s3api(req)
  949         self.assertEqual(status.split()[0], '204')
  950 
  951         self.assertNotIn(('HEAD', '/v1/AUTH_test/bucket/object'),
  952                          self.swift.calls)
  953         self.assertIn(('DELETE', '/v1/AUTH_test/bucket/object'),
  954                       self.swift.calls)
  955         _, path = self.swift.calls[-1]
  956         self.assertEqual(path.count('?'), 0)
  957 
  958     @s3acl
  959     def test_object_DELETE_multipart(self):
  960         req = Request.blank('/bucket/object',
  961                             environ={'REQUEST_METHOD': 'DELETE'},
  962                             headers={'Authorization': 'AWS test:tester:hmac',
  963                                      'Date': self.get_date_header()})
  964         status, headers, body = self.call_s3api(req)
  965         self.assertEqual(status.split()[0], '204')
  966 
  967         self.assertIn(('HEAD', '/v1/AUTH_test/bucket/object'),
  968                       self.swift.calls)
  969         self.assertEqual(('DELETE', '/v1/AUTH_test/bucket/object'),
  970                          self.swift.calls[-1])
  971         _, path = self.swift.calls[-1]
  972         self.assertEqual(path.count('?'), 0)
  973 
  974     @s3acl
  975     def test_object_DELETE_missing(self):
  976         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  977                             swob.HTTPNotFound, {}, None)
  978         req = Request.blank('/bucket/object',
  979                             environ={'REQUEST_METHOD': 'DELETE'},
  980                             headers={'Authorization': 'AWS test:tester:hmac',
  981                                      'Date': self.get_date_header()})
  982         status, headers, body = self.call_s3api(req)
  983         self.assertEqual(status.split()[0], '204')
  984 
  985         self.assertIn(('HEAD', '/v1/AUTH_test/bucket/object'),
  986                       self.swift.calls)
  987         self.assertNotIn(('DELETE', '/v1/AUTH_test/bucket/object'),
  988                          self.swift.calls)
  989 
  990     @s3acl
  991     def test_slo_object_DELETE(self):
  992         self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
  993                             swob.HTTPOk,
  994                             {'x-static-large-object': 'True'},
  995                             None)
  996         self.swift.register('DELETE', '/v1/AUTH_test/bucket/object',
  997                             swob.HTTPOk, {}, '<SLO delete results>')
  998         req = Request.blank('/bucket/object',
  999                             environ={'REQUEST_METHOD': 'DELETE'},
 1000                             headers={'Authorization': 'AWS test:tester:hmac',
 1001                                      'Date': self.get_date_header(),
 1002                                      'Content-Type': 'foo/bar'})
 1003         status, headers, body = self.call_s3api(req)
 1004         self.assertEqual(status.split()[0], '204')
 1005         self.assertEqual(body, '')
 1006 
 1007         self.assertIn(('HEAD', '/v1/AUTH_test/bucket/object'),
 1008                       self.swift.calls)
 1009         self.assertIn(('DELETE', '/v1/AUTH_test/bucket/object'
 1010                                  '?multipart-manifest=delete'),
 1011                       self.swift.calls)
 1012         _, path, headers = self.swift.calls_with_headers[-1]
 1013         path, query_string = path.split('?', 1)
 1014         query = {}
 1015         for q in query_string.split('&'):
 1016             key, arg = q.split('=')
 1017             query[key] = arg
 1018         self.assertEqual(query['multipart-manifest'], 'delete')
 1019         self.assertNotIn('Content-Type', headers)
 1020 
 1021     def _test_object_for_s3acl(self, method, account):
 1022         req = Request.blank('/bucket/object',
 1023                             environ={'REQUEST_METHOD': method},
 1024                             headers={'Authorization': 'AWS %s:hmac' % account,
 1025                                      'Date': self.get_date_header()})
 1026         return self.call_s3api(req)
 1027 
 1028     def _test_set_container_permission(self, account, permission):
 1029         grants = [Grant(User(account), permission)]
 1030         headers = \
 1031             encode_acl('container',
 1032                        ACL(Owner('test:tester', 'test:tester'), grants))
 1033         self.swift.register('HEAD', '/v1/AUTH_test/bucket',
 1034                             swob.HTTPNoContent, headers, None)
 1035 
 1036     @s3acl(s3acl_only=True)
 1037     def test_object_GET_without_permission(self):
 1038         status, headers, body = self._test_object_for_s3acl('GET',
 1039                                                             'test:other')
 1040         self.assertEqual(self._get_error_code(body), 'AccessDenied')
 1041 
 1042     @s3acl(s3acl_only=True)
 1043     def test_object_GET_with_read_permission(self):
 1044         status, headers, body = self._test_object_for_s3acl('GET',
 1045                                                             'test:read')
 1046         self.assertEqual(status.split()[0], '200')
 1047 
 1048     @s3acl(s3acl_only=True)
 1049     def test_object_GET_with_fullcontrol_permission(self):
 1050         status, headers, body = \
 1051             self._test_object_for_s3acl('GET', 'test:full_control')
 1052         self.assertEqual(status.split()[0], '200')
 1053 
 1054     @s3acl(s3acl_only=True)
 1055     def test_object_PUT_without_permission(self):
 1056         status, headers, body = self._test_object_for_s3acl('PUT',
 1057                                                             'test:other')
 1058         self.assertEqual(self._get_error_code(body), 'AccessDenied')
 1059 
 1060     @s3acl(s3acl_only=True)
 1061     def test_object_PUT_with_owner_permission(self):
 1062         status, headers, body = self._test_object_for_s3acl('PUT',
 1063                                                             'test:tester')
 1064         self.assertEqual(status.split()[0], '200')
 1065 
 1066     @s3acl(s3acl_only=True)
 1067     def test_object_PUT_with_write_permission(self):
 1068         account = 'test:other'
 1069         self._test_set_container_permission(account, 'WRITE')
 1070         status, headers, body = self._test_object_for_s3acl('PUT', account)
 1071         self.assertEqual(status.split()[0], '200')
 1072 
 1073     @s3acl(s3acl_only=True)
 1074     def test_object_PUT_with_fullcontrol_permission(self):
 1075         account = 'test:other'
 1076         self._test_set_container_permission(account, 'FULL_CONTROL')
 1077         status, headers, body = \
 1078             self._test_object_for_s3acl('PUT', account)
 1079         self.assertEqual(status.split()[0], '200')
 1080 
 1081     @s3acl(s3acl_only=True)
 1082     def test_object_DELETE_without_permission(self):
 1083         account = 'test:other'
 1084         status, headers, body = self._test_object_for_s3acl('DELETE',
 1085                                                             account)
 1086         self.assertEqual(self._get_error_code(body), 'AccessDenied')
 1087 
 1088     @s3acl(s3acl_only=True)
 1089     def test_object_DELETE_with_owner_permission(self):
 1090         status, headers, body = self._test_object_for_s3acl('DELETE',
 1091                                                             'test:tester')
 1092         self.assertEqual(status.split()[0], '204')
 1093 
 1094     @s3acl(s3acl_only=True)
 1095     def test_object_DELETE_with_write_permission(self):
 1096         account = 'test:other'
 1097         self._test_set_container_permission(account, 'WRITE')
 1098         status, headers, body = self._test_object_for_s3acl('DELETE',
 1099                                                             account)
 1100         self.assertEqual(status.split()[0], '204')
 1101 
 1102     @s3acl(s3acl_only=True)
 1103     def test_object_DELETE_with_fullcontrol_permission(self):
 1104         account = 'test:other'
 1105         self._test_set_container_permission(account, 'FULL_CONTROL')
 1106         status, headers, body = self._test_object_for_s3acl('DELETE', account)
 1107         self.assertEqual(status.split()[0], '204')
 1108 
 1109     def _test_object_copy_for_s3acl(self, account, src_permission=None,
 1110                                     src_path='/src_bucket/src_obj'):
 1111         owner = 'test:tester'
 1112         grants = [Grant(User(account), src_permission)] \
 1113             if src_permission else [Grant(User(owner), 'FULL_CONTROL')]
 1114         src_o_headers = \
 1115             encode_acl('object', ACL(Owner(owner, owner), grants))
 1116         src_o_headers.update({'last-modified': self.last_modified})
 1117         self.swift.register(
 1118             'HEAD', join('/v1/AUTH_test', src_path.lstrip('/')),
 1119             swob.HTTPOk, src_o_headers, None)
 1120 
 1121         req = Request.blank(
 1122             '/bucket/object',
 1123             environ={'REQUEST_METHOD': 'PUT'},
 1124             headers={'Authorization': 'AWS %s:hmac' % account,
 1125                      'X-Amz-Copy-Source': src_path,
 1126                      'Date': self.get_date_header()})
 1127 
 1128         return self.call_s3api(req)
 1129 
 1130     @s3acl(s3acl_only=True)
 1131     def test_object_PUT_copy_with_owner_permission(self):
 1132         status, headers, body = \
 1133             self._test_object_copy_for_s3acl('test:tester')
 1134         self.assertEqual(status.split()[0], '200')
 1135 
 1136     @s3acl(s3acl_only=True)
 1137     def test_object_PUT_copy_with_fullcontrol_permission(self):
 1138         status, headers, body = \
 1139             self._test_object_copy_for_s3acl('test:full_control',
 1140                                              'FULL_CONTROL')
 1141         self.assertEqual(status.split()[0], '200')
 1142 
 1143     @s3acl(s3acl_only=True)
 1144     def test_object_PUT_copy_with_grantee_permission(self):
 1145         status, headers, body = \
 1146             self._test_object_copy_for_s3acl('test:write', 'READ')
 1147         self.assertEqual(status.split()[0], '200')
 1148 
 1149     @s3acl(s3acl_only=True)
 1150     def test_object_PUT_copy_without_src_obj_permission(self):
 1151         status, headers, body = \
 1152             self._test_object_copy_for_s3acl('test:write')
 1153         self.assertEqual(status.split()[0], '403')
 1154 
 1155     @s3acl(s3acl_only=True)
 1156     def test_object_PUT_copy_without_dst_container_permission(self):
 1157         status, headers, body = \
 1158             self._test_object_copy_for_s3acl('test:other', 'READ')
 1159         self.assertEqual(status.split()[0], '403')
 1160 
 1161     @s3acl(s3acl_only=True)
 1162     def test_object_PUT_copy_empty_src_path(self):
 1163         self.swift.register('PUT', '/v1/AUTH_test/bucket/object',
 1164                             swob.HTTPPreconditionFailed, {}, None)
 1165         status, headers, body = self._test_object_copy_for_s3acl(
 1166             'test:write', 'READ', src_path='')
 1167         self.assertEqual(status.split()[0], '400')
 1168 
 1169 
 1170 class TestS3ApiObjNonUTC(TestS3ApiObj):
 1171     def setUp(self):
 1172         self.orig_tz = os.environ.get('TZ', '')
 1173         os.environ['TZ'] = 'EST+05EDT,M4.1.0,M10.5.0'
 1174         time.tzset()
 1175         super(TestS3ApiObjNonUTC, self).setUp()
 1176 
 1177     def tearDown(self):
 1178         super(TestS3ApiObjNonUTC, self).tearDown()
 1179         os.environ['TZ'] = self.orig_tz
 1180         time.tzset()
 1181 
 1182 if __name__ == '__main__':
 1183     unittest.main()