272 lines
9.9 KiB
Python
272 lines
9.9 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2010-2011 OpenStack, LLC
|
|
# All Rights Reserved.
|
|
#
|
|
# 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
|
|
import webob
|
|
|
|
from glance.common import wsgi
|
|
from glance.common import utils
|
|
from glance.common import exception
|
|
|
|
|
|
class RequestTest(unittest.TestCase):
|
|
def test_content_type_missing(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
self.assertRaises(exception.InvalidContentType,
|
|
request.get_content_type, ('application/xml'))
|
|
|
|
def test_content_type_unsupported(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Content-Type"] = "text/html"
|
|
self.assertRaises(exception.InvalidContentType,
|
|
request.get_content_type, ('application/xml'))
|
|
|
|
def test_content_type_with_charset(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Content-Type"] = "application/json; charset=UTF-8"
|
|
result = request.get_content_type(('application/json'))
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_from_accept_xml(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Accept"] = "application/xml"
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_from_accept_json(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Accept"] = "application/json"
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_from_accept_xml_json(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Accept"] = "application/xml, application/json"
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_from_accept_json_xml_quality(self):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Accept"] = \
|
|
"application/json; q=0.3, application/xml; q=0.9"
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_accept_default(self):
|
|
request = wsgi.Request.blank('/tests/123.unsupported')
|
|
request.headers["Accept"] = "application/unsupported1"
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
|
|
class ResourceTest(unittest.TestCase):
|
|
def test_get_action_args(self):
|
|
env = {
|
|
'wsgiorg.routing_args': [
|
|
None,
|
|
{
|
|
'controller': None,
|
|
'format': None,
|
|
'action': 'update',
|
|
'id': 12,
|
|
},
|
|
],
|
|
}
|
|
|
|
expected = {'action': 'update', 'id': 12}
|
|
actual = wsgi.Resource(None, None, None).get_action_args(env)
|
|
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_dispatch(self):
|
|
class Controller(object):
|
|
def index(self, shirt, pants=None):
|
|
return (shirt, pants)
|
|
|
|
resource = wsgi.Resource(None, None, None)
|
|
actual = resource.dispatch(Controller(), 'index', 'on', pants='off')
|
|
expected = ('on', 'off')
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_dispatch_default(self):
|
|
class Controller(object):
|
|
def default(self, shirt, pants=None):
|
|
return (shirt, pants)
|
|
|
|
resource = wsgi.Resource(None, None, None)
|
|
actual = resource.dispatch(Controller(), 'index', 'on', pants='off')
|
|
expected = ('on', 'off')
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_dispatch_no_default(self):
|
|
class Controller(object):
|
|
def show(self, shirt, pants=None):
|
|
return (shirt, pants)
|
|
|
|
resource = wsgi.Resource(None, None, None)
|
|
self.assertRaises(AttributeError, resource.dispatch, Controller(),
|
|
'index', 'on', pants='off')
|
|
|
|
|
|
class JSONResponseSerializerTest(unittest.TestCase):
|
|
def test_to_json(self):
|
|
fixture = {"key": "value"}
|
|
expected = '{"key": "value"}'
|
|
actual = wsgi.JSONResponseSerializer().to_json(fixture)
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_default(self):
|
|
fixture = {"key": "value"}
|
|
response = webob.Response()
|
|
wsgi.JSONResponseSerializer().default(response, fixture)
|
|
self.assertEqual(response.status_int, 200)
|
|
self.assertEqual(response.content_type, 'application/json')
|
|
self.assertEqual(response.body, '{"key": "value"}')
|
|
|
|
|
|
class JSONRequestDeserializerTest(unittest.TestCase):
|
|
def test_has_body_no_content_length(self):
|
|
request = wsgi.Request.blank('/')
|
|
request.method = 'POST'
|
|
request.body = 'asdf'
|
|
request.headers.pop('Content-Length')
|
|
self.assertFalse(wsgi.JSONRequestDeserializer().has_body(request))
|
|
|
|
def test_has_body_zero_content_length(self):
|
|
request = wsgi.Request.blank('/')
|
|
request.method = 'POST'
|
|
request.body = 'asdf'
|
|
request.headers['Content-Length'] = 0
|
|
self.assertFalse(wsgi.JSONRequestDeserializer().has_body(request))
|
|
|
|
def test_has_body_has_content_length(self):
|
|
request = wsgi.Request.blank('/')
|
|
request.method = 'POST'
|
|
request.body = 'asdf'
|
|
self.assertTrue('Content-Length' in request.headers)
|
|
self.assertTrue(wsgi.JSONRequestDeserializer().has_body(request))
|
|
|
|
def test_no_body_no_content_length(self):
|
|
request = wsgi.Request.blank('/')
|
|
self.assertFalse(wsgi.JSONRequestDeserializer().has_body(request))
|
|
|
|
def test_from_json(self):
|
|
fixture = '{"key": "value"}'
|
|
expected = {"key": "value"}
|
|
actual = wsgi.JSONRequestDeserializer().from_json(fixture)
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_default_no_body(self):
|
|
request = wsgi.Request.blank('/')
|
|
actual = wsgi.JSONRequestDeserializer().default(request)
|
|
expected = {}
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_default_with_body(self):
|
|
request = wsgi.Request.blank('/')
|
|
request.method = 'POST'
|
|
request.body = '{"key": "value"}'
|
|
actual = wsgi.JSONRequestDeserializer().default(request)
|
|
expected = {"body": {"key": "value"}}
|
|
self.assertEqual(actual, expected)
|
|
|
|
|
|
class TestHelpers(unittest.TestCase):
|
|
|
|
def test_headers_are_unicode(self):
|
|
"""
|
|
Verifies that the headers returned by conversion code are unicode.
|
|
|
|
Headers are passed via http in non-testing mode, which automatically
|
|
converts them to unicode. Verifying that the method does the
|
|
conversion proves that we aren't passing data that works in tests
|
|
but will fail in production.
|
|
"""
|
|
fixture = {'name': 'fake public image',
|
|
'is_public': True,
|
|
'type': 'kernel',
|
|
'size': 19,
|
|
'location': "file:///tmp/glance-tests/2",
|
|
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
|
|
headers = utils.image_meta_to_http_headers(fixture)
|
|
for k, v in headers.iteritems():
|
|
self.assert_(isinstance(v, unicode), "%s is not unicode" % v)
|
|
|
|
def test_data_passed_properly_through_headers(self):
|
|
"""
|
|
Verifies that data is the same after being passed through headers
|
|
"""
|
|
fixture = {'name': 'fake public image',
|
|
'is_public': True,
|
|
'deleted': False,
|
|
'type': 'kernel',
|
|
'name': None,
|
|
'size': 19,
|
|
'location': "file:///tmp/glance-tests/2",
|
|
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
|
|
headers = utils.image_meta_to_http_headers(fixture)
|
|
|
|
class FakeResponse():
|
|
pass
|
|
|
|
response = FakeResponse()
|
|
response.headers = headers
|
|
result = utils.get_image_meta_from_headers(response)
|
|
for k, v in fixture.iteritems():
|
|
self.assertEqual(v, result[k])
|
|
|
|
def test_boolean_header_values(self):
|
|
"""
|
|
Tests that boolean headers like is_public can be set
|
|
to True if any of ('True', 'On', 1) is provided, case-insensitive
|
|
"""
|
|
fixtures = [{'is_public': True},
|
|
{'is_public': 'True'},
|
|
{'is_public': 'true'}]
|
|
|
|
expected = {'is_public': True}
|
|
|
|
class FakeResponse():
|
|
pass
|
|
|
|
for fixture in fixtures:
|
|
headers = utils.image_meta_to_http_headers(fixture)
|
|
|
|
response = FakeResponse()
|
|
response.headers = headers
|
|
result = utils.get_image_meta_from_headers(response)
|
|
for k, v in expected.items():
|
|
self.assertEqual(v, result[k])
|
|
|
|
# Ensure False for other values...
|
|
fixtures = [{'is_public': False},
|
|
{'is_public': 'Off'},
|
|
{'is_public': 'on'},
|
|
{'is_public': '1'},
|
|
{'is_public': 'False'}]
|
|
|
|
expected = {'is_public': False}
|
|
|
|
for fixture in fixtures:
|
|
headers = utils.image_meta_to_http_headers(fixture)
|
|
|
|
response = FakeResponse()
|
|
response.headers = headers
|
|
result = utils.get_image_meta_from_headers(response)
|
|
for k, v in expected.items():
|
|
self.assertEqual(v, result[k])
|