1 files changed, 119 insertions, 0 deletions
diff --git a/keystone-moon/keystone/tests/unit/test_middleware.py b/keystone-moon/keystone/tests/unit/test_middleware.py
new file mode 100644
index 00000000..3a26dd24
--- /dev/null
+++ b/keystone-moon/keystone/tests/unit/test_middleware.py
@@ -0,0 +1,119 @@
+# 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.
+
+from oslo_config import cfg
+import webob
+
+from keystone import middleware
+from keystone.tests import unit as tests
+
+
+CONF = cfg.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)