PrivacyScanData/S3/new3/openstack_swift_s3-master/tests/unit/test_swifts3.py

# Copyright (c) 2011 OpenStack, LLC.
#
# 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
from datetime import datetime
import cgi
import hashlib

from webob import Request, Response
from webob.exc import HTTPUnauthorized, HTTPCreated, HTTPNoContent,\
     HTTPAccepted, HTTPBadRequest, HTTPNotFound, HTTPConflict
import xml.dom.minidom
import simplejson

import mock

from swifts3 import middleware as swift3


class FakeApp(object):
    def __init__(self):
        self.app = self
        self.response_args = []

    def __call__(self, env, start_response):
        return "FAKE APP"

    def do_start_response(self, *args):
        self.response_args.extend(args)


class FakeAppService(FakeApp):
    def __init__(self, status=200):
        FakeApp.__init__(self)
        self.status = status
        self.buckets = (('apple', 1, 200), ('orange', 3, 430))

    def __call__(self, env, start_response):
        if self.status == 200:
            start_response(Response().status, [('Content-Type', 'text/xml')])
            json_pattern = ['"name":%s', '"count":%s', '"bytes":%s']
            json_pattern = '{' + ','.join(json_pattern) + '}'
            json_out = []
            for b in self.buckets:
                name = simplejson.dumps(b[0])
                json_out.append(json_pattern %
                                (name, b[1], b[2]))
            account_list = '[' + ','.join(json_out) + ']'
            return account_list
        elif self.status == 401:
            start_response(HTTPUnauthorized().status, [])
        else:
            start_response(HTTPBadRequest().status, [])


class FakeAppBucket(FakeApp):
    def __init__(self, status=200):
        FakeApp.__init__(self)
        self.status = status
        self.objects = (('rose', '2011-01-05T02:19:14.275290', 0, 303),
                        ('viola', '2011-01-05T02:19:14.275290', 0, 3909),
                        ('lily', '2011-01-05T02:19:14.275290', 0, 3909))

    def __call__(self, env, start_response):
        if env['REQUEST_METHOD'] == 'GET':
            if self.status == 200:
                start_response(Response().status,
                               [('Content-Type', 'text/xml')])
                json_pattern = ['"name":%s', '"last_modified":%s', '"hash":%s',
                                '"bytes":%s']
                json_pattern = '{' + ','.join(json_pattern) + '}'
                json_out = []
                for b in self.objects:
                    name = simplejson.dumps(b[0])
                    time = simplejson.dumps(b[1])
                    json_out.append(json_pattern %
                                    (name, time, b[2], b[3]))
                account_list = '[' + ','.join(json_out) + ']'
                return account_list
            elif self.status == 401:
                start_response(HTTPUnauthorized().status, [])
            elif self.status == 404:
                start_response(HTTPNotFound().status, [])
            else:
                start_response(HTTPBadRequest().status, [])
        elif env['REQUEST_METHOD'] == 'PUT':
            if self.status == 201:
                start_response(HTTPCreated().status, [])
            elif self.status == 401:
                start_response(HTTPUnauthorized().status, [])
            elif self.status == 202:
                start_response(HTTPAccepted().status, [])
            else:
                start_response(HTTPBadRequest().status, [])
        elif env['REQUEST_METHOD'] == 'DELETE':
            if self.status == 204:
                start_response(HTTPNoContent().status, [])
            elif self.status == 401:
                start_response(HTTPUnauthorized().status, [])
            elif self.status == 404:
                start_response(HTTPNotFound().status, [])
            elif self.status == 409:
                start_response(HTTPConflict().status, [])
            else:
                start_response(HTTPBadRequest().status, [])


class FakeAppBucketMPU(FakeAppBucket):
    def __init__(self, status=200):
        FakeApp.__init__(self)
        self.status = status
        self.objects = [('rose/id/meta', '2011-01-05T02:19:14.275290', 0, 303)]


