openstack
/
keystone
Code Issues Proposed changes

add the various role tests

This commit is contained in:
termie 2012-01-04 17:31:54 -08:00
parent 5c89972ffe
commit ebe158f750
4 changed files with 67 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
keystonelight/backends/kvs.py
View File

@ -66,6 +66,9 @@ class KvsIdentity(object):
def list_users(self):
return self.db.get('user_list', [])
def list_roles(self):
return self.db.get('role_list', [])
# These should probably be part of the high-level API
def add_user_to_tenant(self, tenant_id, user_id):
user_ref = self.get_user(user_id)
@ -171,6 +174,9 @@ class KvsIdentity(object):
def create_role(self, id, role):
self.db.set('role-%s' % id, role)
role_list = set(self.db.get('role_list', []))
role_list.add(id)
self.db.set('role_list', list(role_list))
return role
def update_role(self, id, role):
@ -179,6 +185,9 @@ class KvsIdentity(object):
def delete_role(self, id):
self.db.delete('role-%s' % id)
role_list = set(self.db.get('role_list', []))
role_list.remove(id)
self.db.set('role_list', list(role_list))
return None

3
keystonelight/identity.py
View File

@ -42,6 +42,9 @@ class Manager(object):
def list_users(self, context):
return self.driver.list_users()
def list_roles(self, context):
return self.driver.list_roles()
# These should probably be the high-level API calls
def add_user_to_tenant(self, context, user_id, tenant_id):
self.driver.add_user_to_tenant(user_id, tenant_id)

71
keystonelight/keystone_compat.py
View File

@ -235,33 +235,39 @@ class KeystoneAdminCrudExtension(wsgi.ExtensionRouter):
# Service Operations
mapper.connect("/OS-KSADM/services",
controller=service_controller,
action="get_services",
conditions=dict(method=["GET"]))
controller=service_controller,
action="get_services",
conditions=dict(method=["GET"]))
mapper.connect("/OS-KSADM/services",
controller=service_controller,
action="create_service",
conditions=dict(method=["POST"]))
controller=service_controller,
action="create_service",
conditions=dict(method=["POST"]))
mapper.connect("/OS-KSADM/services/{service_id}",
controller=service_controller,
action="delete_service",
conditions=dict(method=["DELETE"]))
controller=service_controller,
action="delete_service",
conditions=dict(method=["DELETE"]))
mapper.connect("/OS-KSADM/services/{service_id}",
controller=service_controller,
action="get_service",
conditions=dict(method=["GET"]))
controller=service_controller,
action="get_service",
conditions=dict(method=["GET"]))
# Role Operations
mapper.connect("/OS-KSADM/roles", controller=role_controller,
action="create_role", conditions=dict(method=["POST"]))
mapper.connect("/OS-KSADM/roles", controller=role_controller,
action="get_roles", conditions=dict(method=["GET"]))
mapper.connect("/OS-KSADM/roles",
controller=role_controller,
action="create_role",
conditions=dict(method=["POST"]))
mapper.connect("/OS-KSADM/roles",
controller=role_controller,
action="get_roles",
conditions=dict(method=["GET"]))
mapper.connect("/OS-KSADM/roles/{role_id}",
controller=role_controller, action="get_role",
conditions=dict(method=["GET"]))
controller=role_controller,
action="get_role",
conditions=dict(method=["GET"]))
mapper.connect("/OS-KSADM/roles/{role_id}",
controller=role_controller, action="delete_role",
conditions=dict(method=["DELETE"]))
controller=role_controller,
action="delete_role",
conditions=dict(method=["DELETE"]))
super(KeystoneAdminCrudExtension, self).__init__(
application, options, mapper)
@ -648,6 +654,31 @@ class KeystoneRoleController(service.BaseApplication):
def get_user_roles(self, context, user_id, tenant_id=None):
raise NotImplemented()
# CRUD extension
def get_role(self, context, role_id):
self.assert_admin(context)
role_ref = self.identity_api.get_role(context, role_id)
if not role_ref:
raise exc.HTTPNotFound()
return {'role': role_ref}
def create_role(self, context, role):
role_id = uuid.uuid4().hex
role['id'] = role_id
role_ref = self.identity_api.create_role(context, role_id, role)
return {'role': role_ref}
def delete_role(self, context, role_id):
self.assert_admin(context)
role_ref = self.identity_api.delete_role(context, role_id)
def get_roles(self, context):
self.assert_admin(context)
roles = self.identity_api.list_roles(context)
# TODO(termie): probably inefficient at some point
return {'roles': [self.identity_api.get_role(context, x)
for x in roles]}
# COMPAT(diablo): CRUD extension
def get_role_refs(self, context, user_id):
"""Ultimate hack to get around having to make role_refs first-class.

8
tests/test_keystoneclient_compat.py
View File

@ -184,7 +184,7 @@ class MasterCompatTestCase(CompatTestCase):
def test_role_get(self):
client = self.foo_client()
role = client.roles.get('keystone_admin')
self.assertEquals(role.name, 'keystone_admin')
self.assertEquals(role.id, 'keystone_admin')
def test_role_create_and_delete(self):
from keystoneclient import exceptions as client_exceptions
@ -194,10 +194,10 @@ class MasterCompatTestCase(CompatTestCase):
role = client.roles.create(test_role)
self.assertEquals(role.name, test_role)
role = client.roles.get(test_role)
role = client.roles.get(role)
self.assertEquals(role.name, test_role)
client.roles.delete(test_role)
client.roles.delete(role)
self.assertRaises(client_exceptions.NotFound, client.roles.get,
test_role)
@ -210,7 +210,7 @@ class MasterCompatTestCase(CompatTestCase):
def test_roles_get_by_user(self):
client = self.foo_client()
roles = client.roles.get_user_role_refs('FOO')
roles = client.roles.get_user_role_refs('foo')
self.assertTrue(len(roles) > 0)
def test_service_create_and_delete(self):