# Copyright (c) 2011 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import hmac
import unittest
from hashlib import sha1
from StringIO import StringIO
from time import time

from swift.common.swob import Request, Response
from swift.common.middleware import tempauth, formpost
from swift.common.utils import split_path


class FakeApp(object):

    def __init__(self, status_headers_body_iter=None,
                 check_no_query_string=True):
        self.status_headers_body_iter = status_headers_body_iter
        if not self.status_headers_body_iter:
            self.status_headers_body_iter = iter([('404 Not Found', {
                'x-test-header-one-a': 'value1',
                'x-test-header-two-a': 'value2',
                'x-test-header-two-b': 'value3'}, '')])
        self.requests = []
        self.check_no_query_string = check_no_query_string

    def __call__(self, env, start_response):
        if self.check_no_query_string and env.get('QUERY_STRING'):
            raise Exception('Query string %s should have been discarded!' %
                            env['QUERY_STRING'])
        body = ''
        while True:
            chunk = env['wsgi.input'].read()
            if not chunk:
                break
            body += chunk
        env['wsgi.input'] = StringIO(body)
        self.requests.append(Request.blank('', environ=env))
        if env.get('swift.authorize_override') and \
                env.get('REMOTE_USER') != '.wsgi.pre_authed':
            raise Exception(
                'Invalid REMOTE_USER %r with swift.authorize_override' % (
                    env.get('REMOTE_USER'),))
        if 'swift.authorize' in env:
            resp = env['swift.authorize'](self.requests[-1])
            if resp:
                return resp(env, start_response)
        status, headers, body = self.status_headers_body_iter.next()
        return Response(status=status, headers=headers,
                        body=body)(env, start_response)


class TestParseAttrs(unittest.TestCase):

    def test_basic_content_type(self):
        name, attrs = formpost._parse_attrs('text/plain')
        self.assertEquals(name, 'text/plain')
        self.assertEquals(attrs, {})

    def test_content_type_with_charset(self):
        name, attrs = formpost._parse_attrs('text/plain; charset=UTF8')
        self.assertEquals(name, 'text/plain')
        self.assertEquals(attrs, {'charset': 'UTF8'})

    def test_content_disposition(self):
        name, attrs = formpost._parse_attrs(
            'form-data; name="somefile"; filename="test.html"')
        self.assertEquals(name, 'form-data')
        self.assertEquals(attrs, {'name': 'somefile', 'filename': 'test.html'})