class FakeAppObject(FakeApp):
    def __init__(self, status=200):
        FakeApp.__init__(self)
        self.status = status
        str_body = {"object": {"name": "bucket_new2",
                               "last_modified": "2012-08-27T14:55:42.963Z",
                               "hash": "1b2cf535f27731c974343645a3985328",
                               "bytes": 123,
                              }}
        self.object_body = simplejson.dumps(str_body)
        self.response_headers = {'Content-Type': 'text/html',
                                 'Content-Length': len(self.object_body),
                                 'x-object-meta-test': 'swift',
                                 'etag': '1b2cf535f27731c974343645a3985328',
                                 'last-modified': '2011-01-05T02:19:14.275290'}

    def __call__(self, env, start_response):
        if env['REQUEST_METHOD'] == 'GET' or env['REQUEST_METHOD'] == 'HEAD':
            if self.status == 200:
                start_response(Response().status,
                               self.response_headers.items())
                if env['REQUEST_METHOD'] == 'GET':
                    return self.object_body
            elif self.status == 401:
                start_response(HTTPUnauthorized().status, [])
            elif self.status == 404:
                start_response(HTTPNotFound().status, [])
            else:
                start_response(HTTPBadRequest().status, [])
        elif env['REQUEST_METHOD'] == 'PUT':
            if self.status == 201:
                start_response(HTTPCreated().status,
                               [('etag', self.response_headers['etag'])])
            elif self.status == 401:
                start_response(HTTPUnauthorized().status, [])
            elif self.status == 404:
                start_response(HTTPNotFound().status, [])
            else:
                start_response(HTTPBadRequest().status, [])
        elif env['REQUEST_METHOD'] == 'DELETE':
            if self.status == 204:
                start_response(HTTPNoContent().status, [])
            elif self.status == 401:
                start_response(HTTPUnauthorized().status, [])
            elif self.status == 404:
                start_response(HTTPNotFound().status, [])
            else:
                start_response(HTTPBadRequest().status, [])


def start_response(*args):
    pass


