"Fossies" - the Fresh Open Source Software Archive

Member "senlin-8.0.0/senlin/tests/unit/api/common/test_wsgi.py" (16 Oct 2019, 16998 Bytes) of package /linux/misc/openstack/senlin-8.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_wsgi.py": 6.0.0_vs_7.0.0.

    1 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 # not use this file except in compliance with the License. You may obtain
    3 # a copy of the License at
    4 #
    5 #         http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 # Unless required by applicable law or agreed to in writing, software
    8 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 # License for the specific language governing permissions and limitations
   11 # under the License.
   12 
   13 import socket
   14 
   15 import fixtures
   16 import mock
   17 from oslo_config import cfg
   18 from oslo_utils import encodeutils
   19 import six
   20 import webob
   21 
   22 from senlin.api.common import version_request as vr
   23 from senlin.api.common import wsgi
   24 from senlin.common import exception
   25 from senlin.tests.unit.common import base
   26 
   27 CONF = cfg.CONF
   28 
   29 
   30 class RequestTest(base.SenlinTestCase):
   31 
   32     def test_content_type_missing(self):
   33         request = wsgi.Request.blank('/tests/123')
   34         self.assertRaises(exception.InvalidContentType,
   35                           request.get_content_type, ('application/xml'))
   36 
   37     def test_content_type_unsupported(self):
   38         request = wsgi.Request.blank('/tests/123')
   39         request.headers["Content-Type"] = "text/html"
   40         self.assertRaises(exception.InvalidContentType,
   41                           request.get_content_type, ('application/xml'))
   42 
   43     def test_content_type_with_charset(self):
   44         request = wsgi.Request.blank('/tests/123')
   45         request.headers["Content-Type"] = "application/json; charset=UTF-8"
   46         result = request.get_content_type(('application/json'))
   47         self.assertEqual("application/json", result)
   48 
   49     def test_content_type_from_accept_xml(self):
   50         request = wsgi.Request.blank('/tests/123')
   51         request.headers["Accept"] = "application/xml"
   52         result = request.best_match_content_type()
   53         self.assertEqual("application/json", result)
   54 
   55     def test_content_type_from_accept_json(self):
   56         request = wsgi.Request.blank('/tests/123')
   57         request.headers["Accept"] = "application/json"
   58         result = request.best_match_content_type()
   59         self.assertEqual("application/json", result)
   60 
   61     def test_content_type_from_accept_xml_json(self):
   62         request = wsgi.Request.blank('/tests/123')
   63         request.headers["Accept"] = "application/xml, application/json"
   64         result = request.best_match_content_type()
   65         self.assertEqual("application/json", result)
   66 
   67     def test_content_type_from_accept_json_xml_quality(self):
   68         request = wsgi.Request.blank('/tests/123')
   69         request.headers["Accept"] = ("application/json; q=0.3, "
   70                                      "application/xml; q=0.9")
   71         result = request.best_match_content_type()
   72         self.assertEqual("application/json", result)
   73 
   74     def test_content_type_accept_default(self):
   75         request = wsgi.Request.blank('/tests/123.unsupported')
   76         request.headers["Accept"] = "application/unsupported1"
   77         result = request.best_match_content_type()
   78         self.assertEqual("application/json", result)
   79 
   80 
   81 class ResourceTest(base.SenlinTestCase):
   82 
   83     def test_get_action_args(self):
   84         env = {
   85             'wsgiorg.routing_args': [
   86                 None,
   87                 {
   88                     'controller': None,
   89                     'format': None,
   90                     'action': 'update',
   91                     'id': 12,
   92                 },
   93             ],
   94         }
   95 
   96         expected = {'action': 'update', 'id': 12}
   97         actual = wsgi.Resource(None).get_action_args(env)
   98 
   99         self.assertEqual(expected, actual)
  100 
  101     def test_get_action_args_invalid_index(self):
  102         env = {'wsgiorg.routing_args': []}
  103         expected = {}
  104         actual = wsgi.Resource(None).get_action_args(env)
  105         self.assertEqual(expected, actual)
  106 
  107     def test_get_action_args_del_controller_error(self):
  108         actions = {'format': None,
  109                    'action': 'update',
  110                    'id': 12}
  111         env = {'wsgiorg.routing_args': [None, actions]}
  112         expected = {'action': 'update', 'id': 12}
  113         actual = wsgi.Resource(None).get_action_args(env)
  114         self.assertEqual(expected, actual)
  115 
  116     def test_get_action_args_del_format_error(self):
  117         actions = {'action': 'update', 'id': 12}
  118         env = {'wsgiorg.routing_args': [None, actions]}
  119         expected = {'action': 'update', 'id': 12}
  120         actual = wsgi.Resource(None).get_action_args(env)
  121         self.assertEqual(expected, actual)
  122 
  123     def test_dispatch(self):
  124         class Controller(object):
  125             def index(self, shirt, pants=None):
  126                 return (shirt, pants)
  127 
  128         resource = wsgi.Resource(None)
  129         actual = resource.dispatch(Controller(), 'index', 'on', pants='off')
  130         expected = ('on', 'off')
  131         self.assertEqual(expected, actual)
  132 
  133     def test_dispatch_default(self):
  134         class Controller(object):
  135             def default(self, shirt, pants=None):
  136                 return (shirt, pants)
  137 
  138         resource = wsgi.Resource(None)
  139         actual = resource.dispatch(Controller(), 'index', 'on', pants='off')
  140         expected = ('on', 'off')
  141         self.assertEqual(expected, actual)
  142 
  143     def test_dispatch_no_default(self):
  144         class Controller(object):
  145             def show(self, shirt, pants=None):
  146                 return (shirt, pants)
  147 
  148         resource = wsgi.Resource(None)
  149         self.assertRaises(AttributeError, resource.dispatch, Controller(),
  150                           'index', 'on', pants='off')
  151 
  152     def test_resource_call_error_handle(self):
  153         class Controller(object):
  154             def delete(self, req, identity):
  155                 return (req, identity)
  156 
  157         actions = {'action': 'delete', 'id': 12, 'body': 'data'}
  158         env = {'wsgiorg.routing_args': [None, actions]}
  159         request = wsgi.Request.blank('/tests/123', environ=env)
  160         request.body = encodeutils.safe_encode('{"foo" : "value"}')
  161         resource = wsgi.Resource(Controller())
  162 
  163         # The Resource does not throw webob.HTTPExceptions, since they
  164         # would be considered responses by wsgi and the request flow would end,
  165         # instead they are wrapped so they can reach the fault application
  166         # where they are converted to a JSON response
  167         e = self.assertRaises(exception.HTTPExceptionDisguise,
  168                               resource, request)
  169         self.assertIsInstance(e.exc, webob.exc.HTTPBadRequest)
  170 
  171     @mock.patch.object(wsgi, 'translate_exception')
  172     def test_resource_call_error_handle_localized(self, mock_translate):
  173         class Controller(object):
  174             def delete(self, req, identity):
  175                 return (req, identity)
  176 
  177         def fake_translate_exception(ex, locale):
  178             return translated_ex
  179 
  180         mock_translate.side_effect = fake_translate_exception
  181         actions = {'action': 'delete', 'id': 12, 'body': 'data'}
  182         env = {'wsgiorg.routing_args': [None, actions]}
  183         request = wsgi.Request.blank('/tests/123', environ=env)
  184         request.body = encodeutils.safe_encode('{"foo" : "value"}')
  185         message_es = "No Encontrado"
  186         translated_ex = webob.exc.HTTPBadRequest(message_es)
  187 
  188         resource = wsgi.Resource(Controller())
  189 
  190         e = self.assertRaises(exception.HTTPExceptionDisguise,
  191                               resource, request)
  192         self.assertEqual(message_es, six.text_type(e.exc))
  193 
  194     def test_resource_call_with_version_header(self):
  195         class Controller(object):
  196             def dance(self, req):
  197                 return {'foo': 'bar'}
  198 
  199         actions = {'action': 'dance'}
  200         env = {'wsgiorg.routing_args': [None, actions]}
  201         request = wsgi.Request.blank('/tests/123', environ=env)
  202         request.version_request = vr.APIVersionRequest('1.0')
  203 
  204         resource = wsgi.Resource(Controller())
  205         resp = resource(request)
  206         self.assertEqual('{"foo": "bar"}', encodeutils.safe_decode(resp.body))
  207         self.assertTrue(hasattr(resp, 'headers'))
  208         expected = 'clustering 1.0'
  209         self.assertEqual(expected, resp.headers['OpenStack-API-Version'])
  210         self.assertEqual('OpenStack-API-Version', resp.headers['Vary'])
  211 
  212 
  213 class ControllerTest(base.SenlinTestCase):
  214 
  215     @mock.patch('senlin.rpc.client.EngineClient')
  216     def test_init(self, mock_client):
  217         x_client = mock.Mock()
  218         mock_client.return_value = x_client
  219         data = mock.Mock()
  220 
  221         c = wsgi.Controller(data)
  222 
  223         self.assertEqual(data, c.options)
  224         self.assertEqual(x_client, c.rpc_client)
  225 
  226     def test_default(self):
  227         data = mock.Mock()
  228         c = wsgi.Controller(data)
  229 
  230         self.assertRaises(webob.exc.HTTPNotFound, c.default, mock.Mock())
  231 
  232 
  233 class ResourceExceptionHandlingTest(base.SenlinTestCase):
  234     scenarios = [
  235         ('client_exceptions', dict(
  236             exception=exception.NotAuthenticated,
  237             exception_catch=exception.NotAuthenticated)),
  238         ('webob_bad_request', dict(
  239             exception=webob.exc.HTTPBadRequest,
  240             exception_catch=exception.HTTPExceptionDisguise)),
  241         ('webob_not_found', dict(
  242             exception=webob.exc.HTTPNotFound,
  243             exception_catch=exception.HTTPExceptionDisguise)),
  244     ]
  245 
  246     def test_resource_client_exceptions_dont_log_error(self):
  247         class Controller(object):
  248             def __init__(self, exception_to_raise):
  249                 self.exception_to_raise = exception_to_raise
  250 
  251             def raise_exception(self, req, body):
  252                 raise self.exception_to_raise()
  253         actions = {'action': 'raise_exception', 'body': 'data'}
  254         env = {'wsgiorg.routing_args': [None, actions]}
  255         request = wsgi.Request.blank('/tests/123', environ=env)
  256         request.body = encodeutils.safe_encode('{"foo": "value"}')
  257         resource = wsgi.Resource(Controller(self.exception))
  258         e = self.assertRaises(self.exception_catch, resource, request)
  259         e = e.exc if hasattr(e, 'exc') else e
  260         self.assertNotIn(six.text_type(e), self.LOG.output)
  261 
  262 
  263 class GetSocketTestCase(base.SenlinTestCase):
  264 
  265     def setUp(self):
  266         super(GetSocketTestCase, self).setUp()
  267         self.useFixture(fixtures.MonkeyPatch(
  268             "senlin.api.common.wsgi.get_bind_addr",
  269             lambda x, y: ('192.168.0.13', 1234)))
  270         addr_info_list = [(2, 1, 6, '', ('192.168.0.13', 80)),
  271                           (2, 2, 17, '', ('192.168.0.13', 80)),
  272                           (2, 3, 0, '', ('192.168.0.13', 80))]
  273         self.useFixture(fixtures.MonkeyPatch(
  274             "senlin.api.common.wsgi.socket.getaddrinfo",
  275             lambda *x: addr_info_list))
  276         self.useFixture(fixtures.MonkeyPatch(
  277             "senlin.api.common.wsgi.time.time",
  278             mock.Mock(side_effect=[0, 1, 5, 10, 20, 35])))
  279         wsgi.cfg.CONF.senlin_api.cert_file = '/etc/ssl/cert'
  280         wsgi.cfg.CONF.senlin_api.key_file = '/etc/ssl/key'
  281         wsgi.cfg.CONF.senlin_api.ca_file = '/etc/ssl/ca_cert'
  282         wsgi.cfg.CONF.senlin_api.tcp_keepidle = 600
  283 
  284     def test_correct_configure_socket(self):
  285         mock_socket = mock.Mock()
  286         self.useFixture(fixtures.MonkeyPatch(
  287             'senlin.api.common.wsgi.ssl.wrap_socket',
  288             mock_socket))
  289         self.useFixture(fixtures.MonkeyPatch(
  290             'senlin.api.common.wsgi.eventlet.listen',
  291             lambda *x, **y: mock_socket))
  292         server = wsgi.Server(name='senlin-api', conf=cfg.CONF.senlin_api)
  293         server.default_port = 1234
  294         server.configure_socket()
  295         self.assertIn(mock.call.setsockopt(socket.SOL_SOCKET,
  296                                            socket.SO_REUSEADDR, 1),
  297                       mock_socket.mock_calls)
  298         self.assertIn(mock.call.setsockopt(socket.SOL_SOCKET,
  299                                            socket.SO_KEEPALIVE, 1),
  300                       mock_socket.mock_calls)
  301         if hasattr(socket, 'TCP_KEEPIDLE'):
  302             self.assertIn(mock.call().setsockopt(
  303                 socket.IPPROTO_TCP,
  304                 socket.TCP_KEEPIDLE,
  305                 wsgi.cfg.CONF.senlin_api.tcp_keepidle), mock_socket.mock_calls)
  306 
  307     def test_get_socket_without_all_ssl_reqs(self):
  308         wsgi.cfg.CONF.senlin_api.key_file = None
  309         self.assertRaises(RuntimeError,
  310                           wsgi.get_socket, wsgi.cfg.CONF.senlin_api, 1234)
  311 
  312     def test_get_socket_with_bind_problems(self):
  313         self.useFixture(fixtures.MonkeyPatch(
  314             'senlin.api.common.wsgi.eventlet.listen',
  315             mock.Mock(side_effect=(
  316                 [wsgi.socket.error(socket.errno.EADDRINUSE)] * 3 + [None]))))
  317         self.useFixture(fixtures.MonkeyPatch(
  318             'senlin.api.common.wsgi.ssl.wrap_socket',
  319             lambda *x, **y: None))
  320 
  321         self.assertRaises(RuntimeError,
  322                           wsgi.get_socket, wsgi.cfg.CONF.senlin_api, 1234)
  323 
  324     def test_get_socket_with_unexpected_socket_errno(self):
  325         self.useFixture(fixtures.MonkeyPatch(
  326             'senlin.api.common.wsgi.eventlet.listen',
  327             mock.Mock(side_effect=wsgi.socket.error(socket.errno.ENOMEM))))
  328         self.useFixture(fixtures.MonkeyPatch(
  329             'senlin.api.common.wsgi.ssl.wrap_socket',
  330             lambda *x, **y: None))
  331         self.assertRaises(wsgi.socket.error, wsgi.get_socket,
  332                           wsgi.cfg.CONF.senlin_api, 1234)
  333 
  334 
  335 class FakeController(wsgi.Controller):
  336 
  337     @wsgi.Controller.api_version('2.0')
  338     def index(self, req):
  339         return {'foo': 'bar'}
  340 
  341     def foo(self, req):
  342         return {'bar': 'zoo'}
  343 
  344     @wsgi.Controller.api_version('2.0', '3.0')
  345     def dance(self, req):
  346         return {'score': 100}
  347 
  348     @wsgi.Controller.api_version('4.0')   # noqa
  349     def dance(self, req):
  350         return {'score': 60}
  351 
  352 
  353 class MicroversionTest(base.SenlinTestCase):
  354 
  355     def test_versioned_request_empty(self):
  356         data = mock.Mock()
  357         request = wsgi.Request.blank('/tests/123')
  358         request.version_request = vr.APIVersionRequest('1.0')
  359         c = FakeController(data)
  360 
  361         ex = self.assertRaises(exception.MethodVersionNotFound,
  362                                c.index, request)
  363         self.assertEqual("API version '1.0' is not supported on "
  364                          "this method.", six.text_type(ex))
  365 
  366         res = c.foo(request)
  367         self.assertEqual({'bar': 'zoo'}, res)
  368 
  369         ex = self.assertRaises(exception.MethodVersionNotFound,
  370                                c.dance, request)
  371         self.assertEqual("API version '1.0' is not supported on "
  372                          "this method.", six.text_type(ex))
  373 
  374     def test_versioned_request_lower(self):
  375         data = mock.Mock()
  376         request = wsgi.Request.blank('/tests/123')
  377         request.version_request = vr.APIVersionRequest('2.0')
  378         c = FakeController(data)
  379 
  380         res = c.index(request)
  381         self.assertEqual({'foo': 'bar'}, res)
  382 
  383         res = c.foo(request)
  384         self.assertEqual({'bar': 'zoo'}, res)
  385 
  386         res = c.dance(request)
  387         self.assertEqual({'score': 100}, res)
  388 
  389     def test_versioned_request_middle(self):
  390         data = mock.Mock()
  391         request = wsgi.Request.blank('/tests/123')
  392         request.version_request = vr.APIVersionRequest('2.5')
  393         c = FakeController(data)
  394 
  395         res = c.index(request)
  396         self.assertEqual({'foo': 'bar'}, res)
  397 
  398         res = c.foo(request)
  399         self.assertEqual({'bar': 'zoo'}, res)
  400 
  401         res = c.dance(request)
  402         self.assertEqual({'score': 100}, res)
  403 
  404     def test_versioned_request_upper(self):
  405         data = mock.Mock()
  406         request = wsgi.Request.blank('/tests/123')
  407         request.version_request = vr.APIVersionRequest('3.0')
  408         c = FakeController(data)
  409 
  410         res = c.index(request)
  411         self.assertEqual({'foo': 'bar'}, res)
  412 
  413         res = c.foo(request)
  414         self.assertEqual({'bar': 'zoo'}, res)
  415 
  416         res = c.dance(request)
  417         self.assertEqual({'score': 100}, res)
  418 
  419     def test_versioned_request_too_high(self):
  420         data = mock.Mock()
  421         request = wsgi.Request.blank('/tests/123')
  422         request.version_request = vr.APIVersionRequest('3.5')
  423         c = FakeController(data)
  424 
  425         res = c.index(request)
  426         self.assertEqual({'foo': 'bar'}, res)
  427 
  428         res = c.foo(request)
  429         self.assertEqual({'bar': 'zoo'}, res)
  430 
  431         ex = self.assertRaises(exception.MethodVersionNotFound,
  432                                c.dance, request)
  433         self.assertEqual("API version '3.5' is not supported on "
  434                          "this method.", six.text_type(ex))
  435 
  436     def test_versioned_request_inner_functions(self):
  437         data = mock.Mock()
  438         request = wsgi.Request.blank('/tests/123')
  439         request.version_request = vr.APIVersionRequest('3.0')
  440         c = FakeController(data)
  441 
  442         res = c.dance(request)
  443         self.assertEqual({'score': 100}, res)
  444 
  445         request.version_request = vr.APIVersionRequest('4.0')
  446         res = c.dance(request)
  447         self.assertEqual({'score': 60}, res)