class TestIterRequests(unittest.TestCase):

    def test_bad_start(self):
        it = formpost._iter_requests(StringIO('blah'), 'unique')
        exc = None
        try:
            it.next()
        except formpost.FormInvalid as err:
            exc = err
        self.assertEquals(str(exc), 'invalid starting boundary')

    def test_empty(self):
        it = formpost._iter_requests(StringIO('--unique'), 'unique')
        fp = it.next()
        self.assertEquals(fp.read(), '')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_basic(self):
        it = formpost._iter_requests(
            StringIO('--unique\r\nabcdefg\r\n--unique--'), 'unique')
        fp = it.next()
        self.assertEquals(fp.read(), 'abcdefg')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_basic2(self):
        it = formpost._iter_requests(
            StringIO('--unique\r\nabcdefg\r\n--unique\r\nhijkl\r\n--unique--'),
            'unique')
        fp = it.next()
        self.assertEquals(fp.read(), 'abcdefg')
        fp = it.next()
        self.assertEquals(fp.read(), 'hijkl')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_tiny_reads(self):
        it = formpost._iter_requests(
            StringIO('--unique\r\nabcdefg\r\n--unique\r\nhijkl\r\n--unique--'),
            'unique')
        fp = it.next()
        self.assertEquals(fp.read(2), 'ab')
        self.assertEquals(fp.read(2), 'cd')
        self.assertEquals(fp.read(2), 'ef')
        self.assertEquals(fp.read(2), 'g')
        self.assertEquals(fp.read(2), '')
        fp = it.next()
        self.assertEquals(fp.read(), 'hijkl')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_big_reads(self):
        it = formpost._iter_requests(
            StringIO('--unique\r\nabcdefg\r\n--unique\r\nhijkl\r\n--unique--'),
            'unique')
        fp = it.next()
        self.assertEquals(fp.read(65536), 'abcdefg')
        self.assertEquals(fp.read(), '')
        fp = it.next()
        self.assertEquals(fp.read(), 'hijkl')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_broken_mid_stream(self):
        # We go ahead and accept whatever is sent instead of rejecting the
        # whole request, in case the partial form is still useful.
        it = formpost._iter_requests(
            StringIO('--unique\r\nabc'), 'unique')
        fp = it.next()
        self.assertEquals(fp.read(), 'abc')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_readline(self):
        it = formpost._iter_requests(
            StringIO('--unique\r\nab\r\ncd\ref\ng\r\n--unique\r\nhi\r\n\r\n'
                     'jkl\r\n\r\n--unique--'), 'unique')
        fp = it.next()
        self.assertEquals(fp.readline(), 'ab\r\n')
        self.assertEquals(fp.readline(), 'cd\ref\ng')
        self.assertEquals(fp.readline(), '')
        fp = it.next()
        self.assertEquals(fp.readline(), 'hi\r\n')
        self.assertEquals(fp.readline(), '\r\n')
        self.assertEquals(fp.readline(), 'jkl\r\n')
        exc = None
        try:
            it.next()
        except StopIteration as err:
            exc = err
        self.assertTrue(exc is not None)

    def test_readline_with_tiny_chunks(self):
        orig_read_chunk_size = formpost.READ_CHUNK_SIZE
        try:
            formpost.READ_CHUNK_SIZE = 2
            it = formpost._iter_requests(
                StringIO('--unique\r\nab\r\ncd\ref\ng\r\n--unique\r\nhi\r\n'
                         '\r\njkl\r\n\r\n--unique--'), 'unique')
            fp = it.next()
            self.assertEquals(fp.readline(), 'ab\r\n')
            self.assertEquals(fp.readline(), 'cd\ref\ng')
            self.assertEquals(fp.readline(), '')
            fp = it.next()
            self.assertEquals(fp.readline(), 'hi\r\n')
            self.assertEquals(fp.readline(), '\r\n')
            self.assertEquals(fp.readline(), 'jkl\r\n')
            exc = None
            try:
                it.next()
            except StopIteration as err:
                exc = err
            self.assertTrue(exc is not None)
        finally:
            formpost.READ_CHUNK_SIZE = orig_read_chunk_size


class TestCappedFileLikeObject(unittest.TestCase):

    def test_whole(self):
        self.assertEquals(
            formpost._CappedFileLikeObject(StringIO('abc'), 10).read(), 'abc')

    def test_exceeded(self):
        exc = None
        try:
            formpost._CappedFileLikeObject(StringIO('abc'), 2).read()
        except EOFError as err:
            exc = err
        self.assertEquals(str(exc), 'max_file_size exceeded')

    def test_whole_readline(self):
        fp = formpost._CappedFileLikeObject(StringIO('abc\ndef'), 10)
        self.assertEquals(fp.readline(), 'abc\n')
        self.assertEquals(fp.readline(), 'def')
        self.assertEquals(fp.readline(), '')

    def test_exceeded_readline(self):
        fp = formpost._CappedFileLikeObject(StringIO('abc\ndef'), 5)
        self.assertEquals(fp.readline(), 'abc\n')
        exc = None
        try:
            self.assertEquals(fp.readline(), 'def')
        except EOFError as err:
            exc = err
        self.assertEquals(str(exc), 'max_file_size exceeded')

    def test_read_sized(self):
        fp = formpost._CappedFileLikeObject(StringIO('abcdefg'), 10)
        self.assertEquals(fp.read(2), 'ab')
        self.assertEquals(fp.read(2), 'cd')
        self.assertEquals(fp.read(2), 'ef')
        self.assertEquals(fp.read(2), 'g')
        self.assertEquals(fp.read(2), '')


