d12fcadaa0
Add support for doing language resolution for a request, based on the Accept-Language HTTP header. Using the lazy gettext functionality, from oslo gettextutils, it is possible to use the resolved language to translate exception messages to the user requested language and return that translation from the API. Partially implements bp user-locale-api. Change-Id: I63edc8463836bfff257daa8a2c66ed5d3a444254
324 lines
14 KiB
Python
324 lines
14 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright (c) 2012 Intel Corporation.
|
|
# 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.
|
|
#
|
|
# @author: Zhongyue Luo, Intel Corporation.
|
|
#
|
|
|
|
import mock
|
|
from webob import exc
|
|
import webtest
|
|
|
|
from neutron.api.v2 import resource as wsgi_resource
|
|
from neutron.common import exceptions as q_exc
|
|
from neutron import context
|
|
from neutron.openstack.common import gettextutils
|
|
from neutron.tests import base
|
|
from neutron import wsgi
|
|
|
|
|
|
class RequestTestCase(base.BaseTestCase):
|
|
def setUp(self):
|
|
super(RequestTestCase, self).setUp()
|
|
self.req = wsgi_resource.Request({'foo': 'bar'})
|
|
|
|
def test_content_type_missing(self):
|
|
request = wsgi.Request.blank('/tests/123', method='POST')
|
|
request.body = "<body />"
|
|
self.assertEqual(None, request.get_content_type())
|
|
|
|
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()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_from_accept(self):
|
|
for content_type in ('application/xml',
|
|
'application/json'):
|
|
request = wsgi.Request.blank('/tests/123')
|
|
request.headers["Accept"] = content_type
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, content_type)
|
|
|
|
def test_content_type_from_accept_best(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")
|
|
|
|
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/xml")
|
|
|
|
def test_content_type_from_query_extension(self):
|
|
request = wsgi.Request.blank('/tests/123.xml')
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/xml")
|
|
|
|
request = wsgi.Request.blank('/tests/123.json')
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
request = wsgi.Request.blank('/tests/123.invalid')
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/json")
|
|
|
|
def test_content_type_accept_and_query_extension(self):
|
|
request = wsgi.Request.blank('/tests/123.xml')
|
|
request.headers["Accept"] = "application/json"
|
|
result = request.best_match_content_type()
|
|
self.assertEqual(result, "application/xml")
|
|
|
|
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")
|
|
|
|
def test_context_with_neutron_context(self):
|
|
ctxt = context.Context('fake_user', 'fake_tenant')
|
|
self.req.environ['neutron.context'] = ctxt
|
|
self.assertEqual(self.req.context, ctxt)
|
|
|
|
def test_context_without_neutron_context(self):
|
|
self.assertTrue(self.req.context.is_admin)
|
|
|
|
def test_best_match_language(self):
|
|
# Here we test that we are actually invoking language negotiation
|
|
# by webop and also that the default locale always available is en-US
|
|
request = wsgi.Request.blank('/')
|
|
gettextutils.get_available_languages = mock.MagicMock()
|
|
gettextutils.get_available_languages.return_value = ['known-language',
|
|
'es', 'zh']
|
|
request.headers['Accept-Language'] = 'known-language'
|
|
language = request.best_match_language()
|
|
self.assertEqual(language, 'known-language')
|
|
request.headers['Accept-Language'] = 'unknown-language'
|
|
language = request.best_match_language()
|
|
self.assertEqual(language, 'en_US')
|
|
|
|
|
|
class ResourceTestCase(base.BaseTestCase):
|
|
|
|
def test_unmapped_neutron_error_with_json(self):
|
|
msg = u'\u7f51\u7edc'
|
|
|
|
class TestException(q_exc.NeutronException):
|
|
message = msg
|
|
expected_res = {'body': {'NeutronError': msg}}
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = TestException()
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'json'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPInternalServerError.code)
|
|
self.assertEqual(wsgi.JSONDeserializer().deserialize(res.body),
|
|
expected_res)
|
|
|
|
def test_unmapped_neutron_error_with_xml(self):
|
|
msg = u'\u7f51\u7edc'
|
|
|
|
class TestException(q_exc.NeutronException):
|
|
message = msg
|
|
expected_res = {'body': {'NeutronError': msg}}
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = TestException()
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'xml'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPInternalServerError.code)
|
|
self.assertEqual(wsgi.XMLDeserializer().deserialize(res.body),
|
|
expected_res)
|
|
|
|
@mock.patch('neutron.openstack.common.gettextutils.Message.data',
|
|
new_callable=mock.PropertyMock)
|
|
def test_unmapped_neutron_error_localized(self, mock_translation):
|
|
gettextutils.install('blaa', lazy=True)
|
|
msg_translation = 'Translated error'
|
|
mock_translation.return_value = msg_translation
|
|
msg = _('Unmapped error')
|
|
|
|
class TestException(q_exc.NeutronException):
|
|
message = msg
|
|
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = TestException()
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'json'})}
|
|
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPInternalServerError.code)
|
|
self.assertIn(msg_translation,
|
|
str(wsgi.JSONDeserializer().deserialize(res.body)))
|
|
|
|
def test_mapped_neutron_error_with_json(self):
|
|
msg = u'\u7f51\u7edc'
|
|
|
|
class TestException(q_exc.NeutronException):
|
|
message = msg
|
|
expected_res = {'body': {'NeutronError': msg}}
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = TestException()
|
|
|
|
faults = {TestException: exc.HTTPGatewayTimeout}
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller,
|
|
faults=faults))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'json'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPGatewayTimeout.code)
|
|
self.assertEqual(wsgi.JSONDeserializer().deserialize(res.body),
|
|
expected_res)
|
|
|
|
def test_mapped_neutron_error_with_xml(self):
|
|
msg = u'\u7f51\u7edc'
|
|
|
|
class TestException(q_exc.NeutronException):
|
|
message = msg
|
|
expected_res = {'body': {'NeutronError': msg}}
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = TestException()
|
|
|
|
faults = {TestException: exc.HTTPGatewayTimeout}
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller,
|
|
faults=faults))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'xml'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPGatewayTimeout.code)
|
|
self.assertEqual(wsgi.XMLDeserializer().deserialize(res.body),
|
|
expected_res)
|
|
|
|
@mock.patch('neutron.openstack.common.gettextutils.Message.data',
|
|
new_callable=mock.PropertyMock)
|
|
def test_mapped_neutron_error_localized(self, mock_translation):
|
|
gettextutils.install('blaa', lazy=True)
|
|
msg_translation = 'Translated error'
|
|
mock_translation.return_value = msg_translation
|
|
msg = _('Unmapped error')
|
|
|
|
class TestException(q_exc.NeutronException):
|
|
message = msg
|
|
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = TestException()
|
|
faults = {TestException: exc.HTTPGatewayTimeout}
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller,
|
|
faults=faults))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'json'})}
|
|
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPGatewayTimeout.code)
|
|
self.assertIn(msg_translation,
|
|
str(wsgi.JSONDeserializer().deserialize(res.body)))
|
|
|
|
def test_http_error(self):
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = exc.HTTPGatewayTimeout()
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPGatewayTimeout.code)
|
|
|
|
def test_unhandled_error_with_json(self):
|
|
expected_res = {'body': {'NeutronError':
|
|
_('Request Failed: internal server error '
|
|
'while processing your request.')}}
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = Exception()
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'json'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPInternalServerError.code)
|
|
self.assertEqual(wsgi.JSONDeserializer().deserialize(res.body),
|
|
expected_res)
|
|
|
|
def test_unhandled_error_with_xml(self):
|
|
expected_res = {'body': {'NeutronError':
|
|
_('Request Failed: internal server error '
|
|
'while processing your request.')}}
|
|
controller = mock.MagicMock()
|
|
controller.test.side_effect = Exception()
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test',
|
|
'format': 'xml'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPInternalServerError.code)
|
|
self.assertEqual(wsgi.XMLDeserializer().deserialize(res.body),
|
|
expected_res)
|
|
|
|
def test_status_200(self):
|
|
controller = mock.MagicMock()
|
|
controller.test = lambda request: {'foo': 'bar'}
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test'})}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_status_204(self):
|
|
controller = mock.MagicMock()
|
|
controller.test = lambda request: {'foo': 'bar'}
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'delete'})}
|
|
res = resource.delete('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, 204)
|
|
|
|
def test_no_route_args(self):
|
|
controller = mock.MagicMock()
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {}
|
|
res = resource.get('', extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPInternalServerError.code)
|
|
|
|
def test_post_with_body(self):
|
|
controller = mock.MagicMock()
|
|
controller.test = lambda request, body: {'foo': 'bar'}
|
|
|
|
resource = webtest.TestApp(wsgi_resource.Resource(controller))
|
|
|
|
environ = {'wsgiorg.routing_args': (None, {'action': 'test'})}
|
|
res = resource.post('', params='{"key": "val"}',
|
|
extra_environ=environ, expect_errors=True)
|
|
self.assertEqual(res.status_int, 200)
|