diff --git a/.unittests b/.unittests new file mode 100755 index 0000000..012c5c5 --- /dev/null +++ b/.unittests @@ -0,0 +1,6 @@ +#!/bin/bash + +nosetests swift3/test/unit --exe --with-coverage --cover-package swift3 --cover-erase +exitcode=$? +rm -f .coverage +exit $exitcode diff --git a/swift3/test/unit/__init__.py b/swift3/test/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/swift3/test/unit/test_swift3.py b/swift3/test/unit/test_swift3.py new file mode 100644 index 0000000..56663ac --- /dev/null +++ b/swift3/test/unit/test_swift3.py @@ -0,0 +1,611 @@ +# 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 + +from swift3 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, []) + return [] + + +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, []) + return [] + + +class FakeAppObject(FakeApp): + def __init__(self, status=200): + FakeApp.__init__(self) + self.status = status + self.object_body = 'hello' + 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: + if 'HTTP_RANGE' in env: + resp = Response(body=self.object_body, + conditional_response=True) + return resp(env, start_response) + 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, []) + return [] + + +def start_response(*args): + pass + + +class TestSwift3(unittest.TestCase): + def setUp(self): + self.app = swift3.filter_factory({})(FakeApp()) + + 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, 'NoSuchBucket') + code = self._test_method_error(FakeAppBucket, 'GET', '/bucket', 0) + self.assertEquals(code, 'InvalidURI') + + 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, 'NoSuchBucket') + 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'}) + 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) + 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(''.join(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_GET_Range(self): + local_app = swift3.filter_factory({})(FakeAppObject()) + req = Request.blank('/bucket/object', + environ={'REQUEST_METHOD': 'GET'}, + headers={'Authorization': 'AWS test:tester:hmac', + 'Range': 'bytes=0-3'}) + resp = local_app(req.environ, local_app.app.do_start_response) + self.assertEquals(local_app.app.response_args[0].split()[0], '206') + + headers = dict(local_app.app.response_args[1]) + self.assertTrue('Content-Range' in headers) + self.assertTrue(headers['Content-Range'].startswith('bytes 0-3')) + + 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, 'NoSuchBucket') + 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', []) + return [] + 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') + + 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_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', []) + return [] + 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') + +if __name__ == '__main__': + unittest.main()