text: # 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):
          # Test that we are actually invoking language negotiation by webop
          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')
  
          # If the Accept-Leader is an unknown language, missing or empty,
          # the best match locale should be None
          request.headers['Accept-Language'] = 'unknown-language'
          language = request.best_match_language()
          self.assertEqual(language, None)
          request.headers['Accept-Language'] = ''
          language = request.best_match_language()
          self.assertEqual(language, None)
          request.headers.pop('Accept-Language')
          language = request.best_match_language()
          self.assertEqual(language, None)
  
  
  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': {
                  'type': 'TestException',
                  'message': msg,
                  'detail': ''}}}
          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': {
                  'type': 'TestException',
                  'message': msg,
                  'detail': ''}}}
          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.get_localized_message')
      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': {
                  'type': 'TestException',
                  'message': msg,
                  'detail': ''}}}
          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': {
                  'type': 'TestException',
                  'message': msg,
                  'detail': ''}}}
          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.get_localized_message')
      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)
  