# 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 slogging 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)
self.assertEquals(req.headers, req2.headers)
self.assertFalse(req.headers is req2.headers)
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()