# Copyright 2016 NTT DATA # 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 inspect import mock from six.moves import http_client as http import testscenarios import webob from oslo_serialization import jsonutils from masakari.api import api_version_request as api_version from masakari.api.openstack import extensions from masakari.api.openstack import wsgi from masakari.api import versioned_method from masakari import exception from masakari import test from masakari.tests.unit.api.openstack import fakes from masakari.tests.unit import matchers class MicroversionedTest(testscenarios.WithScenarios, test.NoDBTestCase): header_name = 'OpenStack-API-Version' def _make_microversion_header(self, value): return {self.header_name: value} class RequestTest(MicroversionedTest): def test_content_type_missing(self): request = wsgi.Request.blank('/tests/123', method='POST') request.body = b"" self.assertIsNone(request.get_content_type()) def test_content_type_unsupported(self): request = wsgi.Request.blank('/tests/123', method='POST') request.headers["Content-Type"] = "text/html" request.body = b"asdf
" self.assertRaises(exception.InvalidContentType, 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_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_from_request(self): self.stub_out('masakari.i18n.get_available_languages', fakes.fake_get_available_languages) request = wsgi.Request.blank('/') accepted = 'bogus;q=1.1, en-gb;q=0.7,en-us,en;q=.5,*;q=.7' request.headers = {'Accept-Language': accepted} self.assertEqual(request.best_match_language(), 'en_US') def test_asterisk(self): # asterisk should match first available if there # are not any other available matches self.stub_out('masakari.i18n.get_available_languages', fakes.fake_get_available_languages) request = wsgi.Request.blank('/') accepted = '*,es;q=.5' request.headers = {'Accept-Language': accepted} self.assertEqual(request.best_match_language(), 'en_GB') def test_prefix(self): self.stub_out('masakari.i18n.get_available_languages', fakes.fake_get_available_languages) request = wsgi.Request.blank('/') accepted = 'zh' request.headers = {'Accept-Language': accepted} self.assertEqual(request.best_match_language(), 'zh_CN') def test_secondary(self): self.stub_out('masakari.i18n.get_available_languages', fakes.fake_get_available_languages) request = wsgi.Request.blank('/') accepted = 'nn,en-gb;q=.5' request.headers = {'Accept-Language': accepted} self.assertEqual(request.best_match_language(), 'en_GB') def test_none_found(self): self.stub_out('masakari.i18n.get_available_languages', fakes.fake_get_available_languages) request = wsgi.Request.blank('/') accepted = 'nb-no' request.headers = {'Accept-Language': accepted} self.assertIsNone(request.best_match_language()) def test_no_lang_header(self): self.stub_out('masakari.i18n.get_available_languages', fakes.fake_get_available_languages) request = wsgi.Request.blank('/') accepted = '' request.headers = {'Accept-Language': accepted} self.assertIsNone(request.best_match_language()) def test_api_version_request_header_none(self): request = wsgi.Request.blank('/') request.set_api_version_request() self.assertEqual(api_version.APIVersionRequest( api_version.DEFAULT_API_VERSION), request.api_version_request) @mock.patch("masakari.api.api_version_request.max_api_version") def test_api_version_request_header(self, mock_maxver): mock_maxver.return_value = api_version.APIVersionRequest("1.0") request = wsgi.Request.blank('/') request.headers = self._make_microversion_header('1.0') request.set_api_version_request() self.assertEqual(api_version.APIVersionRequest("1.0"), request.api_version_request) class ActionDispatcherTest(test.NoDBTestCase): def test_dispatch(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: 'pants' self.assertEqual(serializer.dispatch({}, action='create'), 'pants') def test_dispatch_action_None(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: 'pants' serializer.default = lambda x: 'trousers' self.assertEqual(serializer.dispatch({}, action=None), 'trousers') def test_dispatch_default(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: 'pants' serializer.default = lambda x: 'trousers' self.assertEqual(serializer.dispatch({}, action='update'), 'trousers') class JSONDictSerializerTest(test.NoDBTestCase): def test_json(self): input_dict = dict(segments=dict(a=(2, 3))) expected_json = '{"segments":{"a":[2,3]}}' serializer = wsgi.JSONDictSerializer() result = serializer.serialize(input_dict) result = result.replace('\n', '').replace(' ', '') self.assertEqual(result, expected_json) class JSONDeserializerTest(test.NoDBTestCase): def test_json(self): data = """{"a": { "a1": "1", "a2": "2", "bs": ["1", "2", "3", {"c": {"c1": "1"}}], "d": {"e": "1"}, "f": "1"}}""" as_dict = { 'body': { 'a': { 'a1': '1', 'a2': '2', 'bs': ['1', '2', '3', {'c': {'c1': '1'}}], 'd': {'e': '1'}, 'f': '1', }, }, } deserializer = wsgi.JSONDeserializer() self.assertEqual(deserializer.deserialize(data), as_dict) def test_json_valid_utf8(self): data = b"""{"segment": {"recovery_method": "auto", "name": "\xe6\xa6\x82\xe5\xbf\xb5", "service_type": "COMPUTE_HOST" }} """ as_dict = { 'body': { u'segment': {u'recovery_method': 'auto', u'name': u'\u6982\u5ff5', u'service_type': u'COMPUTE_HOST' } } } deserializer = wsgi.JSONDeserializer() self.assertEqual(deserializer.deserialize(data), as_dict) def test_json_invalid_utf8(self): """Send invalid utf-8 to JSONDeserializer.""" data = b"""{"segment": { "name": "\xf0\x28\x8c\x28", "recovery_method": "auto", "description": "compute hosts with shared storage enabled." "service_type": "COMPUTE_HOST"}} """ deserializer = wsgi.JSONDeserializer() self.assertRaises(exception.MalformedRequestBody, deserializer.deserialize, data) class ResourceTest(MicroversionedTest): def get_req_id_header_name(self, request): return 'x-openstack-request-id' def test_resource_call_with_method_get(self): class Controller(object): def index(self, req): return 'success' app = fakes.TestRouter(Controller()) # the default method is GET req = webob.Request.blank('/tests') response = req.get_response(app) self.assertEqual(b'success', response.body) self.assertEqual(response.status_int, http.OK) req.body = b'{"body": {"key": "value"}}' response = req.get_response(app) self.assertEqual(b'success', response.body) self.assertEqual(response.status_int, http.OK) req.content_type = 'application/json' response = req.get_response(app) self.assertEqual(b'success', response.body) self.assertEqual(response.status_int, http.OK) def test_resource_call_with_method_post(self): class Controller(object): @extensions.expected_errors(http.BAD_REQUEST) def create(self, req, body): if expected_body != body: msg = "The request body invalid" raise webob.exc.HTTPBadRequest(explanation=msg) return "success" # verify the method: POST app = fakes.TestRouter(Controller()) req = webob.Request.blank('/tests', method="POST", content_type='application/json') req.body = b'{"body": {"key": "value"}}' expected_body = {'body': { "key": "value" } } response = req.get_response(app) self.assertEqual(response.status_int, http.OK) self.assertEqual(b'success', response.body) # verify without body expected_body = None req.body = None response = req.get_response(app) self.assertEqual(response.status_int, http.OK) self.assertEqual(b'success', response.body) # the body is validated in the controller expected_body = {'body': None} response = req.get_response(app) expected_unsupported_type_body = {'badRequest': {'message': 'The request body invalid', 'code': http.BAD_REQUEST}} self.assertEqual(response.status_int, http.BAD_REQUEST) self.assertEqual(expected_unsupported_type_body, jsonutils.loads(response.body)) def test_resource_call_with_method_put(self): class Controller(object): def update(self, req, id, body): if expected_body != body: msg = "The request body invalid" raise webob.exc.HTTPBadRequest(explanation=msg) return "success" # verify the method: PUT app = fakes.TestRouter(Controller()) req = webob.Request.blank('/tests/test_id', method="PUT", content_type='application/json') req.body = b'{"body": {"key": "value"}}' expected_body = {'body': { "key": "value" } } response = req.get_response(app) self.assertEqual(b'success', response.body) self.assertEqual(response.status_int, http.OK) req.body = None expected_body = None response = req.get_response(app) self.assertEqual(response.status_int, http.OK) # verify no content_type is contained in the request req = webob.Request.blank('/tests/test_id', method="PUT", content_type='application/xml') req.content_type = 'application/xml' req.body = b'{"body": {"key": "value"}}' response = req.get_response(app) expected_unsupported_type_body = {'badMediaType': {'message': 'Unsupported Content-Type', 'code': http.UNSUPPORTED_MEDIA_TYPE}} self.assertEqual(response.status_int, http.UNSUPPORTED_MEDIA_TYPE) self.assertEqual(expected_unsupported_type_body, jsonutils.loads(response.body)) def test_resource_call_with_method_delete(self): class Controller(object): def delete(self, req, id): return "success" # verify the method: DELETE app = fakes.TestRouter(Controller()) req = webob.Request.blank('/tests/test_id', method="DELETE") response = req.get_response(app) self.assertEqual(response.status_int, http.OK) self.assertEqual(b'success', response.body) # ignore the body req.body = b'{"body": {"key": "value"}}' response = req.get_response(app) self.assertEqual(response.status_int, http.OK) self.assertEqual(b'success', response.body) def test_resource_not_authorized(self): class Controller(object): def index(self, req): raise exception.Forbidden() req = webob.Request.blank('/tests') app = fakes.TestRouter(Controller()) response = req.get_response(app) self.assertEqual(response.status_int, http.FORBIDDEN) def test_dispatch(self): class Controller(object): def index(self, req, pants=None): return pants controller = Controller() resource = wsgi.Resource(controller) method, extensions = resource.get_method(None, 'index', None, '') actual = resource.dispatch(method, None, {'pants': 'off'}) expected = 'off' self.assertEqual(actual, expected) def test_get_method_unknown_controller_method(self): class Controller(object): def index(self, req, pants=None): return pants controller = Controller() resource = wsgi.Resource(controller) self.assertRaises(AttributeError, resource.get_method, None, 'create', None, '') def test_get_method_action_json(self): class Controller(wsgi.Controller): @wsgi.action('fooAction') def _action_foo(self, req, id, body): return body controller = Controller() resource = wsgi.Resource(controller) method, extensions = resource.get_method(None, 'action', 'application/json', '{"fooAction": true}') self.assertEqual(controller._action_foo, method) def test_get_method_action_bad_body(self): class Controller(wsgi.Controller): @wsgi.action('fooAction') def _action_foo(self, req, id, body): return body controller = Controller() resource = wsgi.Resource(controller) self.assertRaises(exception.MalformedRequestBody, resource.get_method, None, 'action', 'application/json', '{}') def test_get_method_unknown_controller_action(self): class Controller(wsgi.Controller): @wsgi.action('fooAction') def _action_foo(self, req, id, body): return body controller = Controller() resource = wsgi.Resource(controller) self.assertRaises(KeyError, resource.get_method, None, 'action', 'application/json', '{"barAction": true}') def test_get_method_action_method(self): class Controller(object): def action(self, req, pants=None): return pants controller = Controller() resource = wsgi.Resource(controller) method, extensions = resource.get_method(None, 'action', 'application/xml', 'true