120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
# 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 import tests
|
|
|
|
|
|
CONF = config.CONF
|
|
|
|
|
|
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
|
|
|
|
|
|
def make_response(**kwargs):
|
|
body = kwargs.pop('body', None)
|
|
return webob.Response(body)
|
|
|
|
|
|
class TokenAuthMiddlewareTest(tests.TestCase):
|
|
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('MAGIC', context['token_id'])
|
|
|
|
|
|
class AdminTokenAuthMiddlewareTest(tests.TestCase):
|
|
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'])
|
|
|
|
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'])
|
|
|
|
|
|
class PostParamsMiddlewareTest(tests.TestCase):
|
|
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({"arg1": "one"}, params)
|
|
|
|
|
|
class JsonBodyMiddlewareTest(tests.TestCase):
|
|
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({"arg1": "one", "arg2": ["a"]}, params)
|
|
|
|
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(400, resp.status_int)
|
|
|
|
def test_not_dict_body(self):
|
|
req = make_request(body='42',
|
|
content_type='application/json',
|
|
method='POST')
|
|
resp = middleware.JsonBodyMiddleware(None).process_request(req)
|
|
self.assertEqual(400, resp.status_int)
|
|
self.assertTrue('valid JSON object' in resp.json['error']['message'])
|
|
|
|
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({"arg1": "one", "arg2": ["a"]}, params)
|
|
|
|
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(400, resp.status_int)
|
|
|
|
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)
|