x/vmware-nsx
Code Issues Proposed changes

Merge "Remove deprecated fields in keystone auth middleware"

Browse Source
This commit is contained in:
Jenkins 2013-11-15 00:12:38 +00:00 committed by Gerrit Code Review
commit 36cb6b0e1b
2 changed files with 12 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
neutron/auth.py
View File

@ -31,16 +31,16 @@ class NeutronKeystoneContext(wsgi.Middleware):
@webob.dec.wsgify @webob.dec.wsgify
def __call__(self, req): def __call__(self, req):
# Determine the user ID # Determine the user ID
user_id = req.headers.get('X_USER_ID', req.headers.get('X_USER')) user_id = req.headers.get('X_USER_ID')
if not user_id: if not user_id:
LOG.debug(_("Neither X_USER_ID nor X_USER found in request")) LOG.debug(_("X_USER_ID is not found in request"))
return webob.exc.HTTPUnauthorized() return webob.exc.HTTPUnauthorized()
# Determine the tenant # Determine the tenant
tenant_id = req.headers.get('X_TENANT_ID', req.headers.get('X_TENANT')) tenant_id = req.headers.get('X_PROJECT_ID')
# Suck out the roles # Suck out the roles
roles = [r.strip() for r in req.headers.get('X_ROLE', '').split(',')] roles = [r.strip() for r in req.headers.get('X_ROLES', '').split(',')]
# Create a context with the authentication data # Create a context with the authentication data
ctx = context.Context(user_id, tenant_id, roles=roles) ctx = context.Context(user_id, tenant_id, roles=roles)

44
neutron/tests/unit/test_auth.py
View File

@ -36,58 +36,28 @@ class NeutronKeystoneContextTestCase(base.BaseTestCase):
self.request.headers['X_AUTH_TOKEN'] = 'testauthtoken' self.request.headers['X_AUTH_TOKEN'] = 'testauthtoken'
def test_no_user_no_user_id(self): def test_no_user_no_user_id(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid' self.request.headers['X_PROJECT_ID'] = 'testtenantid'
response = self.request.get_response(self.middleware) response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '401 Unauthorized') self.assertEqual(response.status, '401 Unauthorized')
def test_with_user(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid'
self.request.headers['X_USER_ID'] = 'testuserid'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.user_id, 'testuserid')
def test_with_user_id(self): def test_with_user_id(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid' self.request.headers['X_PROJECT_ID'] = 'testtenantid'
self.request.headers['X_USER'] = 'testuser'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.user_id, 'testuser')
def test_user_id_trumps_user(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid'
self.request.headers['X_USER_ID'] = 'testuserid' self.request.headers['X_USER_ID'] = 'testuserid'
self.request.headers['X_USER'] = 'testuser'
response = self.request.get_response(self.middleware) response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK') self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.user_id, 'testuserid') self.assertEqual(self.context.user_id, 'testuserid')
def test_with_tenant_id(self): def test_with_tenant_id(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid' self.request.headers['X_PROJECT_ID'] = 'testtenantid'
self.request.headers['X_USER_ID'] = 'test_user_id' self.request.headers['X_USER_ID'] = 'test_user_id'
response = self.request.get_response(self.middleware) response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK') self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.tenant_id, 'testtenantid') self.assertEqual(self.context.tenant_id, 'testtenantid')
def test_with_tenant(self):
self.request.headers['X_TENANT'] = 'testtenant'
self.request.headers['X_USER_ID'] = 'test_user_id'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.tenant_id, 'testtenant')
def test_tenant_id_trumps_tenant(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid'
self.request.headers['X_TENANT'] = 'testtenant'
self.request.headers['X_USER_ID'] = 'testuserid'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.tenant_id, 'testtenantid')
def test_roles_no_admin(self): def test_roles_no_admin(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid' self.request.headers['X_PROJECT_ID'] = 'testtenantid'
self.request.headers['X_USER_ID'] = 'testuserid' self.request.headers['X_USER_ID'] = 'testuserid'
self.request.headers['X_ROLE'] = 'role1, role2 , role3,role4,role5' self.request.headers['X_ROLES'] = 'role1, role2 , role3,role4,role5'
response = self.request.get_response(self.middleware) response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK') self.assertEqual(response.status, '200 OK')
self.assertEqual(self.context.roles, ['role1', 'role2', 'role3', self.assertEqual(self.context.roles, ['role1', 'role2', 'role3',
@ -95,9 +65,9 @@ class NeutronKeystoneContextTestCase(base.BaseTestCase):
self.assertEqual(self.context.is_admin, False) self.assertEqual(self.context.is_admin, False)
def test_roles_with_admin(self): def test_roles_with_admin(self):
self.request.headers['X_TENANT_ID'] = 'testtenantid' self.request.headers['X_PROJECT_ID'] = 'testtenantid'
self.request.headers['X_USER_ID'] = 'testuserid' self.request.headers['X_USER_ID'] = 'testuserid'
self.request.headers['X_ROLE'] = ('role1, role2 , role3,role4,role5,' self.request.headers['X_ROLES'] = ('role1, role2 , role3,role4,role5,'
'AdMiN') 'AdMiN')
response = self.request.get_response(self.middleware) response = self.request.get_response(self.middleware)
self.assertEqual(response.status, '200 OK') self.assertEqual(response.status, '200 OK')