# 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 ddt import mock import six import webob import inspect from manila.api.openstack import api_version_request as api_version from manila.api.openstack import wsgi from manila import context from manila import exception from manila import policy from manila import test from manila.tests.api import fakes @ddt.ddt class RequestTest(test.TestCase): def test_content_type_missing(self): request = wsgi.Request.blank('/tests/123', method='POST') request.body = six.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 = six.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("application/json", result) def test_content_type_from_accept(self): content_type = 'application/json' request = wsgi.Request.blank('/tests/123') request.headers["Accept"] = content_type result = request.best_match_content_type() self.assertEqual(content_type, result) 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("application/json", result) 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("application/json", result) def test_content_type_from_query_extension(self): request = wsgi.Request.blank('/tests/123.json') result = request.best_match_content_type() self.assertEqual("application/json", result) request = wsgi.Request.blank('/tests/123.invalid') result = request.best_match_content_type() self.assertEqual("application/json", result) 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("application/json", result) def test_cache_and_retrieve_resources(self): request = wsgi.Request.blank('/foo') # Test that trying to retrieve a cached object on # an empty cache fails gracefully self.assertIsNone(request.cached_resource()) self.assertIsNone(request.cached_resource_by_id('r-0')) resources = [{'id': 'r-%s' % x} for x in range(3)] # Cache an empty list of resources using the default name request.cache_resource([]) self.assertEqual({}, request.cached_resource()) self.assertIsNone(request.cached_resource('r-0')) # Cache some resources request.cache_resource(resources[:2]) # Cache one resource request.cache_resource(resources[2]) # Cache a different resource name other_resource = {'id': 'o-0'} request.cache_resource(other_resource, name='other-resource') self.assertEqual(resources[0], request.cached_resource_by_id('r-0')) self.assertEqual(resources[1], request.cached_resource_by_id('r-1')) self.assertEqual(resources[2], request.cached_resource_by_id('r-2')) self.assertIsNone(request.cached_resource_by_id('r-3')) self.assertEqual( {'r-0': resources[0], 'r-1': resources[1], 'r-2': resources[2]}, request.cached_resource()) self.assertEqual( other_resource, request.cached_resource_by_id('o-0', name='other-resource')) @ddt.data( 'share_type', ) def test_cache_and_retrieve_resources_by_resource(self, resource_name): cache_all_func = 'cache_db_%ss' % resource_name cache_one_func = 'cache_db_%s' % resource_name get_db_all_func = 'get_db_%ss' % resource_name get_db_one_func = 'get_db_%s' % resource_name r = wsgi.Request.blank('/foo') amount = 5 res_range = range(amount) resources = [{'id': 'id%s' % x} for x in res_range] # Store 2 getattr(r, cache_all_func)(resources[:amount - 1]) # Store 1 getattr(r, cache_one_func)(resources[amount - 1]) for i in res_range: self.assertEqual( resources[i], getattr(r, get_db_one_func)('id%s' % i), ) self.assertIsNone(getattr(r, get_db_one_func)('id%s' % amount)) self.assertEqual( {'id%s' % i: resources[i] for i in res_range}, getattr(r, get_db_all_func)()) def test_set_api_version_request_exception(self): min_version = api_version.APIVersionRequest('2.0') max_version = api_version.APIVersionRequest('2.45') self.mock_object(api_version, 'max_api_version', mock.Mock(return_value=max_version)) self.mock_object(api_version, 'min_api_version', mock.Mock(return_value=min_version)) headers = {'X-OpenStack-Manila-API-Version': '2.50'} request = wsgi.Request.blank( 'https://openstack.acme.com/v2/shares', method='GET', headers=headers, script_name='/v2/shares') self.assertRaises(exception.InvalidGlobalAPIVersion, request.set_api_version_request) self.assertEqual(api_version.APIVersionRequest('2.50'), request.api_version_request) @ddt.data('', '/share', '/v1', '/v2/shares', '/v1.1/', '/share/v1', '/shared-file-sytems/v2', '/share/v3.5/share-replicas', '/shared-file-sytems/v2/shares/xyzzy/action') def test_set_api_version_request(self, resource): min_version = api_version.APIVersionRequest('2.0') max_version = api_version.APIVersionRequest('3.0') self.mock_object(api_version, 'max_api_version', mock.Mock(return_value=max_version)) self.mock_object(api_version, 'min_api_version', mock.Mock(return_value=min_version)) request = wsgi.Request.blank( 'https://openstack.acme.com%s' % resource, method='GET', headers={'X-OpenStack-Manila-API-Version': '2.117'}, script_name=resource) self.assertIsNone(request.set_api_version_request()) if not resource or not ('/v1' in resource or '/v2' in resource): self.assertEqual(api_version.APIVersionRequest(), request.api_version_request) elif 'v1' in resource: self.assertEqual(api_version.APIVersionRequest('1.0'), request.api_version_request) else: self.assertEqual(api_version.APIVersionRequest('2.117'), request.api_version_request) def test_set_api_version_request_no_version_header(self): min_version = api_version.APIVersionRequest('2.0') max_version = api_version.APIVersionRequest('2.45') self.mock_object(api_version, 'max_api_version', mock.Mock(return_value=max_version)) self.mock_object(api_version, 'min_api_version', mock.Mock(return_value=min_version)) headers = {} request = wsgi.Request.blank( 'https://openstack.acme.com/v2/shares', method='GET', headers=headers, script_name='/v2/shares') self.assertIsNone(request.set_api_version_request()) self.assertEqual(api_version.APIVersionRequest('2.0'), request.api_version_request) @ddt.data(None, 'true', 'false') def test_set_api_version_request_experimental_header(self, experimental): min_version = api_version.APIVersionRequest('2.0') max_version = api_version.APIVersionRequest('2.45') self.mock_object(api_version, 'max_api_version', mock.Mock(return_value=max_version)) self.mock_object(api_version, 'min_api_version', mock.Mock(return_value=min_version)) headers = {'X-OpenStack-Manila-API-Version': '2.38'} if experimental: headers['X-OpenStack-Manila-API-Experimental'] = experimental request = wsgi.Request.blank( 'https://openstack.acme.com/v2/shares', method='GET', headers=headers, script_name='/v2/shares') self.assertIsNone(request.set_api_version_request()) self.assertEqual(request.api_version_request, api_version.APIVersionRequest('2.38')) expected_experimental = experimental == 'true' or False self.assertEqual(expected_experimental, request.api_version_request.experimental) class ActionDispatcherTest(test.TestCase): def test_dispatch(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: 'pants' self.assertEqual('pants', serializer.dispatch({}, action='create')) def test_dispatch_action_None(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: 'pants' serializer.default = lambda x: 'trousers' self.assertEqual('trousers', serializer.dispatch({}, action=None)) def test_dispatch_default(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: 'pants' serializer.default = lambda x: 'trousers' self.assertEqual('trousers', serializer.dispatch({}, action='update')) class DictSerializerTest(test.TestCase): def test_dispatch_default(self): serializer = wsgi.DictSerializer() self.assertEqual('', serializer.serialize({}, 'update')) class JSONDictSerializerTest(test.TestCase): def test_json(self): input_dict = dict(servers=dict(a=(2, 3))) expected_json = six.b('{"servers":{"a":[2,3]}}') serializer = wsgi.JSONDictSerializer() result = serializer.serialize(input_dict) result = result.replace(six.b('\n'), six.b('')).replace(six.b(' '), six.b('')) self.assertEqual(expected_json, result) class TextDeserializerTest(test.TestCase): def test_dispatch_default(self): deserializer = wsgi.TextDeserializer() self.assertEqual({}, deserializer.deserialize({}, 'update')) class JSONDeserializerTest(test.TestCase): 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(as_dict, deserializer.deserialize(data)) class ResourceTest(test.TestCase): def test_resource_call(self): class Controller(object): def index(self, req): return 'off' req = webob.Request.blank('/tests') app = fakes.TestRouter(Controller()) response = req.get_response(app) self.assertEqual(six.b('off'), response.body) self.assertEqual(200, response.status_int) def test_resource_not_authorized(self): class Controller(object): def index(self, req): raise exception.NotAuthorized() req = webob.Request.blank('/tests') app = fakes.TestRouter(Controller()) response = req.get_response(app) self.assertEqual(403, response.status_int) 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(expected, actual) def test_get_method_undefined_controller_action(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