class TestSwift3(unittest.TestCase):
    def setUp(self):
        self.app = swift3.filter_factory({})(FakeApp())
        self.MULTIPART_UPLOAD_PREFIX = 'mpu.'

    def test_non_s3_request_passthrough(self):
        req = Request.blank('/something')
        resp = self.app(req.environ, start_response)
        self.assertEquals(resp, 'FAKE APP')

    def test_bad_format_authorization(self):
        req = Request.blank('/something',
                            headers={'Authorization': 'hoge'})
        resp = self.app(req.environ, start_response)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'Error')
        code = dom.getElementsByTagName('Code')[0].childNodes[0].nodeValue
        self.assertEquals(code, 'InvalidArgument')

    def test_bad_method(self):
        req = Request.blank('/',
                            environ={'REQUEST_METHOD': 'PUT'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = self.app(req.environ, start_response)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'Error')
        code = dom.getElementsByTagName('Code')[0].childNodes[0].nodeValue
        self.assertEquals(code, 'InvalidURI')

    def _test_method_error(self, cl, method, path, status):
        local_app = swift3.filter_factory({})(cl(status))
        req = Request.blank(path,
                            environ={'REQUEST_METHOD': method},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, start_response)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'Error')
        return dom.getElementsByTagName('Code')[0].childNodes[0].nodeValue

    def test_service_GET_error(self):
        code = self._test_method_error(FakeAppService, 'GET', '/', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppService, 'GET', '/', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_service_GET(self):
        local_app = swift3.filter_factory({})(FakeAppService())
        req = Request.blank('/',
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'ListAllMyBucketsResult')

        buckets = [n for n in dom.getElementsByTagName('Bucket')]
        listing = [n for n in buckets[0].childNodes if n.nodeName != '#text']
        self.assertEquals(len(listing), 2)

        names = []
        for b in buckets:
            if b.childNodes[0].nodeName == 'Name':
                names.append(b.childNodes[0].childNodes[0].nodeValue)

        self.assertEquals(len(names), len(FakeAppService().buckets))
        for i in FakeAppService().buckets:
            self.assertTrue(i[0] in names)

    def test_bucket_GET_error(self):
        code = self._test_method_error(FakeAppBucket, 'GET', '/bucket', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppBucket, 'GET', '/bucket', 404)
        self.assertEquals(code, 'InvalidBucketName')
        code = self._test_method_error(FakeAppBucket, 'GET', '/bucket', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_bucket_GET_uploads(self):
        local_app = swift3.filter_factory({})(FakeAppBucketMPU())
        bucket_name = 'junk'
        req = Request.blank('/%s?uploads' % bucket_name,
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName,
                          'ListMultipartUploadsResult')
        name = dom.getElementsByTagName('Bucket')[0].childNodes[0].nodeValue
        self.assertEquals(name, bucket_name)

        objects = [n for n in dom.getElementsByTagName('Upload')]

        names = []
        for o in objects:
            if o.childNodes[0].nodeName == 'Key':
                names.append(o.childNodes[0].childNodes[0].nodeValue)
            if o.childNodes[-1].nodeName == 'Initiated':
                self.assertTrue(
                    o.childNodes[-1].childNodes[0].nodeValue.endswith('Z'))

        self.assertEquals(len(names), len(FakeAppBucketMPU().objects))
        for i in FakeAppBucketMPU().objects:
            for name in names:
                if name in i[0]:
                    self.assertTrue(name in i[0])

    def test_bucket_GET(self):
        local_app = swift3.filter_factory({})(FakeAppBucket())
        bucket_name = 'junk'
        req = Request.blank('/%s' % bucket_name,
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'ListBucketResult')
        name = dom.getElementsByTagName('Name')[0].childNodes[0].nodeValue
        self.assertEquals(name, bucket_name)

        objects = [n for n in dom.getElementsByTagName('Contents')]
        listing = [n for n in objects[0].childNodes if n.nodeName != '#text']

        names = []
        for o in objects:
            if o.childNodes[0].nodeName == 'Key':
                names.append(o.childNodes[0].childNodes[0].nodeValue)
            if o.childNodes[1].nodeName == 'LastModified':
                self.assertTrue(
                    o.childNodes[1].childNodes[0].nodeValue.endswith('Z'))

        self.assertEquals(len(names), len(FakeAppBucket().objects))
        for i in FakeAppBucket().objects:
            self.assertTrue(i[0] in names)

    def test_bucket_GET_is_truncated(self):
        local_app = swift3.filter_factory({})(FakeAppBucket())
        bucket_name = 'junk'

        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET',
                         'QUERY_STRING': 'max-keys=3'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('IsTruncated')[0].
                childNodes[0].nodeValue, 'false')

        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET',
                         'QUERY_STRING': 'max-keys=2'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('IsTruncated')[0].
                childNodes[0].nodeValue, 'true')

    def test_bucket_GET_max_keys(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.query_string = env['QUERY_STRING']
                start_response('200 OK', [])
                return '[]'
        fake_app = FakeApp()
        local_app = swift3.filter_factory({})(fake_app)
        bucket_name = 'junk'

        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET',
                         'QUERY_STRING': 'max-keys=5'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('MaxKeys')[0].
                childNodes[0].nodeValue, '5')
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assert_(args['limit'] == '6')

        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET',
                         'QUERY_STRING': 'max-keys=5000'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('MaxKeys')[0].
                childNodes[0].nodeValue, '1000')
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assertEquals(args['limit'], '1001')

    def test_bucket_GET_passthroughs(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.query_string = env['QUERY_STRING']
                start_response('200 OK', [])
                return '[]'
        fake_app = FakeApp()
        local_app = swift3.filter_factory({})(fake_app)
        bucket_name = 'junk'
        req = Request.blank('/%s' % bucket_name,
                environ={'REQUEST_METHOD': 'GET', 'QUERY_STRING':
                         'delimiter=a&marker=b&prefix=c'},
                headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, lambda *args: None)
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.getElementsByTagName('Prefix')[0].
                childNodes[0].nodeValue, 'c')
        self.assertEquals(dom.getElementsByTagName('Marker')[0].
                childNodes[0].nodeValue, 'b')
        self.assertEquals(dom.getElementsByTagName('Delimiter')[0].
                childNodes[0].nodeValue, 'a')
        args = dict(cgi.parse_qsl(fake_app.query_string))
        self.assertEquals(args['delimiter'], 'a')
        self.assertEquals(args['marker'], 'b')
        self.assertEquals(args['prefix'], 'c')

    def test_bucket_PUT_error(self):
        code = self._test_method_error(FakeAppBucket, 'PUT', '/bucket', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppBucket, 'PUT', '/bucket', 202)
        self.assertEquals(code, 'BucketAlreadyExists')
        code = self._test_method_error(FakeAppBucket, 'PUT', '/bucket', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_bucket_PUT(self):
        local_app = swift3.filter_factory({})(FakeAppBucket(201))
        req = Request.blank('/bucket',
                            environ={'REQUEST_METHOD': 'PUT'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

    def test_bucket_DELETE_error(self):
        code = self._test_method_error(FakeAppBucket, 'DELETE', '/bucket', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppBucket, 'DELETE', '/bucket', 404)
        self.assertEquals(code, 'InvalidBucketName')
        code = self._test_method_error(FakeAppBucket, 'DELETE', '/bucket', 409)
        self.assertEquals(code, 'BucketNotEmpty')
        code = self._test_method_error(FakeAppBucket, 'DELETE', '/bucket', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_bucket_DELETE(self):
        local_app = swift3.filter_factory({})(FakeAppBucket(204))
        req = Request.blank('/bucket',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        with mock.patch('swifts3.middleware.BucketController'
                                             '.mpu_bucket_deletion') as mocked:
            mocked.return_value = Response(status=204)
            resp = local_app(req.environ, local_app.app.do_start_response)
            self.assertEquals(local_app.app.response_args[0].split()[0], '204')

    def test_bucket_DELETE_mpu(self):
        class FakeObjects(object):
            status_int = 200
            body = simplejson.dumps([{'name': 'some_name/meta'},
                                     {'name': 'some_name'}])
            headers = {"test": "test"}

        local_app = swift3.filter_factory({})(FakeAppBucket(204))
        req = Request.blank('/bucket',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        with mock.patch('webob.Request.get_response') as mocked:
            with mock.patch('swifts3.middleware.BucketController'
                       '.mpu_bucket_deletion_list_request') as mocked_list_req:
                mocked.return_value = Response(status=204)
                mocked_list_req.return_value = FakeObjects()
                resp = local_app(req.environ, local_app.app.do_start_response)
                self.assertEquals(local_app.app.response_args[0].split()[0],
                                                                        '204')

    def _check_acl(self, owner, resp):
        dom = xml.dom.minidom.parseString("".join(resp))
        self.assertEquals(dom.firstChild.nodeName, 'AccessControlPolicy')
        name = dom.getElementsByTagName('Permission')[0].childNodes[0].\
                                                                      nodeValue
        self.assertEquals(name, 'FULL_CONTROL')
        name = dom.getElementsByTagName('ID')[0].childNodes[0].nodeValue
        self.assertEquals(name, owner)

    def test_bucket_acl_GET(self):
        local_app = swift3.filter_factory({})(FakeAppBucket())
        bucket_name = 'junk'
        req = Request.blank('/%s?acl' % bucket_name,
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self._check_acl('test:tester', resp)

    def _test_object_GETorHEAD(self, method):
        local_app = swift3.filter_factory({})(FakeAppObject())
        req = Request.blank('/bucket/object',
                            environ={'REQUEST_METHOD': method},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        headers = dict(local_app.app.response_args[1])
        for key, val in local_app.app.response_headers.iteritems():
            if key in ('Content-Length', 'Content-Type', 'Content-Encoding',
                       'etag', 'last-modified'):
                self.assertTrue(key in headers)

                if key == 'Content-Length' and method == 'HEAD':
                    self.assertEquals(headers[key], '0')
                else:
                    self.assertEquals(headers[key], val)

            elif key.startswith('x-object-meta-'):
                self.assertTrue('x-amz-meta-' + key[14:] in headers)
                self.assertEquals(headers['x-amz-meta-' + key[14:]], val)

        if method == 'GET':
            self.assertEquals(resp, local_app.app.object_body)

    def test_object_HEAD(self):
        self._test_object_GETorHEAD('HEAD')

    def test_object_GET_error(self):
        code = self._test_method_error(FakeAppObject, 'GET',
                                       '/bucket/object', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppObject, 'GET',
                                       '/bucket/object', 404)
        self.assertEquals(code, 'NoSuchKey')
        code = self._test_method_error(FakeAppObject, 'GET',
                                       '/bucket/object', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_object_GET(self):
        self._test_object_GETorHEAD('GET')

    def test_object_PUT_error(self):
        code = self._test_method_error(FakeAppObject, 'PUT',
                                       '/bucket/object', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppObject, 'PUT',
                                       '/bucket/object', 404)
        self.assertEquals(code, 'InvalidBucketName')
        code = self._test_method_error(FakeAppObject, 'PUT',
                                       '/bucket/object', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_object_PUT(self):
        local_app = swift3.filter_factory({})(FakeAppObject(201))
        req = Request.blank('/bucket/object',
                            environ={'REQUEST_METHOD': 'PUT'},
                            headers={
                                   'Authorization': 'AWS test:tester:hmac',
                                   'x-amz-storage-class': 'REDUCED_REDUNDANCY',
                                   'Content-MD5': 'Gyz1NfJ3Mcl0NDZFo5hTKA==',
                                    })
        req.date = datetime.now()
        req.content_type = 'text/plain'
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        headers = dict(local_app.app.response_args[1])
        self.assertEquals(headers['ETag'],
                          "\"%s\"" % local_app.app.response_headers['etag'])

    def test_object_PUT_headers(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.req = Request(env)
                start_response('200 OK', [])
        app = FakeApp()
        local_app = swift3.filter_factory({})(app)
        req = Request.blank('/bucket/object',
                        environ={'REQUEST_METHOD': 'PUT'},
                        headers={'Authorization': 'AWS test:tester:hmac',
                                 'X-Amz-Storage-Class': 'REDUCED_REDUNDANCY',
                                 'X-Amz-Meta-Something': 'oh hai',
                                 'X-Amz-Copy-Source': '/some/source',
                                 'Content-MD5': 'ffoHqOWd280dyE1MT4KuoQ=='})
        req.date = datetime.now()
        req.content_type = 'text/plain'
        resp = local_app(req.environ, lambda *args: None)
        self.assertEquals(app.req.headers['ETag'],
                          '7dfa07a8e59ddbcd1dc84d4c4f82aea1')
        self.assertEquals(app.req.headers['X-Object-Meta-Something'], 'oh hai')
        self.assertEquals(app.req.headers['X-Copy-From'], '/some/source')

    def test_object_DELETE_error(self):
        code = self._test_method_error(FakeAppObject, 'DELETE',
                                       '/bucket/object', 401)
        self.assertEquals(code, 'AccessDenied')
        code = self._test_method_error(FakeAppObject, 'DELETE',
                                       '/bucket/object', 404)
        self.assertEquals(code, 'NoSuchKey')
        code = self._test_method_error(FakeAppObject, 'DELETE',
                                       '/bucket/object', 0)
        self.assertEquals(code, 'InvalidURI')

    def test_object_DELETE(self):
        local_app = swift3.filter_factory({})(FakeAppObject(204))
        req = Request.blank('/bucket/object',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self.assertEquals(local_app.app.response_args[0].split()[0], '204')

        local_app = swift3.filter_factory({})(FakeAppObject(204))
        req = Request.blank('/bucket/object',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'Authorization': 'AWS test:tester:hmac',
                                     'X-Object-Manifest': '123'})

        class FakeResponse(object):
            status_int = 200
            headers = {'X-Object-Manifest': \
                       'somefile_new2/18dc9b42566848abb58d8bb1ccbaf37a/'}

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse()
            resp = local_app(req.environ, local_app.app.do_start_response)
            self.assertEquals(local_app.app.response_args[0].split()[0], '204')

    def test_object_acl_GET(self):
        local_app = swift3.filter_factory({})(FakeAppObject())
        req = Request.blank('/bucket/object?acl',
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})
        resp = local_app(req.environ, local_app.app.do_start_response)
        self._check_acl('test:tester', resp)

    def test_multipart_GET(self):
        class FakeResponse(object):
            status_int = 200
            body = [{"name": "test_bucket",
                     "last_modified": "2011-01-05T02:19:14.275290",
                     "hash": 0,
                     "bytes": 3909}]
            body = simplejson.dumps(body)

        local_app = swift3.filter_factory({})(FakeAppObject(200))
        bucket_name = 'bucket'
        req = Request.blank('/%s/object?uploadId=deadbeef' % bucket_name,
                            environ={'REQUEST_METHOD': 'GET'},
                            headers={'Authorization': 'AWS test:tester:hmac'})

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse()
            resp = local_app(req.environ, local_app.app.do_start_response)
            self.assertEquals(local_app.app.response_args[0].split()[0], '200')

    def test_multipart_GET_error(self):
        code = self._test_method_error(
                   FakeAppObject,
                   'GET',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef',
                   0
                   )
        self.assertEquals(code, 'NoSuchUpload')

        code = self._test_method_error(
                   FakeAppObject,
                   'GET',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=123&part-number-marker=abc',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppObject,
                   'GET',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=!@#12',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

    def test_multipart_POST(self):
        class FakeResponse201(object):
            status_int = 201
            headers = {"ETag": "1b2cf535f27731c974343645a3985328"}

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse201()
            local_app = swift3.filter_factory({})(FakeAppObject(201))
            req = Request.blank('/' + self.MULTIPART_UPLOAD_PREFIX + \
                                'bucket/object?uploads=123',
                                environ={'REQUEST_METHOD': 'POST'},
                                headers={'Authorization': \
                                                      'AWS test:tester:hmac'})
            resp = local_app(req.environ, local_app.app.do_start_response)
            self.assertEquals(local_app.app.response_args[0].split()[0], '200')

        with mock.patch('swifts3.middleware.meta_request_head') as mocked:
            with mock.patch('webob.Request.get_response') as mocked_manifest:
                mocked.return_value = Response(status=200,
                                               headers={"test": "test"})
                mocked_manifest.return_value = FakeResponse201()
                local_app = swift3.filter_factory({})(FakeAppObject(201))
                req = Request.blank('/' + self.MULTIPART_UPLOAD_PREFIX + \
                                    'bucket/object?uploadId=deadbeef',
                                    environ={'REQUEST_METHOD': 'POST'},
                                    headers={'Authorization': \
                                                       'AWS test:tester:hmac'})
                resp = local_app(req.environ, local_app.app.do_start_response)
                self.assertEquals(local_app.app.response_args[0].split()[0],
                                                                         '200')

    def test_multipart_POST_error(self):
        class FakeResponse204(object):
            status_int = 204

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse204()
            code = self._test_method_error(
                       FakeAppService,
                       'POST',
                       '/' + self.MULTIPART_UPLOAD_PREFIX + \
                       'bucket/object?uploads',
                       0
                       )
            self.assertEquals(code, 'InvalidURI')

        class FakeResponse404(object):
            status_int = 404

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse404()
            code = self._test_method_error(
                       FakeAppService,
                       'POST',
                       '/' + self.MULTIPART_UPLOAD_PREFIX + \
                       'bucket/object?uploads',
                       404
                       )
            self.assertEquals(code, 'InvalidBucketName')

            code = self._test_method_error(
                       FakeAppService,
                       'POST',
                       '/' + self.MULTIPART_UPLOAD_PREFIX + \
                       'bucket/object?uploadId=123',
                       404
                       )
            self.assertEquals(code, 'NoSuchUpload')

        code = self._test_method_error(
                   FakeAppService,
                   'POST',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploads',
                   401
                   )
        self.assertEquals(code, 'AccessDenied')

        code = self._test_method_error(
                   FakeAppService,
                   'POST',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploads',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppService,
                   'POST',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef',
                   401
                   )
        self.assertEquals(code, 'AccessDenied')

        code = self._test_method_error(
                   FakeAppService,
                   'POST',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppService,
                   'POST',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=ad2@!',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        class FakeResponse200(object):
            status_int = 200
            headers = {"test": "test"}

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse200()
            code = self._test_method_error(
                       FakeAppObject,
                       'POST',
                       '/' + self.MULTIPART_UPLOAD_PREFIX + \
                       'bucket/object?uploadId=deadbeef',
                       0
                       )
            self.assertEquals(code, 'InvalidURI')

    def test_multipart_PUT_error(self):
        class FakeResponse(object):
            status_int = 200
            headers = {"test": "test"}

        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse()
            code = self._test_method_error(
                       FakeAppObject,
                       'PUT',
                       '/' + self.MULTIPART_UPLOAD_PREFIX + \
                       'bucket/object?uploadId=deadbeef&partNumber=1',
                       0
                       )
            self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppObject,
                   'PUT',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef&partNumber=1',
                   0
                   )
        self.assertEquals(code, 'NoSuchUpload')

        code = self._test_method_error(
                   FakeAppObject,
                   'PUT',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=#@!$',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppObject,
                   'PUT',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=123',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppObject,
                   'PUT',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef&partNumber=abc',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

    def test_multipart_DELETE(self):
        class FakeResponse200(object):
            status_int = 200
            body = [{"name": "test"}]
            body = simplejson.dumps(body)

        local_app = swift3.filter_factory({})(FakeAppObject(200))
        with mock.patch('webob.Request.get_response') as mocked:
            mocked.return_value = FakeResponse200()
            req = Request.blank('/bucket/object?uploadId=deadbeef',
                                environ={'REQUEST_METHOD': 'DELETE'},
                                headers={
                                     'Authorization': 'AWS test:tester:hmac',
                                     'Content-MD5': 'Gyz1NfJ3Mcl0NDZFo5hTKA==',
                                      })
            resp = local_app(req.environ, local_app.app.do_start_response)
            self.assertEquals(local_app.app.response_args[0].split()[0], '204')

    def test_multipart_DELETE_error(self):
        code = self._test_method_error(
                   FakeAppObject,
                   'DELETE',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=#@!$',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

        code = self._test_method_error(
                   FakeAppObject,
                   'DELETE',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef',
                   401
                   )
        self.assertEquals(code, 'AccessDenied')

        code = self._test_method_error(
                   FakeAppObject,
                   'DELETE',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef',
                   404
                   )
        self.assertEquals(code, 'InvalidBucketName')

        code = self._test_method_error(
                   FakeAppObject,
                   'DELETE',
                   '/' + self.MULTIPART_UPLOAD_PREFIX + \
                   'bucket/object?uploadId=deadbeef',
                   0
                   )
        self.assertEquals(code, 'InvalidURI')

    def test_canonical_string(self):
        """
        The hashes here were generated by running the same requests against
        boto.utils.canonical_string
        """
        def verify(hash, path, headers):
            req = Request.blank(path, headers=headers)
            self.assertEquals(hash,
                    hashlib.md5(swift3.canonical_string(req)).hexdigest())

        verify('6dd08c75e42190a1ce9468d1fd2eb787', '/bucket/object',
                {'Content-Type': 'text/plain', 'X-Amz-Something': 'test',
                 'Date': 'whatever'})

        verify('c8447135da232ae7517328f3429df481', '/bucket/object',
                {'Content-Type': 'text/plain', 'X-Amz-Something': 'test'})

        verify('bf49304103a4de5c325dce6384f2a4a2', '/bucket/object',
                {'content-type': 'text/plain'})

        verify('be01bd15d8d47f9fe5e2d9248cc6f180', '/bucket/object', {})

        verify('8d28cc4b8322211f6cc003256cd9439e', 'bucket/object',
                {'Content-MD5': 'somestuff'})

        verify('a822deb31213ad09af37b5a7fe59e55e', '/bucket/object?acl', {})

        verify('cce5dd1016595cb706c93f28d3eaa18f', '/bucket/object',
                {'Content-Type': 'text/plain', 'X-Amz-A': 'test',
                 'X-Amz-Z': 'whatever', 'X-Amz-B': 'lalala',
                 'X-Amz-Y': 'lalalalalalala'})

        verify('7506d97002c7d2de922cc0ec34af8846', '/bucket/object',
                {'Content-Type': None, 'X-Amz-Something': 'test'})

        verify('28f76d6162444a193b612cd6cb20e0be', '/bucket/object',
                {'Content-Type': None,
                 'X-Amz-Date': 'Mon, 11 Jul 2011 10:52:57 +0000',
                 'Date': 'Tue, 12 Jul 2011 10:52:57 +0000'})

        verify('ed6971e3eca5af4ee361f05d7c272e49', '/bucket/object',
                {'Content-Type': None,
                 'Date': 'Tue, 12 Jul 2011 10:52:57 +0000'})

        req1 = Request.blank('/', headers={'Content-Type': None,
                                           'X-Amz-Something': 'test'})
        req2 = Request.blank('/', headers={'Content-Type': '',
                                           'X-Amz-Something': 'test'})
        req3 = Request.blank('/', headers={'X-Amz-Something': 'test'})

        self.assertEquals(swift3.canonical_string(req1),
                swift3.canonical_string(req2))
        self.assertEquals(swift3.canonical_string(req2),
                swift3.canonical_string(req3))

    def test_signed_urls(self):
        class FakeApp(object):
            def __call__(self, env, start_response):
                self.req = Request(env)
                start_response('200 OK', [])
        app = FakeApp()
        local_app = swift3.filter_factory({})(app)
        req = Request.blank('/bucket/object?Signature=X&Expires=Y&'
                'AWSAccessKeyId=Z', environ={'REQUEST_METHOD': 'GET'})
        req.date = datetime.now()
        req.content_type = 'text/plain'
        resp = local_app(req.environ, lambda *args: None)
        self.assertEquals(app.req.headers['Authorization'], 'AWS Z:X')
        self.assertEquals(app.req.headers['Date'], 'Y')