¡@
OpenStack Study: test_middleware.py**** CubicPower OpenStack Study ****
# Copyright 2012 OpenStack Foundation # # 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 webob from keystone import config from keystone import middleware from keystone.openstack.common import jsonutils from keystone import tests CONF = config.CONF
**** CubicPower OpenStack Study **** def make_request(**kwargs): accept = kwargs.pop('accept', None) method = kwargs.pop('method', 'GET') body = kwargs.pop('body', None) req = webob.Request.blank('/', **kwargs) req.method = method if body is not None: req.body = body if accept is not None: req.accept = accept return req
**** CubicPower OpenStack Study **** def make_response(**kwargs): body = kwargs.pop('body', None) return webob.Response(body)
**** CubicPower OpenStack Study **** class TokenAuthMiddlewareTest(tests.TestCase):
**** CubicPower OpenStack Study **** def test_request(self): req = make_request() req.headers[middleware.AUTH_TOKEN_HEADER] = 'MAGIC' middleware.TokenAuthMiddleware(None).process_request(req) context = req.environ[middleware.CONTEXT_ENV] self.assertEqual(context['token_id'], 'MAGIC')
**** CubicPower OpenStack Study **** class AdminTokenAuthMiddlewareTest(tests.TestCase):
**** CubicPower OpenStack Study **** def test_request_admin(self): req = make_request() req.headers[middleware.AUTH_TOKEN_HEADER] = CONF.admin_token middleware.AdminTokenAuthMiddleware(None).process_request(req) context = req.environ[middleware.CONTEXT_ENV] self.assertTrue(context['is_admin'])
**** CubicPower OpenStack Study **** def test_request_non_admin(self): req = make_request() req.headers[middleware.AUTH_TOKEN_HEADER] = 'NOT-ADMIN' middleware.AdminTokenAuthMiddleware(None).process_request(req) context = req.environ[middleware.CONTEXT_ENV] self.assertFalse(context['is_admin'])
**** CubicPower OpenStack Study **** class PostParamsMiddlewareTest(tests.TestCase):
**** CubicPower OpenStack Study **** def test_request_with_params(self): req = make_request(body="arg1=one", method='POST') middleware.PostParamsMiddleware(None).process_request(req) params = req.environ[middleware.PARAMS_ENV] self.assertEqual(params, {"arg1": "one"})
**** CubicPower OpenStack Study **** class JsonBodyMiddlewareTest(tests.TestCase):
**** CubicPower OpenStack Study **** def test_request_with_params(self): req = make_request(body='{"arg1": "one", "arg2": ["a"]}', content_type='application/json', method='POST') middleware.JsonBodyMiddleware(None).process_request(req) params = req.environ[middleware.PARAMS_ENV] self.assertEqual(params, {"arg1": "one", "arg2": ["a"]})
**** CubicPower OpenStack Study **** def test_malformed_json(self): req = make_request(body='{"arg1": "on', content_type='application/json', method='POST') resp = middleware.JsonBodyMiddleware(None).process_request(req) self.assertEqual(resp.status_int, 400)
**** CubicPower OpenStack Study **** def test_no_content_type(self): req = make_request(body='{"arg1": "one", "arg2": ["a"]}', method='POST') middleware.JsonBodyMiddleware(None).process_request(req) params = req.environ[middleware.PARAMS_ENV] self.assertEqual(params, {"arg1": "one", "arg2": ["a"]})
**** CubicPower OpenStack Study **** def test_unrecognized_content_type(self): req = make_request(body='{"arg1": "one", "arg2": ["a"]}', content_type='text/plain', method='POST') resp = middleware.JsonBodyMiddleware(None).process_request(req) self.assertEqual(resp.status_int, 400)
**** CubicPower OpenStack Study **** def test_unrecognized_content_type_without_body(self): req = make_request(content_type='text/plain', method='GET') middleware.JsonBodyMiddleware(None).process_request(req) params = req.environ.get(middleware.PARAMS_ENV, {}) self.assertEqual(params, {})
**** CubicPower OpenStack Study **** class XmlBodyMiddlewareTest(tests.TestCase):
**** CubicPower OpenStack Study **** def test_client_wants_xml_back(self): """Clients requesting XML should get what they ask for.""" body = '{"container": {"attribute": "value"}}' req = make_request(body=body, method='POST', accept='application/xml') middleware.XmlBodyMiddleware(None).process_request(req) resp = make_response(body=body) middleware.XmlBodyMiddleware(None).process_response(req, resp) self.assertEqual(resp.content_type, 'application/xml')
**** CubicPower OpenStack Study **** def test_client_wants_json_back(self): """Clients requesting JSON should definitely not get XML back.""" body = '{"container": {"attribute": "value"}}' req = make_request(body=body, method='POST', accept='application/json') middleware.XmlBodyMiddleware(None).process_request(req) resp = make_response(body=body) middleware.XmlBodyMiddleware(None).process_response(req, resp) self.assertNotIn('application/xml', resp.content_type)
**** CubicPower OpenStack Study **** def test_client_fails_to_specify_accept(self): """If client does not specify an Accept header, default to JSON.""" body = '{"container": {"attribute": "value"}}' req = make_request(body=body, method='POST') middleware.XmlBodyMiddleware(None).process_request(req) resp = make_response(body=body) middleware.XmlBodyMiddleware(None).process_response(req, resp) self.assertNotIn('application/xml', resp.content_type)
**** CubicPower OpenStack Study **** def test_xml_replaced_by_json(self): """XML requests should be replaced by JSON requests.""" req = make_request( body=' content_type='application/xml', method='POST') middleware.XmlBodyMiddleware(None).process_request(req) self.assertEqual(req.content_type, 'application/json') self.assertTrue(jsonutils.loads(req.body))
**** CubicPower OpenStack Study **** def test_json_unnaffected(self): """JSON-only requests should be unaffected by the XML middleware.""" content_type = 'application/json' body = '{"container": {"attribute": "value"}}' req = make_request(body=body, content_type=content_type, method='POST') middleware.XmlBodyMiddleware(None).process_request(req) self.assertEqual(req.body, body) self.assertEqual(req.content_type, content_type)
|