¡@

Home 

OpenStack Study: test_context_middleware.py

OpenStack Index

**** CubicPower OpenStack Study ****

# 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 webob

from glance.api.middleware import context

import glance.context

from glance.tests.unit import base

**** CubicPower OpenStack Study ****

class TestContextMiddleware(base.IsolatedUnitTest):

**** CubicPower OpenStack Study ****

    def _build_request(self, roles=None, identity_status='Confirmed',

                       service_catalog=None):

        req = webob.Request.blank('/')

        req.headers['x-auth-token'] = 'token1'

        req.headers['x-identity-status'] = identity_status

        req.headers['x-user-id'] = 'user1'

        req.headers['x-tenant-id'] = 'tenant1'

        _roles = roles or ['role1', 'role2']

        req.headers['x-roles'] = ','.join(_roles)

        if service_catalog:

            req.headers['x-service-catalog'] = service_catalog

        return req

**** CubicPower OpenStack Study ****

    def _build_middleware(self):

        return context.ContextMiddleware(None)

**** CubicPower OpenStack Study ****

    def test_header_parsing(self):

        req = self._build_request()

        self._build_middleware().process_request(req)

        self.assertEqual(req.context.auth_tok, 'token1')

        self.assertEqual(req.context.user, 'user1')

        self.assertEqual(req.context.tenant, 'tenant1')

        self.assertEqual(req.context.roles, ['role1', 'role2'])

**** CubicPower OpenStack Study ****

    def test_is_admin_flag(self):

        # is_admin check should look for 'admin' role by default

        req = self._build_request(roles=['admin', 'role2'])

        self._build_middleware().process_request(req)

        self.assertTrue(req.context.is_admin)

        # without the 'admin' role, is_admin should be False

        req = self._build_request()

        self._build_middleware().process_request(req)

        self.assertFalse(req.context.is_admin)

        # if we change the admin_role attribute, we should be able to use it

        req = self._build_request()

        self.config(admin_role='role1')

        self._build_middleware().process_request(req)

        self.assertTrue(req.context.is_admin)

**** CubicPower OpenStack Study ****

    def test_roles_case_insensitive(self):

        # accept role from request

        req = self._build_request(roles=['Admin', 'role2'])

        self._build_middleware().process_request(req)

        self.assertTrue(req.context.is_admin)

        # accept role from config

        req = self._build_request(roles=['role1'])

        self.config(admin_role='rOLe1')

        self._build_middleware().process_request(req)

        self.assertTrue(req.context.is_admin)

**** CubicPower OpenStack Study ****

    def test_roles_stripping(self):

        # stripping extra spaces in request

        req = self._build_request(roles=['\trole1'])

        self.config(admin_role='role1')

        self._build_middleware().process_request(req)

        self.assertTrue(req.context.is_admin)

        # stripping extra spaces in config

        req = self._build_request(roles=['\trole1\n'])

        self.config(admin_role=' role1\t')

        self._build_middleware().process_request(req)

        self.assertTrue(req.context.is_admin)

**** CubicPower OpenStack Study ****

    def test_anonymous_access_enabled(self):

        req = self._build_request(identity_status='Nope')

        self.config(allow_anonymous_access=True)

        middleware = self._build_middleware()

        middleware.process_request(req)

        self.assertIsNone(req.context.auth_tok)

        self.assertIsNone(req.context.user)

        self.assertIsNone(req.context.tenant)

        self.assertEqual(req.context.roles, [])

        self.assertFalse(req.context.is_admin)

        self.assertTrue(req.context.read_only)

**** CubicPower OpenStack Study ****

    def test_anonymous_access_defaults_to_disabled(self):

        req = self._build_request(identity_status='Nope')

        middleware = self._build_middleware()

        self.assertRaises(webob.exc.HTTPUnauthorized,

                          middleware.process_request, req)

**** CubicPower OpenStack Study ****

    def test_service_catalog(self):

        catalog_json = "[{}]"

        req = self._build_request(service_catalog=catalog_json)

        self._build_middleware().process_request(req)

        self.assertEqual([{}], req.context.service_catalog)

**** CubicPower OpenStack Study ****

    def test_invalid_service_catalog(self):

        catalog_json = "bad json"

        req = self._build_request(service_catalog=catalog_json)

        middleware = self._build_middleware()

        self.assertRaises(webob.exc.HTTPInternalServerError,

                          middleware.process_request, req)

**** CubicPower OpenStack Study ****

class TestUnauthenticatedContextMiddleware(base.IsolatedUnitTest):

**** CubicPower OpenStack Study ****

    def test_request(self):

        middleware = context.UnauthenticatedContextMiddleware(None)

        req = webob.Request.blank('/')

        middleware.process_request(req)

        self.assertIsNone(req.context.auth_tok)

        self.assertIsNone(req.context.user)

        self.assertIsNone(req.context.tenant)

        self.assertEqual(req.context.roles, [])

        self.assertTrue(req.context.is_admin)

**** CubicPower OpenStack Study ****

    def test_response(self):

        middleware = context.UnauthenticatedContextMiddleware(None)

        req = webob.Request.blank('/')

        req.context = glance.context.RequestContext()

        resp = webob.Response()

        resp.request = req

        middleware.process_response(resp)

        self.assertEqual(resp.headers['x-openstack-request-id'],

                         'req-%s' % req.context.request_id)