swift/test/unit/common/test_internal_proxy.py

# Copyright (c) 2010-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.

# TODO: Tests

import unittest
import webob
import tempfile
import json

from swift.common import internal_proxy

class DumbBaseApplicationFactory(object):

    def __init__(self, status_codes, body=''):
        self.status_codes = status_codes[:]
        self.body = body

    def __call__(self, *a, **kw):
        app = DumbBaseApplication(*a, **kw)
        app.status_codes = self.status_codes
        try:
            app.default_status_code = self.status_codes[-1]
        except IndexError:
            app.default_status_code = 200
        app.body = self.body
        return app

class DumbBaseApplication(object):

    def __init__(self, *a, **kw):
        self.status_codes = []
        self.default_status_code = 200
        self.call_count = 0
        self.body = ''

    def handle_request(self, req):
        self.call_count += 1
        req.path_info_pop()
        if isinstance(self.body, list):
            try:
                body = self.body.pop(0)
            except IndexError:
                body = ''
        else:
            body = self.body
        resp = webob.Response(request=req, body=body,
                              conditional_response=True)
        try:
            resp.status_int = self.status_codes.pop(0)
        except IndexError:
            resp.status_int = self.default_status_code
        return resp

    def update_request(self, req):
        return req


class TestInternalProxy(unittest.TestCase):

    def test_webob_request_copy(self):
        req = webob.Request.blank('/')
        req2 = internal_proxy.webob_request_copy(req)
        self.assertEquals(req.path, req2.path)
        self.assertEquals(req.path_info, req2.path_info)
        self.assertFalse(req is req2)

    def test_handle_request(self):
        status_codes = [200]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy()
        req = webob.Request.blank('/')
        orig_req = internal_proxy.webob_request_copy(req)
        resp = p._handle_request(req)
        self.assertEquals(req.path_info, orig_req.path_info)

    def test_handle_request_with_retries(self):
        status_codes = [500, 200]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy(retries=3)
        req = webob.Request.blank('/')
        orig_req = internal_proxy.webob_request_copy(req)
        resp = p._handle_request(req)
        self.assertEquals(req.path_info, orig_req.path_info)
        self.assertEquals(p.upload_app.call_count, 2)
        self.assertEquals(resp.status_int, 200)

    def test_get_object(self):
        status_codes = [200]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy()
        code, body = p.get_object('a', 'c', 'o')
        body = ''.join(body)
        self.assertEquals(code, 200)
        self.assertEquals(body, '')

    def test_create_container(self):
        status_codes = [200]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy()
        resp = p.create_container('a', 'c')
        self.assertTrue(resp)

    def test_handle_request_with_retries_all_error(self):
        status_codes = [500, 500, 500, 500, 500]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy(retries=3)
        req = webob.Request.blank('/')
        orig_req = internal_proxy.webob_request_copy(req)
        resp = p._handle_request(req)
        self.assertEquals(req.path_info, orig_req.path_info)
        self.assertEquals(p.upload_app.call_count, 3)
        self.assertEquals(resp.status_int, 500)

    def test_get_container_list_empty(self):
        status_codes = [200]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes, body='[]')
        p = internal_proxy.InternalProxy()
        resp = p.get_container_list('a', 'c')
        self.assertEquals(resp, [])

    def test_get_container_list_no_body(self):
        status_codes = [204]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes, body='')
        p = internal_proxy.InternalProxy()
        resp = p.get_container_list('a', 'c')
        self.assertEquals(resp, [])

    def test_get_container_list_full_listing(self):
        status_codes = [200, 200]
        obj_a = dict(name='foo', hash='foo', bytes=3,
                     content_type='text/plain', last_modified='2011/01/01')
        obj_b = dict(name='bar', hash='bar', bytes=3,
                     content_type='text/plain', last_modified='2011/01/01')
        body = [json.dumps([obj_a]), json.dumps([obj_b]), json.dumps([])]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes, body=body)
        p = internal_proxy.InternalProxy()
        resp = p.get_container_list('a', 'c')
        expected = ['foo', 'bar']
        self.assertEquals([x['name'] for x in resp], expected)

    def test_get_container_list_full(self):
        status_codes = [204]
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes, body='')
        p = internal_proxy.InternalProxy()
        resp = p.get_container_list('a', 'c', marker='a', end_marker='b',
                                    limit=100, prefix='/', delimiter='.')
        self.assertEquals(resp, [])

    def test_upload_file(self):
        status_codes = [200, 200]  # container PUT + object PUT
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy()
        with tempfile.NamedTemporaryFile() as file_obj:
            resp = p.upload_file(file_obj.name, 'a', 'c', 'o')
        self.assertTrue(resp)

    def test_upload_file_with_retries(self):
        status_codes = [200, 500, 200]  # container PUT + error + object PUT
        internal_proxy.BaseApplication = DumbBaseApplicationFactory(
                                            status_codes)
        p = internal_proxy.InternalProxy(retries=3)
        with tempfile.NamedTemporaryFile() as file_obj:
            resp = p.upload_file(file_obj, 'a', 'c', 'o')
        self.assertTrue(resp)
        self.assertEquals(p.upload_app.call_count, 3)


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