|
|
@@ -28,7 +28,10 @@ import ldap
|
|
|
from nose.plugins.attrib import attr
|
|
|
from nose.tools import assert_true, assert_equal, assert_false
|
|
|
|
|
|
+import desktop.conf
|
|
|
+from desktop.lib.test_utils import grant_access
|
|
|
from desktop.lib.django_test_util import make_logged_in_client
|
|
|
+from django.conf import settings
|
|
|
from django.contrib.auth.models import User, Group
|
|
|
from django.utils.encoding import smart_unicode
|
|
|
from django.core.urlresolvers import reverse
|
|
|
@@ -60,48 +63,45 @@ class LdapTestConnection(object):
|
|
|
It also includes functionality to fake modifications to an LDAP server. It is designed
|
|
|
as a singleton, to allow for changes to persist across discrete connections.
|
|
|
"""
|
|
|
-
|
|
|
- _instance = None
|
|
|
def __init__(self):
|
|
|
- if LdapTestConnection._instance is None:
|
|
|
- LdapTestConnection._instance = LdapTestConnection._Singleton()
|
|
|
+ self._instance = LdapTestConnection.Data()
|
|
|
|
|
|
def add_user_group_for_test(self, user, group):
|
|
|
- LdapTestConnection._instance.groups[group]['members'].append(user)
|
|
|
+ self._instance.groups[group]['members'].append(user)
|
|
|
|
|
|
def remove_user_group_for_test(self, user, group):
|
|
|
- LdapTestConnection._instance.groups[group]['members'].remove(user)
|
|
|
+ self._instance.groups[group]['members'].remove(user)
|
|
|
|
|
|
def find_users(self, username_pattern, find_by_dn=False, scope=ldap.SCOPE_SUBTREE):
|
|
|
""" Returns info for a particular user """
|
|
|
if find_by_dn:
|
|
|
- data = filter(lambda attrs: attrs['dn'] == username_pattern, LdapTestConnection._instance.users.values())
|
|
|
+ data = filter(lambda attrs: attrs['dn'] == username_pattern, self._instance.users.values())
|
|
|
else:
|
|
|
username_pattern = username_pattern.replace('.','\\.').replace('*', '.*')
|
|
|
- usernames = filter(lambda username: re.match(username_pattern, username), LdapTestConnection._instance.users.keys())
|
|
|
- data = [LdapTestConnection._instance.users.get(username) for username in usernames]
|
|
|
+ usernames = filter(lambda username: re.match(username_pattern, username), self._instance.users.keys())
|
|
|
+ data = [self._instance.users.get(username) for username in usernames]
|
|
|
return data
|
|
|
|
|
|
def find_groups(self, groupname_pattern, find_by_dn=False, scope=ldap.SCOPE_SUBTREE):
|
|
|
""" Return all groups in the system with parents and children """
|
|
|
if find_by_dn:
|
|
|
- data = filter(lambda attrs: attrs['dn'] == groupname_pattern, LdapTestConnection._instance.groups.values())
|
|
|
+ data = filter(lambda attrs: attrs['dn'] == groupname_pattern, self._instance.groups.values())
|
|
|
# SCOPE_SUBTREE means we return all sub-entries of the desired entry along with the desired entry.
|
|
|
if data and scope == ldap.SCOPE_SUBTREE:
|
|
|
- sub_data = filter(lambda attrs: attrs['dn'].endswith(data[0]['dn']), LdapTestConnection._instance.groups.values())
|
|
|
+ sub_data = filter(lambda attrs: attrs['dn'].endswith(data[0]['dn']), self._instance.groups.values())
|
|
|
data.extend(sub_data)
|
|
|
else:
|
|
|
groupname_pattern = groupname_pattern.replace('.','\\.').replace('*', '.*')
|
|
|
- groupnames = filter(lambda username: re.match(groupname_pattern, username), LdapTestConnection._instance.groups.keys())
|
|
|
- data = [LdapTestConnection._instance.groups.get(groupname) for groupname in groupnames]
|
|
|
+ groupnames = filter(lambda username: re.match(groupname_pattern, username), self._instance.groups.keys())
|
|
|
+ data = [self._instance.groups.get(groupname) for groupname in groupnames]
|
|
|
return data
|
|
|
|
|
|
- class _Singleton:
|
|
|
+ class Data:
|
|
|
def __init__(self):
|
|
|
- self.users = {'moe': {'dn': 'uid=moe,ou=People,dc=example,dc=com', 'username':'moe', 'first':'Moe', 'email':'moe@stooges.com'},
|
|
|
- 'lårry': {'dn': 'uid=lårry,ou=People,dc=example,dc=com', 'username':'lårry', 'first':'Larry', 'last':'Stooge', 'email':'larry@stooges.com'},
|
|
|
- 'curly': {'dn': 'uid=curly,ou=People,dc=example,dc=com', 'username':'curly', 'first':'Curly', 'last':'Stooge', 'email':'curly@stooges.com'},
|
|
|
- 'rock': {'dn': 'uid=rock,ou=People,dc=example,dc=com', 'username':'rock', 'first':'rock', 'last':'man', 'email':'rockman@stooges.com'},
|
|
|
+ self.users = {'moe': {'dn': 'uid=moe,ou=People,dc=example,dc=com', 'username':'moe', 'first':'Moe', 'email':'moe@stooges.com', 'groups': ['cn=TestUsers,ou=Groups,dc=example,dc=com']},
|
|
|
+ 'lårry': {'dn': 'uid=lårry,ou=People,dc=example,dc=com', 'username':'lårry', 'first':'Larry', 'last':'Stooge', 'email':'larry@stooges.com', 'groups': ['cn=TestUsers,ou=Groups,dc=example,dc=com', 'cn=Test Administrators,cn=TestUsers,ou=Groups,dc=example,dc=com']},
|
|
|
+ 'curly': {'dn': 'uid=curly,ou=People,dc=example,dc=com', 'username':'curly', 'first':'Curly', 'last':'Stooge', 'email':'curly@stooges.com', 'groups': ['cn=TestUsers,ou=Groups,dc=example,dc=com', 'cn=Test Administrators,cn=TestUsers,ou=Groups,dc=example,dc=com']},
|
|
|
+ 'rock': {'dn': 'uid=rock,ou=People,dc=example,dc=com', 'username':'rock', 'first':'rock', 'last':'man', 'email':'rockman@stooges.com', 'groups': ['cn=Test Administrators,cn=TestUsers,ou=Groups,dc=example,dc=com']},
|
|
|
'otherguy': {'dn': 'uid=otherguy,ou=People,dc=example,dc=com', 'username':'otherguy', 'first':'Other', 'last':'Guy', 'email':'other@guy.com'}}
|
|
|
|
|
|
self.groups = {'TestUsers': {
|
|
|
@@ -435,6 +435,56 @@ def test_user_admin():
|
|
|
assert_true("You must specify a password when creating a new user." in response.content)
|
|
|
|
|
|
|
|
|
+def test_useradmin_ldap_user_group_membership_sync():
|
|
|
+ reset = [desktop.conf.AUTH.USER_GROUP_MEMBERSHIP_SYNCHRONIZATION_BACKEND.set_for_testing('desktop.auth.backend.LdapSynchronizationBackend')]
|
|
|
+ settings.MIDDLEWARE_CLASSES.append('desktop.middleware.UserGroupSynchronizationMiddleware')
|
|
|
+
|
|
|
+ reset_all_users()
|
|
|
+ reset_all_groups()
|
|
|
+
|
|
|
+ # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
|
|
|
+ ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
|
|
|
+
|
|
|
+ try:
|
|
|
+ # Import curly who is part of TestUsers and Test Administrators
|
|
|
+ import_ldap_users('curly', sync_groups=False, import_by_dn=False)
|
|
|
+
|
|
|
+ # Set a password so that we can login
|
|
|
+ user = User.objects.get(username='curly')
|
|
|
+ user.set_password('test')
|
|
|
+ user.save()
|
|
|
+
|
|
|
+ # Should have 0 groups
|
|
|
+ assert_equal(0, user.groups.all().count())
|
|
|
+
|
|
|
+ # Make an authenticated request as curly so that we can see call middleware.
|
|
|
+ c = make_logged_in_client('curly', 'test', is_superuser=False)
|
|
|
+ grant_access("curly", "test", "useradmin")
|
|
|
+ response = c.get('/useradmin/users')
|
|
|
+
|
|
|
+ # Refresh user groups
|
|
|
+ user = User.objects.get(username='curly')
|
|
|
+
|
|
|
+ # Should have 3 groups now. 2 from LDAP and 1 from 'grant_access' call.
|
|
|
+ assert_equal(3, user.groups.all().count(), user.groups.all())
|
|
|
+
|
|
|
+ # Now remove a group and try again.
|
|
|
+ old_group = ldap_access.CACHED_LDAP_CONN._instance.users['curly']['groups'].pop()
|
|
|
+
|
|
|
+ # Make an authenticated request as curly so that we can see call middleware.
|
|
|
+ response = c.get('/useradmin/users')
|
|
|
+
|
|
|
+ # Refresh user groups
|
|
|
+ user = User.objects.get(username='curly')
|
|
|
+
|
|
|
+ # Should have 2 groups now. 1 from LDAP and 1 from 'grant_access' call.
|
|
|
+ assert_equal(3, user.groups.all().count(), user.groups.all())
|
|
|
+ finally:
|
|
|
+ for finish in reset:
|
|
|
+ finish()
|
|
|
+ settings.MIDDLEWARE_CLASSES.remove('desktop.middleware.UserGroupSynchronizationMiddleware')
|
|
|
+
|
|
|
+
|
|
|
def test_useradmin_ldap_group_integration():
|
|
|
reset_all_users()
|
|
|
reset_all_groups()
|
|
|
@@ -442,45 +492,51 @@ def test_useradmin_ldap_group_integration():
|
|
|
# Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
|
|
|
ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
|
|
|
|
|
|
+ # Import groups only
|
|
|
+ import_ldap_groups('TestUsers', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
|
|
|
+ test_users = Group.objects.get(name='TestUsers')
|
|
|
+ assert_true(LdapGroup.objects.filter(group=test_users).exists())
|
|
|
+ assert_equal(test_users.user_set.all().count(), 0)
|
|
|
+
|
|
|
# Import all members of TestUsers
|
|
|
- import_ldap_groups('TestUsers', import_members=True, import_members_recursive=False, import_by_dn=False)
|
|
|
+ import_ldap_groups('TestUsers', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
|
|
|
test_users = Group.objects.get(name='TestUsers')
|
|
|
assert_true(LdapGroup.objects.filter(group=test_users).exists())
|
|
|
- assert_equal(len(test_users.user_set.all()), 3)
|
|
|
+ assert_equal(test_users.user_set.all().count(), 3)
|
|
|
|
|
|
# Should import a group, but will only sync already-imported members
|
|
|
- import_ldap_groups('Test Administrators', import_members=False, import_members_recursive=False, import_by_dn=False)
|
|
|
- assert_equal(len(User.objects.all()), 3)
|
|
|
- assert_equal(len(Group.objects.all()), 2)
|
|
|
+ import_ldap_groups('Test Administrators', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
|
|
|
+ assert_equal(User.objects.all().count(), 3)
|
|
|
+ assert_equal(Group.objects.all().count(), 2)
|
|
|
test_admins = Group.objects.get(name='Test Administrators')
|
|
|
- assert_equal(len(test_admins.user_set.all()), 2)
|
|
|
+ assert_equal(test_admins.user_set.all().count(), 2)
|
|
|
larry = User.objects.get(username='lårry')
|
|
|
assert_equal(test_admins.user_set.all()[0].username, larry.username)
|
|
|
|
|
|
# Only sync already imported
|
|
|
ldap_access.CACHED_LDAP_CONN.remove_user_group_for_test('uid=moe,ou=People,dc=example,dc=com', 'TestUsers')
|
|
|
- import_ldap_groups('TestUsers', import_members=False, import_members_recursive=False, import_by_dn=False)
|
|
|
- assert_equal(len(test_users.user_set.all()), 2)
|
|
|
- assert_equal(len(User.objects.get(username='moe').groups.all()), 0)
|
|
|
+ import_ldap_groups('TestUsers', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
|
|
|
+ assert_equal(test_users.user_set.all().count(), 2)
|
|
|
+ assert_equal(User.objects.get(username='moe').groups.all().count(), 0)
|
|
|
|
|
|
# Import missing user
|
|
|
ldap_access.CACHED_LDAP_CONN.add_user_group_for_test('uid=moe,ou=People,dc=example,dc=com', 'TestUsers')
|
|
|
- import_ldap_groups('TestUsers', import_members=True, import_members_recursive=False, import_by_dn=False)
|
|
|
- assert_equal(len(test_users.user_set.all()), 3)
|
|
|
- assert_equal(len(User.objects.get(username='moe').groups.all()), 1)
|
|
|
+ import_ldap_groups('TestUsers', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
|
|
|
+ assert_equal(test_users.user_set.all().count(), 3)
|
|
|
+ assert_equal(User.objects.get(username='moe').groups.all().count(), 1)
|
|
|
|
|
|
# Import all members of TestUsers and members of subgroups
|
|
|
- import_ldap_groups('TestUsers', import_members=True, import_members_recursive=True, import_by_dn=False)
|
|
|
+ import_ldap_groups('TestUsers', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
|
|
|
test_users = Group.objects.get(name='TestUsers')
|
|
|
assert_true(LdapGroup.objects.filter(group=test_users).exists())
|
|
|
- assert_equal(len(test_users.user_set.all()), 4)
|
|
|
+ assert_equal(test_users.user_set.all().count(), 4)
|
|
|
|
|
|
# Make sure Hue groups with naming collisions don't get marked as LDAP groups
|
|
|
hue_user = User.objects.create(username='otherguy', first_name='Different', last_name='Guy')
|
|
|
hue_group = Group.objects.create(name='OtherGroup')
|
|
|
hue_group.user_set.add(hue_user)
|
|
|
hue_group.save()
|
|
|
- import_ldap_groups('OtherGroup', import_members=False, import_members_recursive=False, import_by_dn=False)
|
|
|
+ import_ldap_groups('OtherGroup', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
|
|
|
assert_false(LdapGroup.objects.filter(group=hue_group).exists())
|
|
|
assert_true(hue_group.user_set.filter(username=hue_user.username).exists())
|
|
|
|
|
|
@@ -492,7 +548,7 @@ def test_useradmin_ldap_user_integration():
|
|
|
ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
|
|
|
|
|
|
# Try importing a user
|
|
|
- import_ldap_users('lårry', import_by_dn=False)
|
|
|
+ import_ldap_users('lårry', sync_groups=False, import_by_dn=False)
|
|
|
larry = User.objects.get(username='lårry')
|
|
|
assert_true(larry.first_name == 'Larry')
|
|
|
assert_true(larry.last_name == 'Stooge')
|
|
|
@@ -502,17 +558,26 @@ def test_useradmin_ldap_user_integration():
|
|
|
# Should be a noop
|
|
|
sync_ldap_users()
|
|
|
sync_ldap_groups()
|
|
|
- assert_equal(len(User.objects.all()), 1)
|
|
|
- assert_equal(len(Group.objects.all()), 0)
|
|
|
+ assert_equal(User.objects.all().count(), 1)
|
|
|
+ assert_equal(Group.objects.all().count(), 0)
|
|
|
|
|
|
# Make sure that if a Hue user already exists with a naming collision, we
|
|
|
# won't overwrite any of that user's information.
|
|
|
hue_user = User.objects.create(username='otherguy', first_name='Different', last_name='Guy')
|
|
|
- import_ldap_users('otherguy', import_by_dn=False)
|
|
|
+ import_ldap_users('otherguy', sync_groups=False, import_by_dn=False)
|
|
|
hue_user = User.objects.get(username='otherguy')
|
|
|
assert_equal(get_profile(hue_user).creation_method, str(UserProfile.CreationMethod.HUE))
|
|
|
assert_equal(hue_user.first_name, 'Different')
|
|
|
|
|
|
+ # Try importing a user and sync groups
|
|
|
+ import_ldap_users('curly', sync_groups=True, import_by_dn=False)
|
|
|
+ curly = User.objects.get(username='curly')
|
|
|
+ assert_equal(curly.first_name, 'Curly')
|
|
|
+ assert_equal(curly.last_name, 'Stooge')
|
|
|
+ assert_equal(curly.email, 'curly@stooges.com')
|
|
|
+ assert_equal(get_profile(curly).creation_method, str(UserProfile.CreationMethod.EXTERNAL))
|
|
|
+ assert_equal(2, curly.groups.all().count(), curly.groups.all())
|
|
|
+
|
|
|
|
|
|
def test_add_ldap_users():
|
|
|
URL = reverse(add_ldap_users)
|