x/python-ganttclient
Code Issues Proposed changes

Made group membership check only search group instead of subtree. Roles in a group are removed when a user is removed from that group. Added test

Browse Source
This commit is contained in:
Vishvananda Ishaya
2010-08-05 16:56:23 -07:00
parent 181b1a719f
commit 02728ec84d
3 changed files with 34 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
nova/auth/fakeldap.py
View File

@@ -28,6 +28,8 @@ import json
from nova import datastore
SCOPE_BASE = 0
SCOPE_ONELEVEL = 1 # not implemented
SCOPE_SUBTREE = 2
MOD_ADD = 0
MOD_DELETE = 1
@@ -188,15 +190,18 @@ class FakeLDAP(object):
Args:
dn -- dn to search under
scope -- only SCOPE_SUBTREE is supported
scope -- only SCOPE_BASE and SCOPE_SUBTREE are supported
query -- query to filter objects by
fields -- fields to return. Returns all fields if not specified
"""
if scope != SCOPE_SUBTREE:
if scope != SCOPE_BASE and scope != SCOPE_SUBTREE:
raise NotImplementedError(str(scope))
redis = datastore.Redis.instance()
keys = redis.keys("%s*%s" % (self.__redis_prefix, dn))
if scope == SCOPE_BASE:
keys = ["%s%s" % (self.__redis_prefix, dn)]
else:
keys = redis.keys("%s*%s" % (self.__redis_prefix, dn))
objects = []
for key in keys:
# get the attributes from redis

25
nova/auth/ldapdriver.py
View File

@@ -272,26 +272,30 @@ class LdapDriver(object):
"""Check if project exists"""
return self.get_project(name) != None
def __find_object(self, dn, query = None):
def __find_object(self, dn, query=None, scope=None):
"""Find an object by dn and query"""
objects = self.__find_objects(dn, query)
objects = self.__find_objects(dn, query, scope)
if len(objects) == 0:
return None
return objects[0]
def __find_dns(self, dn, query=None):
def __find_dns(self, dn, query=None, scope=None):
"""Find dns by query"""
if scope is None: # one of the flags is 0!!
scope = self.ldap.SCOPE_SUBTREE
try:
res = self.conn.search_s(dn, self.ldap.SCOPE_SUBTREE, query)
res = self.conn.search_s(dn, scope, query)
except self.ldap.NO_SUCH_OBJECT:
return []
# just return the DNs
return [dn for dn, attributes in res]
def __find_objects(self, dn, query = None):
def __find_objects(self, dn, query=None, scope=None):
"""Find objects by query"""
if scope is None: # one of the flags is 0!!
scope = self.ldap.SCOPE_SUBTREE
try:
res = self.conn.search_s(dn, self.ldap.SCOPE_SUBTREE, query)
res = self.conn.search_s(dn, scope, query)
except self.ldap.NO_SUCH_OBJECT:
return []
# just return the attributes
@@ -361,7 +365,8 @@ class LdapDriver(object):
if not self.__group_exists(group_dn):
return False
res = self.__find_object(group_dn,
'(member=%s)' % self.__uid_to_dn(uid))
'(member=%s)' % self.__uid_to_dn(uid),
self.ldap.SCOPE_BASE)
return res != None
def __add_to_group(self, uid, group_dn):
@@ -391,7 +396,11 @@ class LdapDriver(object):
if not self.__is_in_group(uid, group_dn):
raise exception.NotFound("User %s is not a member of the group" %
(uid,))
self.__safe_remove_from_group(uid, group_dn)
# NOTE(vish): remove user from group and any sub_groups
sub_dns = self.__find_group_dns_with_member(
group_dn, uid)
for sub_dn in sub_dns:
self.__safe_remove_from_group(uid, sub_dn)
def __safe_remove_from_group(self, uid, group_dn):
"""Remove user from group, deleting group if user is last member"""

10
nova/tests/auth_unittest.py
View File

@@ -135,10 +135,18 @@ class AuthTestCase(test.BaseTestCase):
self.manager.add_to_project('test2', 'testproj')
self.assertTrue(self.manager.get_project('testproj').has_member('test2'))
def test_208_can_remove_user_from_project(self):
def test_207_can_remove_user_from_project(self):
self.manager.remove_from_project('test2', 'testproj')
self.assertFalse(self.manager.get_project('testproj').has_member('test2'))
def test_208_can_remove_add_user_with_role(self):
self.manager.add_to_project('test2', 'testproj')
self.manager.add_role('test2', 'developer', 'testproj')
self.manager.remove_from_project('test2', 'testproj')
self.assertFalse(self.manager.has_role('test2', 'developer', 'testproj'))
self.manager.add_to_project('test2', 'testproj')
self.manager.remove_from_project('test2', 'testproj')
def test_209_can_generate_x509(self):
# MUST HAVE RUN CLOUD SETUP BY NOW
self.cloud = cloud.CloudController()