class TestFormPost(unittest.TestCase):

    def setUp(self):
        self.app = FakeApp()
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)

    def _make_request(self, path, tempurl_keys=(), **kwargs):
        req = Request.blank(path, **kwargs)

        # Fake out the caching layer so that get_account_info() finds its
        # data. Include something that isn't tempurl keys to prove we skip it.
        meta = {'user-job-title': 'Personal Trainer',
                'user-real-name': 'Jim Shortz'}
        for idx, key in enumerate(tempurl_keys):
            meta_name = 'temp-url-key' + (("-%d" % (idx + 1) if idx else ""))
            if key:
                meta[meta_name] = key

        _junk, account, _junk, _junk = split_path(path, 2, 4)
        req.environ['swift.account/' + account] = self._fake_cache_env(
            account, tempurl_keys)
        return req

    def _fake_cache_env(self, account, tempurl_keys=()):
        # Fake out the caching layer so that get_account_info() finds its
        # data. Include something that isn't tempurl keys to prove we skip it.
        meta = {'user-job-title': 'Personal Trainer',
                'user-real-name': 'Jim Shortz'}
        for idx, key in enumerate(tempurl_keys):
            meta_name = 'temp-url-key' + ("-%d" % (idx + 1) if idx else "")
            if key:
                meta[meta_name] = key

        return {'status': 204,
                'container_count': '0',
                'total_object_count': '0',
                'bytes': '0',
                'meta': meta}

    def _make_sig_env_body(self, path, redirect, max_file_size, max_file_count,
                           expires, key, user_agent=True):
        sig = hmac.new(
            key,
            '%s\n%s\n%s\n%s\n%s' % (
                path, redirect, max_file_size, max_file_count, expires),
            sha1).hexdigest()
        body = [
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file1"; '
            'filename="testfile1.txt"',
            'Content-Type: text/plain',
            '',
            'Test File\nOne\n',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file2"; '
            'filename="testfile2.txt"',
            'Content-Type: text/plain',
            '',
            'Test\nFile\nTwo\n',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file3"; filename=""',
            'Content-Type: application/octet-stream',
            '',
            '',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR--',
            '',
        ]
        wsgi_errors = StringIO()
        env = {
            'CONTENT_TYPE': 'multipart/form-data; '
            'boundary=----WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'HTTP_ACCEPT_ENCODING': 'gzip, deflate',
            'HTTP_ACCEPT_LANGUAGE': 'en-us',
            'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;'
            'q=0.9,*/*;q=0.8',
            'HTTP_CONNECTION': 'keep-alive',
            'HTTP_HOST': 'ubuntu:8080',
            'HTTP_ORIGIN': 'file://',
            'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X '
            '10_7_2) AppleWebKit/534.52.7 (KHTML, like Gecko) '
            'Version/5.1.2 Safari/534.52.7',
            'PATH_INFO': path,
            'REMOTE_ADDR': '172.16.83.1',
            'REQUEST_METHOD': 'POST',
            'SCRIPT_NAME': '',
            'SERVER_NAME': '172.16.83.128',
            'SERVER_PORT': '8080',
            'SERVER_PROTOCOL': 'HTTP/1.0',
            'wsgi.errors': wsgi_errors,
            'wsgi.multiprocess': False,
            'wsgi.multithread': True,
            'wsgi.run_once': False,
            'wsgi.url_scheme': 'http',
            'wsgi.version': (1, 0),
        }
        if user_agent is False:
            del env['HTTP_USER_AGENT']

        return sig, env, body

    def test_passthrough(self):
        for method in ('HEAD', 'GET', 'PUT', 'POST', 'DELETE'):
            resp = self._make_request(
                '/v1/a/c/o',
                environ={'REQUEST_METHOD': method}).get_response(self.formpost)
            self.assertEquals(resp.status_int, 401)
            self.assertTrue('FormPost' not in resp.body)

    def test_auth_scheme(self):
        # FormPost rejects
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, int(time() - 10), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        authenticate_v = None
        for h, v in headers:
            if h.lower() == 'www-authenticate':
                authenticate_v = v
        self.assertTrue('FormPost: Form Expired' in body)
        self.assertEquals('Swift realm="unknown"', authenticate_v)

    def test_safari(self):
        key = 'abc'
        path = '/v1/AUTH_test/container'
        redirect = 'http://brim.net'
        max_file_size = 1024
        max_file_count = 10
        expires = int(time() + 86400)
        sig = hmac.new(
            key,
            '%s\n%s\n%s\n%s\n%s' % (
                path, redirect, max_file_size, max_file_count, expires),
            sha1).hexdigest()
        wsgi_input = StringIO('\r\n'.join([
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file1"; '
            'filename="testfile1.txt"',
            'Content-Type: text/plain',
            '',
            'Test File\nOne\n',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file2"; '
            'filename="testfile2.txt"',
            'Content-Type: text/plain',
            '',
            'Test\nFile\nTwo\n',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file3"; filename=""',
            'Content-Type: application/octet-stream',
            '',
            '',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR--',
            '',
        ]))
        wsgi_errors = StringIO()
        env = {
            'CONTENT_TYPE': 'multipart/form-data; '
            'boundary=----WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'HTTP_ACCEPT_ENCODING': 'gzip, deflate',
            'HTTP_ACCEPT_LANGUAGE': 'en-us',
            'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;'
            'q=0.9,*/*;q=0.8',
            'HTTP_CONNECTION': 'keep-alive',
            'HTTP_HOST': 'ubuntu:8080',
            'HTTP_ORIGIN': 'file://',
            'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X '
            '10_7_2) AppleWebKit/534.52.7 (KHTML, like Gecko) '
            'Version/5.1.2 Safari/534.52.7',
            'PATH_INFO': path,
            'REMOTE_ADDR': '172.16.83.1',
            'REQUEST_METHOD': 'POST',
            'SCRIPT_NAME': '',
            'SERVER_NAME': '172.16.83.128',
            'SERVER_PORT': '8080',
            'SERVER_PROTOCOL': 'HTTP/1.0',
            'swift.account/AUTH_test': self._fake_cache_env(
                'AUTH_test', [key]),
            'wsgi.errors': wsgi_errors,
            'wsgi.input': wsgi_input,
            'wsgi.multiprocess': False,
            'wsgi.multithread': True,
            'wsgi.run_once': False,
            'wsgi.url_scheme': 'http',
            'wsgi.version': (1, 0),
        }
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, 'http://brim.net?status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue('http://brim.net?status=201&message=' in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_firefox(self):
        key = 'abc'
        path = '/v1/AUTH_test/container'
        redirect = 'http://brim.net'
        max_file_size = 1024
        max_file_count = 10
        expires = int(time() + 86400)
        sig = hmac.new(
            key,
            '%s\n%s\n%s\n%s\n%s' % (
                path, redirect, max_file_size, max_file_count, expires),
            sha1).hexdigest()
        wsgi_input = StringIO('\r\n'.join([
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="file1"; '
            'filename="testfile1.txt"',
            'Content-Type: text/plain',
            '',
            'Test File\nOne\n',
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="file2"; '
            'filename="testfile2.txt"',
            'Content-Type: text/plain',
            '',
            'Test\nFile\nTwo\n',
            '-----------------------------168072824752491622650073',
            'Content-Disposition: form-data; name="file3"; filename=""',
            'Content-Type: application/octet-stream',
            '',
            '',
            '-----------------------------168072824752491622650073--',
            ''
        ]))
        wsgi_errors = StringIO()
        env = {
            'CONTENT_TYPE': 'multipart/form-data; '
            'boundary=---------------------------168072824752491622650073',
            'HTTP_ACCEPT_CHARSET': 'ISO-8859-1,utf-8;q=0.7,*;q=0.7',
            'HTTP_ACCEPT_ENCODING': 'gzip, deflate',
            'HTTP_ACCEPT_LANGUAGE': 'en-us,en;q=0.5',
            'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;'
            'q=0.9,*/*;q=0.8',
            'HTTP_CONNECTION': 'keep-alive',
            'HTTP_HOST': 'ubuntu:8080',
            'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; '
            'rv:8.0.1) Gecko/20100101 Firefox/8.0.1',
            'PATH_INFO': '/v1/AUTH_test/container',
            'REMOTE_ADDR': '172.16.83.1',
            'REQUEST_METHOD': 'POST',
            'SCRIPT_NAME': '',
            'SERVER_NAME': '172.16.83.128',
            'SERVER_PORT': '8080',
            'SERVER_PROTOCOL': 'HTTP/1.0',
            'swift.account/AUTH_test': self._fake_cache_env(
                'AUTH_test', [key]),
            'wsgi.errors': wsgi_errors,
            'wsgi.input': wsgi_input,
            'wsgi.multiprocess': False,
            'wsgi.multithread': True,
            'wsgi.run_once': False,
            'wsgi.url_scheme': 'http',
            'wsgi.version': (1, 0),
        }
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, 'http://brim.net?status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue('http://brim.net?status=201&message=' in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_chrome(self):
        key = 'abc'
        path = '/v1/AUTH_test/container'
        redirect = 'http://brim.net'
        max_file_size = 1024
        max_file_count = 10
        expires = int(time() + 86400)
        sig = hmac.new(
            key,
            '%s\n%s\n%s\n%s\n%s' % (
                path, redirect, max_file_size, max_file_count, expires),
            sha1).hexdigest()
        wsgi_input = StringIO('\r\n'.join([
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="file1"; '
            'filename="testfile1.txt"',
            'Content-Type: text/plain',
            '',
            'Test File\nOne\n',
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="file2"; '
            'filename="testfile2.txt"',
            'Content-Type: text/plain',
            '',
            'Test\nFile\nTwo\n',
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'Content-Disposition: form-data; name="file3"; filename=""',
            'Content-Type: application/octet-stream',
            '',
            '',
            '------WebKitFormBoundaryq3CFxUjfsDMu8XsA--',
            ''
        ]))
        wsgi_errors = StringIO()
        env = {
            'CONTENT_TYPE': 'multipart/form-data; '
            'boundary=----WebKitFormBoundaryq3CFxUjfsDMu8XsA',
            'HTTP_ACCEPT_CHARSET': 'ISO-8859-1,utf-8;q=0.7,*;q=0.3',
            'HTTP_ACCEPT_ENCODING': 'gzip,deflate,sdch',
            'HTTP_ACCEPT_LANGUAGE': 'en-US,en;q=0.8',
            'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;'
            'q=0.9,*/*;q=0.8',
            'HTTP_CACHE_CONTROL': 'max-age=0',
            'HTTP_CONNECTION': 'keep-alive',
            'HTTP_HOST': 'ubuntu:8080',
            'HTTP_ORIGIN': 'null',
            'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X '
            '10_7_2) AppleWebKit/535.7 (KHTML, like Gecko) '
            'Chrome/16.0.912.63 Safari/535.7',
            'PATH_INFO': '/v1/AUTH_test/container',
            'REMOTE_ADDR': '172.16.83.1',
            'REQUEST_METHOD': 'POST',
            'SCRIPT_NAME': '',
            'SERVER_NAME': '172.16.83.128',
            'SERVER_PORT': '8080',
            'SERVER_PROTOCOL': 'HTTP/1.0',
            'swift.account/AUTH_test': self._fake_cache_env(
                'AUTH_test', [key]),
            'wsgi.errors': wsgi_errors,
            'wsgi.input': wsgi_input,
            'wsgi.multiprocess': False,
            'wsgi.multithread': True,
            'wsgi.run_once': False,
            'wsgi.url_scheme': 'http',
            'wsgi.version': (1, 0),
        }
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, 'http://brim.net?status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue('http://brim.net?status=201&message=' in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_explorer(self):
        key = 'abc'
        path = '/v1/AUTH_test/container'
        redirect = 'http://brim.net'
        max_file_size = 1024
        max_file_count = 10
        expires = int(time() + 86400)
        sig = hmac.new(
            key,
            '%s\n%s\n%s\n%s\n%s' % (
                path, redirect, max_file_size, max_file_count, expires),
            sha1).hexdigest()
        wsgi_input = StringIO('\r\n'.join([
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="file1"; '
            'filename="C:\\testfile1.txt"',
            'Content-Type: text/plain',
            '',
            'Test File\nOne\n',
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="file2"; '
            'filename="C:\\testfile2.txt"',
            'Content-Type: text/plain',
            '',
            'Test\nFile\nTwo\n',
            '-----------------------------7db20d93017c',
            'Content-Disposition: form-data; name="file3"; filename=""',
            'Content-Type: application/octet-stream',
            '',
            '',
            '-----------------------------7db20d93017c--',
            ''
        ]))
        wsgi_errors = StringIO()
        env = {
            'CONTENT_TYPE': 'multipart/form-data; '
            'boundary=---------------------------7db20d93017c',
            'HTTP_ACCEPT_ENCODING': 'gzip, deflate',
            'HTTP_ACCEPT_LANGUAGE': 'en-US',
            'HTTP_ACCEPT': 'text/html, application/xhtml+xml, */*',
            'HTTP_CACHE_CONTROL': 'no-cache',
            'HTTP_CONNECTION': 'Keep-Alive',
            'HTTP_HOST': '172.16.83.128:8080',
            'HTTP_USER_AGENT': 'Mozilla/5.0 (compatible; MSIE 9.0; Windows NT '
            '6.1; WOW64; Trident/5.0)',
            'PATH_INFO': '/v1/AUTH_test/container',
            'REMOTE_ADDR': '172.16.83.129',
            'REQUEST_METHOD': 'POST',
            'SCRIPT_NAME': '',
            'SERVER_NAME': '172.16.83.128',
            'SERVER_PORT': '8080',
            'SERVER_PROTOCOL': 'HTTP/1.0',
            'swift.account/AUTH_test': self._fake_cache_env(
                'AUTH_test', [key]),
            'wsgi.errors': wsgi_errors,
            'wsgi.input': wsgi_input,
            'wsgi.multiprocess': False,
            'wsgi.multithread': True,
            'wsgi.run_once': False,
            'wsgi.url_scheme': 'http',
            'wsgi.version': (1, 0),
        }
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, 'http://brim.net?status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue('http://brim.net?status=201&message=' in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_messed_up_start(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://brim.net', 5, 10,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('XX' + '\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)

        def log_assert_int_status(env, response_status_int):
            self.assertTrue(isinstance(response_status_int, int))

        self.formpost._log_request = log_assert_int_status
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '400 Bad Request')
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: invalid starting boundary' in body)
        self.assertEquals(len(self.app.requests), 0)

    def test_max_file_size_exceeded(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://brim.net', 5, 10,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '400 Bad Request')
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: max_file_size exceeded' in body)
        self.assertEquals(len(self.app.requests), 0)

    def test_max_file_count_exceeded(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://brim.net', 1024, 1,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(
            location,
            'http://brim.net?status=400&message=max%20file%20count%20exceeded')
        self.assertEquals(exc_info, None)
        self.assertTrue(
            'http://brim.net?status=400&message=max%20file%20count%20exceeded'
            in body)
        self.assertEquals(len(self.app.requests), 1)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')

    def test_subrequest_does_not_pass_query(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
        env['QUERY_STRING'] = 'this=should&not=get&passed'
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(
            iter([('201 Created', {}, ''),
                  ('201 Created', {}, '')]),
            check_no_query_string=True)
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        # Make sure we 201 Created, which means we made the final subrequest
        # (and FakeApp verifies that no QUERY_STRING got passed).
        self.assertEquals(status, '201 Created')
        self.assertEquals(exc_info, None)
        self.assertTrue('201 Created' in body)
        self.assertEquals(len(self.app.requests), 2)

    def test_subrequest_fails(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://brim.net', 1024, 10,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('404 Not Found', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, 'http://brim.net?status=404&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue('http://brim.net?status=404&message=' in body)
        self.assertEquals(len(self.app.requests), 1)

    def test_truncated_attr_value(self):
        key = 'abc'
        redirect = 'a' * formpost.MAX_VALUE_LENGTH
        max_file_size = 1024
        max_file_count = 10
        expires = int(time() + 86400)
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', redirect, max_file_size, max_file_count,
            expires, key)
        # Tack on an extra char to redirect, but shouldn't matter since it
        # should get truncated off on read.
        redirect += 'b'
        env['wsgi.input'] = StringIO('\r\n'.join([
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file1"; '
            'filename="testfile1.txt"',
            'Content-Type: text/plain',
            '',
            'Test File\nOne\n',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file2"; '
            'filename="testfile2.txt"',
            'Content-Type: text/plain',
            '',
            'Test\nFile\nTwo\n',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="file3"; filename=""',
            'Content-Type: application/octet-stream',
            '',
            '',
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR--',
            '',
        ]))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(
            location,
            ('a' * formpost.MAX_VALUE_LENGTH) + '?status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue(
            ('a' * formpost.MAX_VALUE_LENGTH) + '?status=201&message=' in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_no_file_to_process(self):
        key = 'abc'
        redirect = 'http://brim.net'
        max_file_size = 1024
        max_file_count = 10
        expires = int(time() + 86400)
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', redirect, max_file_size, max_file_count,
            expires, key)
        env['wsgi.input'] = StringIO('\r\n'.join([
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="redirect"',
            '',
            redirect,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_size"',
            '',
            str(max_file_size),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="max_file_count"',
            '',
            str(max_file_count),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="expires"',
            '',
            str(expires),
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
            'Content-Disposition: form-data; name="signature"',
            '',
            sig,
            '------WebKitFormBoundaryNcxTqxSlX7t4TDkR--',
            '',
        ]))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(
            location,
            'http://brim.net?status=400&message=no%20files%20to%20process')
        self.assertEquals(exc_info, None)
        self.assertTrue(
            'http://brim.net?status=400&message=no%20files%20to%20process'
            in body)
        self.assertEquals(len(self.app.requests), 0)

    def test_formpost_without_useragent(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://redirect', 1024, 10,
            int(time() + 86400), key, user_agent=False)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)

        def start_response(s, h, e=None):
            pass
        body = ''.join(self.formpost(env, start_response))
        self.assertTrue('User-Agent' in self.app.requests[0].headers)
        self.assertEquals(self.app.requests[0].headers['User-Agent'],
                          'FormPost')

    def test_formpost_with_origin(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://redirect', 1024, 10,
            int(time() + 86400), key, user_agent=False)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        env['HTTP_ORIGIN'] = 'http://localhost:5000'
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created',
                                  {'Access-Control-Allow-Origin':
                                   'http://localhost:5000'}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)

        headers = {}

        def start_response(s, h, e=None):
            for k, v in h:
                headers[k] = v
            pass

        body = ''.join(self.formpost(env, start_response))
        self.assertEquals(headers['Access-Control-Allow-Origin'],
                          'http://localhost:5000')

    def test_formpost_with_multiple_keys(self):
        key = 'ernie'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://redirect', 1024, 10,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        # Stick it in X-Account-Meta-Temp-URL-Key-2 and make sure we get it
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', ['bert', key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)

        status = [None]
        headers = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
        body = ''.join(self.formpost(env, start_response))
        print repr(headers)
        self.assertEqual('303 See Other', status[0])
        self.assertEqual(
            'http://redirect?status=201&message=',
            dict(headers[0]).get('Location'))

    def test_redirect(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://redirect', 1024, 10,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, 'http://redirect?status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue(location in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_redirect_with_query(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', 'http://redirect?one=two', 1024, 10,
            int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '303 See Other')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location,
                          'http://redirect?one=two&status=201&message=')
        self.assertEquals(exc_info, None)
        self.assertTrue(location in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_no_redirect(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '201 Created')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('201 Created' in body)
        self.assertEquals(len(self.app.requests), 2)
        self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')
        self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

    def test_no_redirect_expired(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, int(time() - 10), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Form Expired' in body)

    def test_no_redirect_invalid_sig(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        # Change key to invalidate sig
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key + ' is bogus now'])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Invalid Signature' in body)

    def test_no_redirect_with_error(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('XX' + '\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '400 Bad Request')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: invalid starting boundary' in body)

    def test_no_v1(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v2/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Invalid Signature' in body)

    def test_empty_v1(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '//AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Invalid Signature' in body)

    def test_empty_account(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1//container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Invalid Signature' in body)

    def test_wrong_account(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_tst/container', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([
            ('200 Ok', {'x-account-meta-temp-url-key': 'def'}, ''),
            ('201 Created', {}, ''),
            ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Invalid Signature' in body)

    def test_no_container(self):
        key = 'abc'
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test', '', 1024, 10, int(time() + 86400), key)
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '401 Unauthorized')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: Invalid Signature' in body)

    def test_completely_non_int_expires(self):
        key = 'abc'
        expires = int(time() + 86400)
        sig, env, body = self._make_sig_env_body(
            '/v1/AUTH_test/container', '', 1024, 10, expires, key)
        for i, v in enumerate(body):
            if v == str(expires):
                body[i] = 'badvalue'
                break
        env['wsgi.input'] = StringIO('\r\n'.join(body))
        env['swift.account/AUTH_test'] = self._fake_cache_env(
            'AUTH_test', [key])
        self.app = FakeApp(iter([('201 Created', {}, ''),
                                 ('201 Created', {}, '')]))
        self.auth = tempauth.filter_factory({})(self.app)
        self.formpost = formpost.filter_factory({})(self.auth)
        status = [None]
        headers = [None]
        exc_info = [None]

        def start_response(s, h, e=None):
            status[0] = s
            headers[0] = h
            exc_info[0] = e

        body = ''.join(self.formpost(env, start_response))
        status = status[0]
        headers = headers[0]
        exc_info = exc_info[0]
        self.assertEquals(status, '400 Bad Request')
        location = None
        for h, v in headers:
            if h.lower() == 'location':
                location = v
        self.assertEquals(location, None)
        self.assertEquals(exc_info, None)
        self.assertTrue('FormPost: expired not an integer' in body)


if __name__ == '__main__':
    unittest.main()