Browse Source

[useradmin] Make useradmin tests reentrant

Before this patch, the useradmin tests were not reentrant
because it never restored the DEFAULT_USER_GROUP config option
to `"default"` after it finished running.
Erick Tryzelaar 10 năm trước cách đây
mục cha
commit
d295c91

+ 7 - 5
apps/useradmin/src/useradmin/models.py

@@ -205,12 +205,14 @@ class HuePermission(models.Model):
 
 def get_default_user_group(**kwargs):
   default_user_group = useradmin.conf.DEFAULT_USER_GROUP.get()
-  if default_user_group is not None:
-    group, created = auth_models.Group.objects.get_or_create(name=default_user_group)
-    if created:
-      group.save()
+  if default_user_group is None:
+    return None
+
+  group, created = auth_models.Group.objects.get_or_create(name=default_user_group)
+  if created:
+    group.save()
 
-    return group
+  return group
 
 def update_app_permissions(**kwargs):
   """

+ 610 - 642
apps/useradmin/src/useradmin/test_ldap.py

@@ -38,7 +38,7 @@ from views import sync_ldap_users, sync_ldap_groups, import_ldap_users, import_l
                   add_ldap_users, add_ldap_groups, sync_ldap_users_groups
 
 import ldap_access
-from tests import LdapTestConnection, reset_all_groups, reset_all_users
+from tests import BaseUserAdminTests, LdapTestConnection, reset_all_groups, reset_all_users
 
 
 def get_nonsense_config():
@@ -48,709 +48,677 @@ def get_nonsense_config():
   }}
 
 
-def test_useradmin_ldap_user_group_membership_sync():
-  settings.MIDDLEWARE_CLASSES.append('useradmin.middleware.LdapSynchronizationMiddleware')
+class TestUserAdminLdap(BaseUserAdminTests):
+  def test_useradmin_ldap_user_group_membership_sync(self):
+    settings.MIDDLEWARE_CLASSES.append('useradmin.middleware.LdapSynchronizationMiddleware')
 
-  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()
-  # Make sure LDAP groups exist or they won't sync
-  import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'TestUsers', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
-  import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
-
-  reset = []
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+    # Make sure LDAP groups exist or they won't sync
+    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'TestUsers', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
+    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
 
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+    reset = []
 
-  try:
-    # Import curly who is part of TestUsers and Test Administrators
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'curly', sync_groups=False, import_by_dn=False)
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
 
-    # Set a password so that we can login
-    user = User.objects.get(username='curly')
-    user.set_password('test')
-    user.save()
+    try:
+      # Import curly who is part of TestUsers and Test Administrators
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'curly', sync_groups=False, import_by_dn=False)
 
-    # Should have 0 groups
-    assert_equal(0, user.groups.all().count())
+      # Set a password so that we can login
+      user = User.objects.get(username='curly')
+      user.set_password('test')
+      user.save()
 
-    # 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')
+      # Should have 0 groups
+      assert_equal(0, user.groups.all().count())
 
-    # Refresh user groups
-    user = User.objects.get(username='curly')
+      # 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')
 
-    # Should have 3 groups now. 2 from LDAP and 1 from 'grant_access' call.
-    assert_equal(3, user.groups.all().count(), user.groups.all())
+      # Refresh user groups
+      user = User.objects.get(username='curly')
 
-    # Now remove a group and try again.
-    old_group = ldap_access.CACHED_LDAP_CONN._instance.users['curly']['groups'].pop()
+      # Should have 3 groups now. 2 from LDAP and 1 from 'grant_access' call.
+      assert_equal(3, user.groups.all().count(), user.groups.all())
 
-    # Make an authenticated request as curly so that we can see call middleware.
-    response = c.get('/useradmin/users')
+      # Now remove a group and try again.
+      old_group = ldap_access.CACHED_LDAP_CONN._instance.users['curly']['groups'].pop()
 
-    # Refresh user groups
-    user = User.objects.get(username='curly')
+      # Make an authenticated request as curly so that we can see call middleware.
+      response = c.get('/useradmin/users')
 
-    # 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:
-    settings.MIDDLEWARE_CLASSES.remove('useradmin.middleware.LdapSynchronizationMiddleware')
+      # Refresh user groups
+      user = User.objects.get(username='curly')
 
-    for finish in reset:
-      finish()
+      # 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:
+      settings.MIDDLEWARE_CLASSES.remove('useradmin.middleware.LdapSynchronizationMiddleware')
 
+      for finish in reset:
+        finish()
 
-def test_useradmin_ldap_suboordinate_group_integration():
-  reset_all_users()
-  reset_all_groups()
 
-  reset = []
+  def test_useradmin_ldap_suboordinate_group_integration(self):
+    reset = []
 
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
-  # Test old subgroups
-  reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("suboordinate"))
+    # Test old subgroups
+    reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("suboordinate"))
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+
+    try:
+      # Import groups only
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(test_users.user_set.all().count(), 3)
+
+      # Should import a group, but will only sync already-imported members
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(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(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(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(ldap_access.CACHED_LDAP_CONN, '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())
+    finally:
+      for finish in reset:
+        finish()
+
+
+  def test_useradmin_ldap_nested_group_integration(self):
+    if is_live_cluster():
+      raise SkipTest('HUE-2897: Skipping because DB may not support unicode')
+
+    reset = []
 
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
-  try:
-    # Import groups only
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(test_users.user_set.all().count(), 3)
-
-    # Should import a group, but will only sync already-imported members
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(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(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(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(ldap_access.CACHED_LDAP_CONN, '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())
-  finally:
-    for finish in reset:
-      finish()
-
-
-def test_useradmin_ldap_nested_group_integration():
-  if is_live_cluster():
-    raise SkipTest('HUE-2897: Skipping because DB may not support unicode')
-
-  reset_all_users()
-  reset_all_groups()
-
-  reset = []
-
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
-
-  # Test old subgroups
-  reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("nested"))
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    # Import groups only
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(test_users.user_set.all().count(), 3)
-
-    # Should import a group, but will only sync already-imported members
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(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
-    assert_equal(test_users.user_set.all().count(), 3)
-    ldap_access.CACHED_LDAP_CONN.remove_user_group_for_test('uid=moe,ou=People,dc=example,dc=com', 'TestUsers')
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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 not members of suboordinate groups (even though specified)
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(test_users.user_set.all().count(), 3)
-
-    # Nested group import
-    # First without recursive import, then with.
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'NestedGroups', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    nested_groups = Group.objects.get(name='NestedGroups')
-    nested_group = Group.objects.get(name='NestedGroup')
-    assert_true(LdapGroup.objects.filter(group=nested_groups).exists())
-    assert_true(LdapGroup.objects.filter(group=nested_group).exists())
-    assert_equal(nested_groups.user_set.all().count(), 0, nested_groups.user_set.all())
-    assert_equal(nested_group.user_set.all().count(), 0, nested_group.user_set.all())
-
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'NestedGroups', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
-    nested_groups = Group.objects.get(name='NestedGroups')
-    nested_group = Group.objects.get(name='NestedGroup')
-    assert_true(LdapGroup.objects.filter(group=nested_groups).exists())
-    assert_true(LdapGroup.objects.filter(group=nested_group).exists())
-    assert_equal(nested_groups.user_set.all().count(), 0, nested_groups.user_set.all())
-    assert_equal(nested_group.user_set.all().count(), 1, nested_group.user_set.all())
-
-    # 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(ldap_access.CACHED_LDAP_CONN, '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())
-  finally:
-    for finish in reset:
-      finish()
-
-
-def test_useradmin_ldap_suboordinate_posix_group_integration():
-  reset_all_users()
-  reset_all_groups()
-
-  reset = []
-
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
-
-  # Test old subgroups
-  reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("suboordinate"))
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    # Import groups only
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
-    test_users = Group.objects.get(name='PosixGroup')
-    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(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    test_users = Group.objects.get(name='PosixGroup')
-    assert_true(LdapGroup.objects.filter(group=test_users).exists())
-    assert_equal(test_users.user_set.all().count(), 2)
-
-    # Should import a group, but will only sync already-imported members
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    assert_equal(User.objects.all().count(), 2, User.objects.all())
-    assert_equal(Group.objects.all().count(), 2, Group.objects.all())
-    test_admins = Group.objects.get(name='Test Administrators')
-    assert_equal(test_admins.user_set.all().count(), 1)
-    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_posix_user_group_for_test('posix_person', 'PosixGroup')
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    assert_equal(test_users.user_set.all().count(), 1)
-    assert_equal(User.objects.get(username='posix_person').groups.all().count(), 0)
-
-    # Import missing user
-    ldap_access.CACHED_LDAP_CONN.add_posix_user_group_for_test('posix_person', 'PosixGroup')
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, 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='posix_person').groups.all().count(), 1)
-
-    # Import all members of PosixGroup and members of subgroups
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
-    test_users = Group.objects.get(name='PosixGroup')
-    assert_true(LdapGroup.objects.filter(group=test_users).exists())
-    assert_equal(test_users.user_set.all().count(), 3)
-
-    # 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(ldap_access.CACHED_LDAP_CONN, '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())
-  finally:
-    for finish in reset:
-      finish()
-
-
-def test_useradmin_ldap_nested_posix_group_integration():
-  reset_all_users()
-  reset_all_groups()
-
-  reset = []
-
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
-
-  # Test nested groups
-  reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("nested"))
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    # Import groups only
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
-    test_users = Group.objects.get(name='PosixGroup')
-    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(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    test_users = Group.objects.get(name='PosixGroup')
-    assert_true(LdapGroup.objects.filter(group=test_users).exists())
-    assert_equal(test_users.user_set.all().count(), 2)
-
-    # Should import a group, but will only sync already-imported members
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    assert_equal(User.objects.all().count(), 2, User.objects.all())
-    assert_equal(Group.objects.all().count(), 2, Group.objects.all())
-    test_admins = Group.objects.get(name='Test Administrators')
-    assert_equal(test_admins.user_set.all().count(), 1)
-    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_posix_user_group_for_test('posix_person', 'PosixGroup')
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
-    assert_equal(test_users.user_set.all().count(), 1)
-    assert_equal(User.objects.get(username='posix_person').groups.all().count(), 0)
-
-    # Import missing user
-    ldap_access.CACHED_LDAP_CONN.add_posix_user_group_for_test('posix_person', 'PosixGroup')
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, 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='posix_person').groups.all().count(), 1)
-
-    # Import all members of PosixGroup and members of subgroups (there should be no subgroups)
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
-    test_users = Group.objects.get(name='PosixGroup')
-    assert_true(LdapGroup.objects.filter(group=test_users).exists())
-    assert_equal(test_users.user_set.all().count(), 2)
-
-    # Import all members of NestedPosixGroups and members of subgroups
-    reset_all_users()
-    reset_all_groups()
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'NestedPosixGroups', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
-    test_users = Group.objects.get(name='NestedPosixGroups')
-    assert_true(LdapGroup.objects.filter(group=test_users).exists())
-    assert_equal(test_users.user_set.all().count(), 0)
-    test_users = Group.objects.get(name='PosixGroup')
-    assert_true(LdapGroup.objects.filter(group=test_users).exists())
-    assert_equal(test_users.user_set.all().count(), 2)
-
-    # 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(ldap_access.CACHED_LDAP_CONN, '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())
-  finally:
-    for finish in reset:
-      finish()
-
-
-def test_useradmin_ldap_user_integration():
-  if is_live_cluster():
-    raise SkipTest('HUE-2897: Skipping because DB may not support unicode')
-
-  done = []
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  done.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    reset_all_users()
-    reset_all_groups()
+    # Test old subgroups
+    reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("nested"))
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+
+    try:
+      # Import groups only
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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(test_users.user_set.all().count(), 3)
+
+      # Should import a group, but will only sync already-imported members
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(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
+      assert_equal(test_users.user_set.all().count(), 3)
+      ldap_access.CACHED_LDAP_CONN.remove_user_group_for_test('uid=moe,ou=People,dc=example,dc=com', 'TestUsers')
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(ldap_access.CACHED_LDAP_CONN, '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 not members of suboordinate groups (even though specified)
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, '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(test_users.user_set.all().count(), 3)
+
+      # Nested group import
+      # First without recursive import, then with.
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'NestedGroups', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      nested_groups = Group.objects.get(name='NestedGroups')
+      nested_group = Group.objects.get(name='NestedGroup')
+      assert_true(LdapGroup.objects.filter(group=nested_groups).exists())
+      assert_true(LdapGroup.objects.filter(group=nested_group).exists())
+      assert_equal(nested_groups.user_set.all().count(), 0, nested_groups.user_set.all())
+      assert_equal(nested_group.user_set.all().count(), 0, nested_group.user_set.all())
+
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'NestedGroups', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
+      nested_groups = Group.objects.get(name='NestedGroups')
+      nested_group = Group.objects.get(name='NestedGroup')
+      assert_true(LdapGroup.objects.filter(group=nested_groups).exists())
+      assert_true(LdapGroup.objects.filter(group=nested_group).exists())
+      assert_equal(nested_groups.user_set.all().count(), 0, nested_groups.user_set.all())
+      assert_equal(nested_group.user_set.all().count(), 1, nested_group.user_set.all())
+
+      # 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(ldap_access.CACHED_LDAP_CONN, '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())
+    finally:
+      for finish in reset:
+        finish()
+
+
+  def test_useradmin_ldap_suboordinate_posix_group_integration(self):
+    reset = []
 
     # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
     ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
-    # Try importing a user
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, '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')
-    assert_true(larry.email == 'larry@stooges.com')
-    assert_true(get_profile(larry).creation_method == str(UserProfile.CreationMethod.EXTERNAL))
-
-    # Should be a noop
-    sync_ldap_users(ldap_access.CACHED_LDAP_CONN)
-    sync_ldap_groups(ldap_access.CACHED_LDAP_CONN)
-    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(ldap_access.CACHED_LDAP_CONN, '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')
+    # Test old subgroups
+    reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("suboordinate"))
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+
+    try:
+      # Import groups only
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
+      test_users = Group.objects.get(name='PosixGroup')
+      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(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      test_users = Group.objects.get(name='PosixGroup')
+      assert_true(LdapGroup.objects.filter(group=test_users).exists())
+      assert_equal(test_users.user_set.all().count(), 2)
+
+      # Should import a group, but will only sync already-imported members
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      assert_equal(User.objects.all().count(), 2, User.objects.all())
+      assert_equal(Group.objects.all().count(), 2, Group.objects.all())
+      test_admins = Group.objects.get(name='Test Administrators')
+      assert_equal(test_admins.user_set.all().count(), 1)
+      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_posix_user_group_for_test('posix_person', 'PosixGroup')
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      assert_equal(test_users.user_set.all().count(), 1)
+      assert_equal(User.objects.get(username='posix_person').groups.all().count(), 0)
+
+      # Import missing user
+      ldap_access.CACHED_LDAP_CONN.add_posix_user_group_for_test('posix_person', 'PosixGroup')
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, 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='posix_person').groups.all().count(), 1)
+
+      # Import all members of PosixGroup and members of subgroups
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
+      test_users = Group.objects.get(name='PosixGroup')
+      assert_true(LdapGroup.objects.filter(group=test_users).exists())
+      assert_equal(test_users.user_set.all().count(), 3)
+
+      # 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(ldap_access.CACHED_LDAP_CONN, '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())
+    finally:
+      for finish in reset:
+        finish()
+
+
+  def test_useradmin_ldap_nested_posix_group_integration(self):
+    reset = []
 
-    # Make sure LDAP groups exist or they won't sync
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'TestUsers', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
-    import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
-    # Try importing a user and sync groups
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, '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())
-
-    reset_all_users()
-    reset_all_groups()
-
-    # Test import case sensitivity
-    done.append(desktop.conf.LDAP.IGNORE_USERNAME_CASE.set_for_testing(True))
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Lårry', sync_groups=False, import_by_dn=False)
-    assert_false(User.objects.filter(username='Lårry').exists())
-    assert_true(User.objects.filter(username='lårry').exists())
-
-    # Test lower case
-    User.objects.filter(username__iexact='Rock').delete()
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Rock', sync_groups=False, import_by_dn=False)
-    assert_false(User.objects.filter(username='Rock').exists())
-    assert_true(User.objects.filter(username='rock').exists())
-
-    done.append(desktop.conf.LDAP.FORCE_USERNAME_LOWERCASE.set_for_testing(True))
-
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Rock', sync_groups=False, import_by_dn=False)
-    assert_false(User.objects.filter(username='Rock').exists())
-    assert_true(User.objects.filter(username='rock').exists())
-
-    User.objects.filter(username='Rock').delete()
-    import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Rock', sync_groups=False, import_by_dn=False)
-    assert_false(User.objects.filter(username='Rock').exists())
-    assert_true(User.objects.filter(username='rock').exists())
-  finally:
-    for finish in done:
-      finish()
-
-
-def test_add_ldap_users():
-  if is_live_cluster():
-    raise SkipTest('HUE-2897: Skipping because the DB may not be case sensitive')
-
-  done = []
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  done.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    URL = reverse(add_ldap_users)
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
-    reset_all_users()
-    reset_all_groups()
+    # Test nested groups
+    reset.append(desktop.conf.LDAP.SUBGROUPS.set_for_testing("nested"))
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+
+    try:
+      # Import groups only
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
+      test_users = Group.objects.get(name='PosixGroup')
+      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(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      test_users = Group.objects.get(name='PosixGroup')
+      assert_true(LdapGroup.objects.filter(group=test_users).exists())
+      assert_equal(test_users.user_set.all().count(), 2)
+
+      # Should import a group, but will only sync already-imported members
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      assert_equal(User.objects.all().count(), 2, User.objects.all())
+      assert_equal(Group.objects.all().count(), 2, Group.objects.all())
+      test_admins = Group.objects.get(name='Test Administrators')
+      assert_equal(test_admins.user_set.all().count(), 1)
+      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_posix_user_group_for_test('posix_person', 'PosixGroup')
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=False, import_members_recursive=False, sync_users=True, import_by_dn=False)
+      assert_equal(test_users.user_set.all().count(), 1)
+      assert_equal(User.objects.get(username='posix_person').groups.all().count(), 0)
+
+      # Import missing user
+      ldap_access.CACHED_LDAP_CONN.add_posix_user_group_for_test('posix_person', 'PosixGroup')
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, 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='posix_person').groups.all().count(), 1)
+
+      # Import all members of PosixGroup and members of subgroups (there should be no subgroups)
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'PosixGroup', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
+      test_users = Group.objects.get(name='PosixGroup')
+      assert_true(LdapGroup.objects.filter(group=test_users).exists())
+      assert_equal(test_users.user_set.all().count(), 2)
+
+      # Import all members of NestedPosixGroups and members of subgroups
+      reset_all_users()
+      reset_all_groups()
+
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'NestedPosixGroups', import_members=True, import_members_recursive=True, sync_users=True, import_by_dn=False)
+      test_users = Group.objects.get(name='NestedPosixGroups')
+      assert_true(LdapGroup.objects.filter(group=test_users).exists())
+      assert_equal(test_users.user_set.all().count(), 0)
+      test_users = Group.objects.get(name='PosixGroup')
+      assert_true(LdapGroup.objects.filter(group=test_users).exists())
+      assert_equal(test_users.user_set.all().count(), 2)
+
+      # 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(ldap_access.CACHED_LDAP_CONN, '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())
+    finally:
+      for finish in reset:
+        finish()
+
+
+  def test_useradmin_ldap_user_integration(self):
+    if is_live_cluster():
+      raise SkipTest('HUE-2897: Skipping because DB may not support unicode')
+
+    done = []
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    done.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+
+    try:
+      # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+      ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+
+      # Try importing a user
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, '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')
+      assert_true(larry.email == 'larry@stooges.com')
+      assert_true(get_profile(larry).creation_method == str(UserProfile.CreationMethod.EXTERNAL))
+
+      # Should be a noop
+      sync_ldap_users(ldap_access.CACHED_LDAP_CONN)
+      sync_ldap_groups(ldap_access.CACHED_LDAP_CONN)
+      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(ldap_access.CACHED_LDAP_CONN, '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')
+
+      # Make sure LDAP groups exist or they won't sync
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'TestUsers', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
+      import_ldap_groups(ldap_access.CACHED_LDAP_CONN, 'Test Administrators', import_members=False, import_members_recursive=False, sync_users=False, import_by_dn=False)
+      # Try importing a user and sync groups
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, '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())
+
+      reset_all_users()
+      reset_all_groups()
+
+      # Test import case sensitivity
+      done.append(desktop.conf.LDAP.IGNORE_USERNAME_CASE.set_for_testing(True))
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Lårry', sync_groups=False, import_by_dn=False)
+      assert_false(User.objects.filter(username='Lårry').exists())
+      assert_true(User.objects.filter(username='lårry').exists())
+
+      # Test lower case
+      User.objects.filter(username__iexact='Rock').delete()
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Rock', sync_groups=False, import_by_dn=False)
+      assert_false(User.objects.filter(username='Rock').exists())
+      assert_true(User.objects.filter(username='rock').exists())
+
+      done.append(desktop.conf.LDAP.FORCE_USERNAME_LOWERCASE.set_for_testing(True))
+
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Rock', sync_groups=False, import_by_dn=False)
+      assert_false(User.objects.filter(username='Rock').exists())
+      assert_true(User.objects.filter(username='rock').exists())
+
+      User.objects.filter(username='Rock').delete()
+      import_ldap_users(ldap_access.CACHED_LDAP_CONN, 'Rock', sync_groups=False, import_by_dn=False)
+      assert_false(User.objects.filter(username='Rock').exists())
+      assert_true(User.objects.filter(username='rock').exists())
+    finally:
+      for finish in done:
+        finish()
+
+
+  def test_add_ldap_users(self):
+    if is_live_cluster():
+      raise SkipTest('HUE-2897: Skipping because the DB may not be case sensitive')
+
+    done = []
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    done.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+
+    try:
+      URL = reverse(add_ldap_users)
+
+      # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+      ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+
+      c = make_logged_in_client('test', is_superuser=True)
+
+      assert_true(c.get(URL))
+
+      response = c.post(URL, dict(server='nonsense', username_pattern='moe', password1='test', password2='test'))
+      assert_true('Location' in response, response)
+      assert_true('/useradmin/users' in response['Location'], response)
+
+      response = c.post(URL, dict(server='nonsense', username_pattern='bad_name', password1='test', password2='test'))
+      assert_true('Could not' in response.context['form'].errors['username_pattern'][0], response)
+
+      # Test wild card
+      response = c.post(URL, dict(server='nonsense', username_pattern='*rr*', password1='test', password2='test'))
+      assert_true('/useradmin/users' in response['Location'], response)
+
+      # Test ignore case
+      done.append(desktop.conf.LDAP.IGNORE_USERNAME_CASE.set_for_testing(True))
+      User.objects.filter(username='moe').delete()
+      assert_false(User.objects.filter(username='Moe').exists())
+      assert_false(User.objects.filter(username='moe').exists())
+      response = c.post(URL, dict(server='nonsense', username_pattern='Moe', password1='test', password2='test'))
+      assert_true('Location' in response, response)
+      assert_true('/useradmin/users' in response['Location'], response)
+      assert_false(User.objects.filter(username='Moe').exists())
+      assert_true(User.objects.filter(username='moe').exists())
+
+      # Test lower case
+      done.append(desktop.conf.LDAP.FORCE_USERNAME_LOWERCASE.set_for_testing(True))
+      User.objects.filter(username__iexact='Rock').delete()
+      assert_false(User.objects.filter(username='Rock').exists())
+      assert_false(User.objects.filter(username='rock').exists())
+      response = c.post(URL, dict(server='nonsense', username_pattern='rock', password1='test', password2='test'))
+      assert_true('Location' in response, response)
+      assert_true('/useradmin/users' in response['Location'], response)
+      assert_false(User.objects.filter(username='Rock').exists())
+      assert_true(User.objects.filter(username='rock').exists())
+
+      # Test regular with spaces (should fail)
+      response = c.post(URL, dict(server='nonsense', username_pattern='user with space', password1='test', password2='test'))
+      assert_true("Username must not contain whitespaces and ':'" in response.context['form'].errors['username_pattern'][0], response)
+
+      # Test dn with spaces in username and dn (should fail)
+      response = c.post(URL, dict(server='nonsense', username_pattern='uid=user with space,ou=People,dc=example,dc=com', password1='test', password2='test', dn=True))
+      assert_true("Could not get LDAP details for users in pattern" in response.content, response.content)
+      response = c.get(reverse(desktop.views.log_view))
+      assert_true("{username}: Username must not contain whitespaces".format(username='user with space') in response.content, response.content)
+
+      # Test dn with spaces in dn, but not username (should succeed)
+      response = c.post(URL, dict(server='nonsense', username_pattern='uid=user without space,ou=People,dc=example,dc=com', password1='test', password2='test', dn=True))
+      assert_true(User.objects.filter(username='spaceless').exists())
+
+    finally:
+      for finish in done:
+        finish()
+
+
+  def test_add_ldap_groups(self):
+    URL = reverse(add_ldap_groups)
 
     # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
     ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
-    c = make_logged_in_client('test', is_superuser=True)
+    c = make_logged_in_client(username='test', is_superuser=True)
 
-    assert_true(c.get(URL))
+    reset = []
 
-    response = c.post(URL, dict(server='nonsense', username_pattern='moe', password1='test', password2='test'))
-    assert_true('Location' in response, response)
-    assert_true('/useradmin/users' in response['Location'], response)
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
 
-    response = c.post(URL, dict(server='nonsense', username_pattern='bad_name', password1='test', password2='test'))
-    assert_true('Could not' in response.context['form'].errors['username_pattern'][0], response)
+    try:
 
-    # Test wild card
-    response = c.post(URL, dict(server='nonsense', username_pattern='*rr*', password1='test', password2='test'))
-    assert_true('/useradmin/users' in response['Location'], response)
+      assert_true(c.get(URL))
 
-    # Test ignore case
-    done.append(desktop.conf.LDAP.IGNORE_USERNAME_CASE.set_for_testing(True))
-    User.objects.filter(username='moe').delete()
-    assert_false(User.objects.filter(username='Moe').exists())
-    assert_false(User.objects.filter(username='moe').exists())
-    response = c.post(URL, dict(server='nonsense', username_pattern='Moe', password1='test', password2='test'))
-    assert_true('Location' in response, response)
-    assert_true('/useradmin/users' in response['Location'], response)
-    assert_false(User.objects.filter(username='Moe').exists())
-    assert_true(User.objects.filter(username='moe').exists())
+      response = c.post(URL, dict(server='nonsense', groupname_pattern='TestUsers'))
+      assert_true('Location' in response, response)
+      assert_true('/useradmin/groups' in response['Location'])
 
-    # Test lower case
-    done.append(desktop.conf.LDAP.FORCE_USERNAME_LOWERCASE.set_for_testing(True))
-    User.objects.filter(username__iexact='Rock').delete()
-    assert_false(User.objects.filter(username='Rock').exists())
-    assert_false(User.objects.filter(username='rock').exists())
-    response = c.post(URL, dict(server='nonsense', username_pattern='rock', password1='test', password2='test'))
-    assert_true('Location' in response, response)
-    assert_true('/useradmin/users' in response['Location'], response)
-    assert_false(User.objects.filter(username='Rock').exists())
-    assert_true(User.objects.filter(username='rock').exists())
+      # Test with space
+      response = c.post(URL, dict(server='nonsense', groupname_pattern='Test Administrators'))
+      assert_true('Location' in response, response)
+      assert_true('/useradmin/groups' in response['Location'], response)
 
-    # Test regular with spaces (should fail)
-    response = c.post(URL, dict(server='nonsense', username_pattern='user with space', password1='test', password2='test'))
-    assert_true("Username must not contain whitespaces and ':'" in response.context['form'].errors['username_pattern'][0], response)
+      response = c.post(URL, dict(server='nonsense', groupname_pattern='toolongnametoolongnametoolongnametoolongname'
+                                                                       'toolongnametoolongnametoolongnametoolongname'
+                                                                       'toolongnametoolongnametoolongnametoolongname'
+                                                                       'toolongnametoolongnametoolongnametoolongname'
+                                                                       'toolongnametoolongnametoolongnametoolongname'
+                                                                       'toolongnametoolongnametoolongnametoolongname'))
+      assert_true('Ensure this value has at most 256 characters' in response.context['form'].errors['groupname_pattern'][0], response)
 
-    # Test dn with spaces in username and dn (should fail)
-    response = c.post(URL, dict(server='nonsense', username_pattern='uid=user with space,ou=People,dc=example,dc=com', password1='test', password2='test', dn=True))
-    assert_true("Could not get LDAP details for users in pattern" in response.content, response.content)
-    response = c.get(reverse(desktop.views.log_view))
-    assert_true("{username}: Username must not contain whitespaces".format(username='user with space') in response.content, response.content)
+      # Test wild card
+      response = c.post(URL, dict(server='nonsense', groupname_pattern='*r*'))
+      assert_true('/useradmin/groups' in response['Location'], response)
+    finally:
+      for finish in reset:
+        finish()
 
-    # Test dn with spaces in dn, but not username (should succeed)
-    response = c.post(URL, dict(server='nonsense', username_pattern='uid=user without space,ou=People,dc=example,dc=com', password1='test', password2='test', dn=True))
-    assert_true(User.objects.filter(username='spaceless').exists())
+  def test_sync_ldap_users_groups(self):
+    URL = reverse(sync_ldap_users_groups)
 
-  finally:
-    for finish in done:
-      finish()
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
+    c = make_logged_in_client('test', is_superuser=True)
 
-def test_add_ldap_groups():
-  URL = reverse(add_ldap_groups)
+    reset = []
 
-  reset_all_users()
-  reset_all_groups()
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
 
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+    try:
+      assert_true(c.get(URL))
+      assert_true(c.post(URL))
+    finally:
+      for finish in reset:
+        finish()
 
-  c = make_logged_in_client(username='test', is_superuser=True)
+  def test_ldap_exception_handling(self):
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    class LdapTestConnectionError(LdapTestConnection):
+      def find_users(self, user, find_by_dn=False):
+        raise ldap.LDAPError('No such object')
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnectionError()
 
-  reset = []
+    c = make_logged_in_client('test', is_superuser=True)
 
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+    reset = []
 
-  try:
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
 
-    assert_true(c.get(URL))
+    try:
+      response = c.post(reverse(add_ldap_users), dict(server='nonsense', username_pattern='moe', password1='test', password2='test'), follow=True)
+      assert_true('There was an error when communicating with LDAP' in response.content, response)
+    finally:
+      for finish in reset:
+        finish()
 
-    response = c.post(URL, dict(server='nonsense', groupname_pattern='TestUsers'))
-    assert_true('Location' in response, response)
-    assert_true('/useradmin/groups' in response['Location'])
 
-    # Test with space
-    response = c.post(URL, dict(server='nonsense', groupname_pattern='Test Administrators'))
-    assert_true('Location' in response, response)
-    assert_true('/useradmin/groups' in response['Location'], response)
+class TestUserAdminLdapWithHadoop(BaseUserAdminTests):
+  requires_hadoop = True
 
-    response = c.post(URL, dict(server='nonsense', groupname_pattern='toolongnametoolongnametoolongnametoolongname'
-                                                                     'toolongnametoolongnametoolongnametoolongname'
-                                                                     'toolongnametoolongnametoolongnametoolongname'
-                                                                     'toolongnametoolongnametoolongnametoolongname'
-                                                                     'toolongnametoolongnametoolongnametoolongname'
-                                                                     'toolongnametoolongnametoolongnametoolongname'))
-    assert_true('Ensure this value has at most 256 characters' in response.context['form'].errors['groupname_pattern'][0], response)
+  def test_ensure_home_directory_add_ldap_users(self):
+    URL = reverse(add_ldap_users)
 
-    # Test wild card
-    response = c.post(URL, dict(server='nonsense', groupname_pattern='*r*'))
-    assert_true('/useradmin/groups' in response['Location'], response)
-  finally:
-    for finish in reset:
-      finish()
-
-def test_sync_ldap_users_groups():
-  URL = reverse(sync_ldap_users_groups)
-
-  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()
-
-  c = make_logged_in_client('test', is_superuser=True)
-
-  reset = []
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    assert_true(c.get(URL))
-    assert_true(c.post(URL))
-  finally:
-    for finish in reset:
-      finish()
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+
+    cluster = pseudo_hdfs4.shared_cluster()
+    c = make_logged_in_client(cluster.superuser, is_superuser=True)
+    cluster.fs.setuser(cluster.superuser)
+
+    reset = []
+
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
 
-def test_ldap_exception_handling():
-  reset_all_users()
-  reset_all_groups()
+    try:
+      assert_true(c.get(URL))
 
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  class LdapTestConnectionError(LdapTestConnection):
-    def find_users(self, user, find_by_dn=False):
-      raise ldap.LDAPError('No such object')
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnectionError()
+      response = c.post(URL, dict(server='nonsense', username_pattern='moe', password1='test', password2='test'))
+      assert_true('/useradmin/users' in response['Location'])
+      assert_false(cluster.fs.exists('/user/moe'))
 
-  c = make_logged_in_client('test', is_superuser=True)
+      # Try same thing with home directory creation.
+      response = c.post(URL, dict(server='nonsense', username_pattern='curly', password1='test', password2='test', ensure_home_directory=True))
+      assert_true('/useradmin/users' in response['Location'])
+      assert_true(cluster.fs.exists('/user/curly'))
 
-  reset = []
+      response = c.post(URL, dict(server='nonsense', username_pattern='bad_name', password1='test', password2='test'))
+      assert_true('Could not' in response.context['form'].errors['username_pattern'][0])
+      assert_false(cluster.fs.exists('/user/bad_name'))
 
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
+      # See if moe, who did not ask for his home directory, has a home directory.
+      assert_false(cluster.fs.exists('/user/moe'))
 
-  try:
-    response = c.post(reverse(add_ldap_users), dict(server='nonsense', username_pattern='moe', password1='test', password2='test'), follow=True)
-    assert_true('There was an error when communicating with LDAP' in response.content, response)
-  finally:
-    for finish in reset:
-      finish()
+      # Try wild card now
+      response = c.post(URL, dict(server='nonsense', username_pattern='*rr*', password1='test', password2='test', ensure_home_directory=True))
+      assert_true('/useradmin/users' in response['Location'])
+      assert_true(cluster.fs.exists('/user/curly'))
+      assert_true(cluster.fs.exists(u'/user/lårry'))
+      assert_false(cluster.fs.exists('/user/otherguy'))
+    finally:
+      # Clean up
+      for finish in reset:
+        finish()
 
-@attr('requires_hadoop')
-def test_ensure_home_directory_add_ldap_users():
-  URL = reverse(add_ldap_users)
+      if cluster.fs.exists('/user/curly'):
+        cluster.fs.rmtree('/user/curly')
+      if cluster.fs.exists(u'/user/lårry'):
+        cluster.fs.rmtree(u'/user/lårry')
+      if cluster.fs.exists('/user/otherguy'):
+        cluster.fs.rmtree('/user/otherguy')
+
+  def test_ensure_home_directory_sync_ldap_users_groups(self):
+    URL = reverse(sync_ldap_users_groups)
+
+    # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
+    ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
 
-  reset_all_users()
-  reset_all_groups()
+    cluster = pseudo_hdfs4.shared_cluster()
+    c = make_logged_in_client(cluster.superuser, is_superuser=True)
+    cluster.fs.setuser(cluster.superuser)
 
-  # Set up LDAP tests to use a LdapTestConnection instead of an actual LDAP connection
-  ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
+    reset = []
 
-  cluster = pseudo_hdfs4.shared_cluster()
-  c = make_logged_in_client(cluster.superuser, is_superuser=True)
-  cluster.fs.setuser(cluster.superuser)
+    # Set to nonsensical value just to force new config usage.
+    # Should continue to use cached connection.
+    reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
 
-  reset = []
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    assert_true(c.get(URL))
-
-    response = c.post(URL, dict(server='nonsense', username_pattern='moe', password1='test', password2='test'))
-    assert_true('/useradmin/users' in response['Location'])
-    assert_false(cluster.fs.exists('/user/moe'))
-
-    # Try same thing with home directory creation.
-    response = c.post(URL, dict(server='nonsense', username_pattern='curly', password1='test', password2='test', ensure_home_directory=True))
-    assert_true('/useradmin/users' in response['Location'])
-    assert_true(cluster.fs.exists('/user/curly'))
-
-    response = c.post(URL, dict(server='nonsense', username_pattern='bad_name', password1='test', password2='test'))
-    assert_true('Could not' in response.context['form'].errors['username_pattern'][0])
-    assert_false(cluster.fs.exists('/user/bad_name'))
-
-    # See if moe, who did not ask for his home directory, has a home directory.
-    assert_false(cluster.fs.exists('/user/moe'))
-
-    # Try wild card now
-    response = c.post(URL, dict(server='nonsense', username_pattern='*rr*', password1='test', password2='test', ensure_home_directory=True))
-    assert_true('/useradmin/users' in response['Location'])
-    assert_true(cluster.fs.exists('/user/curly'))
-    assert_true(cluster.fs.exists(u'/user/lårry'))
-    assert_false(cluster.fs.exists('/user/otherguy'))
-  finally:
-    # Clean up
-    for finish in reset:
-      finish()
+    try:
+      c.post(reverse(add_ldap_users), dict(server='nonsense', username_pattern='curly', password1='test', password2='test'))
+      assert_false(cluster.fs.exists('/user/curly'))
+      assert_true(c.post(URL, dict(server='nonsense', ensure_home_directory=True)))
+      assert_true(cluster.fs.exists('/user/curly'))
+    finally:
+      for finish in reset:
+        finish()
 
-    if cluster.fs.exists('/user/curly'):
-      cluster.fs.rmtree('/user/curly')
-    if cluster.fs.exists(u'/user/lårry'):
-      cluster.fs.rmtree(u'/user/lårry')
-    if cluster.fs.exists('/user/otherguy'):
-      cluster.fs.rmtree('/user/otherguy')
-
-@attr('requires_hadoop')
-def test_ensure_home_directory_sync_ldap_users_groups():
-  URL = reverse(sync_ldap_users_groups)
-
-  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()
-
-  cluster = pseudo_hdfs4.shared_cluster()
-  c = make_logged_in_client(cluster.superuser, is_superuser=True)
-  cluster.fs.setuser(cluster.superuser)
-
-  reset = []
-
-  # Set to nonsensical value just to force new config usage.
-  # Should continue to use cached connection.
-  reset.append(desktop.conf.LDAP.LDAP_SERVERS.set_for_testing(get_nonsense_config()))
-
-  try:
-    c.post(reverse(add_ldap_users), dict(server='nonsense', username_pattern='curly', password1='test', password2='test'))
-    assert_false(cluster.fs.exists('/user/curly'))
-    assert_true(c.post(URL, dict(server='nonsense', ensure_home_directory=True)))
-    assert_true(cluster.fs.exists('/user/curly'))
-  finally:
-    for finish in reset:
-      finish()
-
-    if cluster.fs.exists('/user/curly'):
-      cluster.fs.rmtree('/user/curly')
+      if cluster.fs.exists('/user/curly'):
+        cluster.fs.rmtree('/user/curly')

Những thai đổi đã bị hủy bỏ vì nó quá lớn
+ 544 - 535
apps/useradmin/src/useradmin/test_ldap_deprecated.py


+ 544 - 508
apps/useradmin/src/useradmin/tests.py

@@ -219,515 +219,551 @@ def test_invalid_username():
     response = c.post('/useradmin/users/new', dict(username=bad_name, password1="test", password2="test"))
     assert_true('not allowed' in response.context["form"].errors['username'][0])
 
-def test_group_permissions():
-  reset_all_users()
-  reset_all_groups()
 
-  # Get ourselves set up with a user and a group
-  c = make_logged_in_client(username="test", is_superuser=True)
-  Group.objects.create(name="test-group")
-  test_user = User.objects.get(username="test")
-  test_user.groups.add(Group.objects.get(name="test-group"))
-  test_user.save()
-
-  # Make sure that a superuser can always access applications
-  response = c.get('/useradmin/users')
-  assert_true('Hue Users' in response.content)
-
-  assert_true(len(GroupPermission.objects.all()) == 0)
-  c.post('/useradmin/groups/edit/test-group',
-         dict(name="test-group",
-         members=[User.objects.get(username="test").pk],
-         permissions=[HuePermission.objects.get(app='useradmin',action='access').pk],
-         save="Save"), follow=True)
-  assert_true(len(GroupPermission.objects.all()) == 1)
-
-  # Now test that we have limited access
-  c1 = make_logged_in_client(username="nonadmin", is_superuser=False)
-  response = c1.get('/useradmin/users')
-  assert_true('You do not have permission to access the Useradmin application.' in response.content)
-
-  # Add the non-admin to a group that should grant permissions to the app
-  test_user = User.objects.get(username="nonadmin")
-  test_user.groups.add(Group.objects.get(name='test-group'))
-  test_user.save()
-
-  # Check that we have access now
-  response = c1.get('/useradmin/users')
-  assert_true(get_profile(test_user).has_hue_permission('access','useradmin'))
-  assert_true('Hue Users' in response.content)
-
-  # Make sure we can't modify permissions
-  response = c1.get('/useradmin/permissions/edit/useradmin/access')
-  assert_true('must be a superuser to change permissions' in response.content)
-
-  # And revoke access from the group
-  c.post('/useradmin/permissions/edit/useradmin/access',
-         dict(app='useradmin',
-         priv='access',
-         groups=[],
-         save="Save"), follow=True)
-  assert_true(len(GroupPermission.objects.all()) == 0)
-  assert_false(get_profile(test_user).has_hue_permission('access','useradmin'))
-
-  # We should no longer have access to the app
-  response = c1.get('/useradmin/users')
-  assert_true('You do not have permission to access the Useradmin application.' in response.content)
-
-def test_default_group():
-  reset_all_users()
-  reset_all_groups()
-  useradmin.conf.DEFAULT_USER_GROUP.set_for_testing('test_default')
-  get_default_user_group()
-
-  c = make_logged_in_client(username='test', is_superuser=True)
-
-  # Create default group if it doesn't already exist.
-  assert_true(Group.objects.filter(name='test_default').exists())
-
-  # Try deleting the default group
-  assert_true(Group.objects.filter(name='test_default').exists())
-  response = c.post('/useradmin/groups/delete', {'group_names': ['test_default']})
-  assert_true('default user group may not be deleted' in response.content)
-  assert_true(Group.objects.filter(name='test_default').exists())
-
-  # Change the name of the default group, and try deleting again
-  useradmin.conf.DEFAULT_USER_GROUP.set_for_testing('new_default')
-  response = c.post('/useradmin/groups/delete' , {'group_names': ['test_default']})
-  assert_false(Group.objects.filter(name='test_default').exists())
-  assert_true(Group.objects.filter(name='new_default').exists())
-
-def test_get_profile():
-  # Ensure profiles are created after get_profile is called.
-  reset_all_users()
-  reset_all_groups()
-  c = make_logged_in_client(username='test', password='test', is_superuser=True)
-  assert_equal(0, UserProfile.objects.count())
-  p = get_profile(User.objects.get(username='test'))
-  assert_equal(1, UserProfile.objects.count())
-
-def test_group_admin():
-  reset_all_users()
-  reset_all_groups()
-
-  c = make_logged_in_client(username="test", is_superuser=True)
-  response = c.get('/useradmin/groups')
-  # No groups just yet
-  assert_true(len(response.context["groups"]) == 0)
-  assert_true("Hue Groups" in response.content)
-
-  # Create a group
-  response = c.get('/useradmin/groups/new')
-  assert_equal('/useradmin/groups/new', response.context['action'])
-  c.post('/useradmin/groups/new', dict(name="testgroup"))
-
-  # We should have an empty group in the DB now
-  assert_true(len(Group.objects.all()) == 1)
-  assert_true(Group.objects.filter(name="testgroup").exists())
-  assert_true(len(Group.objects.get(name="testgroup").user_set.all()) == 0)
-
-  # And now, just for kicks, let's try adding a user
-  response = c.post('/useradmin/groups/edit/testgroup',
-                    dict(name="testgroup",
-                    members=[User.objects.get(username="test").pk],
-                    save="Save"), follow=True)
-  assert_true(len(Group.objects.get(name="testgroup").user_set.all()) == 1)
-  assert_true(Group.objects.get(name="testgroup").user_set.filter(username="test").exists())
-
-  # Test some permissions
-  c2 = make_logged_in_client(username="nonadmin", is_superuser=False)
-
-  # Need to give access to the user for the rest of the test
-  group = Group.objects.create(name="access-group")
-  perm = HuePermission.objects.get(app='useradmin', action='access')
-  GroupPermission.objects.create(group=group, hue_permission=perm)
-  test_user = User.objects.get(username="nonadmin")
-  test_user.groups.add(Group.objects.get(name="access-group"))
-  test_user.save()
-
-  # Make sure non-superusers can't do bad things
-  response = c2.get('/useradmin/groups/new')
-  assert_true("You must be a superuser" in response.content)
-  response = c2.get('/useradmin/groups/edit/testgroup')
-  assert_true("You must be a superuser" in response.content)
-
-  response = c2.post('/useradmin/groups/new', dict(name="nonsuperuser"))
-  assert_true("You must be a superuser" in response.content)
-  response = c2.post('/useradmin/groups/edit/testgroup',
-                    dict(name="nonsuperuser",
-                    members=[User.objects.get(username="test").pk],
-                    save="Save"), follow=True)
-  assert_true("You must be a superuser" in response.content)
-
-  # Should be one group left, because we created the other group
-  response = c.post('/useradmin/groups/delete', {'group_names': ['testgroup']})
-  assert_true(len(Group.objects.all()) == 1)
-
-  group_count = len(Group.objects.all())
-  response = c.post('/useradmin/groups/new', dict(name="with space"))
-  assert_equal(len(Group.objects.all()), group_count + 1)
-
-def test_user_admin_password_policy():
-  reset_all_users()
-  reset_all_groups()
-
-  # Set up password policy
-  password_hint = password_error_msg = ("The password must be at least 8 characters long, "
-                                        "and must contain both uppercase and lowercase letters, "
-                                        "at least one number, and at least one special character.")
-  password_rule = "^(?=.*?[A-Z])(?=(.*[a-z]){1,})(?=(.*[\d]){1,})(?=(.*[\W_]){1,}).{8,}$"
-
-  useradmin.conf.PASSWORD_POLICY.IS_ENABLED.set_for_testing(True)
-  useradmin.conf.PASSWORD_POLICY.PWD_RULE.set_for_testing(password_rule)
-  useradmin.conf.PASSWORD_POLICY.PWD_HINT.set_for_testing(password_hint)
-  useradmin.conf.PASSWORD_POLICY.PWD_ERROR_MESSAGE.set_for_testing(password_error_msg)
-  reset_password_policy()
-
-  # Test first-ever login with password policy enabled
-  c = Client()
-
-  response = c.get('/accounts/login/')
-  assert_equal(200, response.status_code)
-  assert_true(response.context['first_login_ever'])
-
-  response = c.post('/accounts/login/', dict(username="test_first_login", password="foo"))
-  assert_true(response.context['first_login_ever'])
-  assert_equal([password_error_msg], response.context["form"]["password"].errors)
-
-  response = c.post('/accounts/login/', dict(username="test_first_login", password="foobarTest1["), follow=True)
-  assert_equal(200, response.status_code)
-  assert_true(User.objects.get(username="test_first_login").is_superuser)
-  assert_true(User.objects.get(username="test_first_login").check_password("foobarTest1["))
-
-  c.get('/accounts/logout')
-
-  # Test changing a user's password
-  c = make_logged_in_client('superuser', is_superuser=True)
-
-  # Test password hint is displayed
-  response = c.get('/useradmin/users/edit/superuser')
-  assert_true(password_hint in response.content)
-
-  # Password is less than 8 characters
-  response = c.post('/useradmin/users/edit/superuser',
-                    dict(username="superuser",
-                         is_superuser=True,
-                         password1="foo",
-                         password2="foo"))
-  assert_equal([password_error_msg], response.context["form"]["password1"].errors)
-
-  # Password is more than 8 characters long but does not have a special character
-  response = c.post('/useradmin/users/edit/superuser',
-                    dict(username="superuser",
-                         is_superuser=True,
-                         password1="foobarTest1",
-                         password2="foobarTest1"))
-  assert_equal([password_error_msg], response.context["form"]["password1"].errors)
-
-  # Password1 and Password2 are valid but they do not match
-  response = c.post('/useradmin/users/edit/superuser',
-                    dict(username="superuser",
-                         is_superuser=True,
-                         password1="foobarTest1??",
-                         password2="foobarTest1?",
-                         password_old="foobarTest1[",
-                         is_active=True))
-  assert_equal(["Passwords do not match."], response.context["form"]["password2"].errors)
-
-  # Password is valid now
-  c.post('/useradmin/users/edit/superuser',
-         dict(username="superuser",
-              is_superuser=True,
-              password1="foobarTest1[",
-              password2="foobarTest1[",
-              password_old="test",
-              is_active=True))
-  assert_true(User.objects.get(username="superuser").is_superuser)
-  assert_true(User.objects.get(username="superuser").check_password("foobarTest1["))
-
-  # Test creating a new user
-  response = c.get('/useradmin/users/new')
-  assert_true(password_hint in response.content)
-
-  # Password is more than 8 characters long but does not have a special character
-  response = c.post('/useradmin/users/new',
-                    dict(username="test_user",
-                         is_superuser=False,
-                         password1="foo",
-                         password2="foo"))
-  assert_equal({'password1': [password_error_msg], 'password2': [password_error_msg]},
-               response.context["form"].errors)
-
-  # Password is more than 8 characters long but does not have a special character
-  response = c.post('/useradmin/users/new',
-                    dict(username="test_user",
-                         is_superuser=False,
-                         password1="foobarTest1",
-                         password2="foobarTest1"))
-
-  assert_equal({'password1': [password_error_msg], 'password2': [password_error_msg]},
-               response.context["form"].errors)
-
-  # Password1 and Password2 are valid but they do not match
-  response = c.post('/useradmin/users/new',
-                    dict(username="test_user",
-                         is_superuser=False,
-                         password1="foobarTest1[",
-                         password2="foobarTest1?"))
-  assert_equal({'password2': ["Passwords do not match."]}, response.context["form"].errors)
-
-  # Password is valid now
-  c.post('/useradmin/users/new',
-         dict(username="test_user",
-              is_superuser=False,
-              password1="foobarTest1[",
-              password2="foobarTest1[", is_active=True))
-  assert_false(User.objects.get(username="test_user").is_superuser)
-  assert_true(User.objects.get(username="test_user").check_password("foobarTest1["))
-
-
-def test_user_admin():
-  FUNNY_NAME = '~`!@#$%^&*()_-+={}[]|\;"<>?/,.'
-  FUNNY_NAME_QUOTED = urllib.quote(FUNNY_NAME)
-
-  reset_all_users()
-  reset_all_groups()
-  useradmin.conf.DEFAULT_USER_GROUP.set_for_testing('test_default')
-
-  useradmin.conf.PASSWORD_POLICY.IS_ENABLED.set_for_testing(False)
-  reset_password_policy()
-
-  c = make_logged_in_client('test', is_superuser=True)
-  user = User.objects.get(username='test')
-
-  # Test basic output.
-  response = c.get('/useradmin/')
-  assert_true(len(response.context["users"]) > 0)
-  assert_true("Hue Users" in response.content)
-
-  # Test editing a superuser
-  # Just check that this comes back
-  response = c.get('/useradmin/users/edit/test')
-  # Edit it, to add a first and last name
-  response = c.post('/useradmin/users/edit/test',
-                    dict(username="test",
-                         first_name=u"Inglés",
-                         last_name=u"Español",
-                         is_superuser="True",
-                         is_active="True"),
-                    follow=True)
-  assert_true("User information updated" in response.content,
-              "Notification should be displayed in: %s" % response.content)
-  # Edit it, can't change username
-  response = c.post('/useradmin/users/edit/test',
-                    dict(username="test2",
-                         first_name=u"Inglés",
-                         last_name=u"Español",
-                         is_superuser="True",
-                         is_active="True"),
-                    follow=True)
-  assert_true("You cannot change a username" in response.content)
-  # Now make sure that those were materialized
-  response = c.get('/useradmin/users/edit/test')
-  assert_equal(smart_unicode("Inglés"), response.context["form"].instance.first_name)
-  assert_true("Español" in response.content)
-  # Shouldn't be able to demote to non-superuser
-  response = c.post('/useradmin/users/edit/test', dict(username="test",
-                        first_name=u"Inglés", last_name=u"Español",
-                        is_superuser=False, is_active=True))
-  assert_true("You cannot remove" in response.content,
-              "Shouldn't be able to remove the last superuser")
-  # Shouldn't be able to delete oneself
-  response = c.post('/useradmin/users/delete', {u'user_ids': [user.id]})
-  assert_true("You cannot remove yourself" in response.content,
-              "Shouldn't be able to delete the last superuser")
-
-  # Let's try changing the password
-  response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", is_superuser=True, password1="foo", password2="foobar"))
-  assert_equal(["Passwords do not match."], response.context["form"]["password2"].errors, "Should have complained about mismatched password")
-  # Old password not confirmed
-  response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", password1="foo", password2="foo", is_active=True, is_superuser=True))
-  assert_equal(["The old password does not match the current password."], response.context["form"]["password_old"].errors, "Should have complained about old password")
-  # Good now
-  response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", password1="foo", password2="foo", password_old="test", is_active=True, is_superuser=True))
-  assert_true(User.objects.get(username="test").is_superuser)
-  assert_true(User.objects.get(username="test").check_password("foo"))
-  # Change it back!
-  response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", password1="test", password2="test", password_old="foo", is_active="True", is_superuser="True"))
-  assert_true(User.objects.get(username="test").check_password("test"))
-  assert_true(make_logged_in_client(username = "test", password = "test"), "Check that we can still login.")
-
-  # Check new user form for default group
-  group = get_default_user_group()
-  response = c.get('/useradmin/users/new')
-  assert_true(response)
-  assert_true(('<option value="%s" selected="selected">%s</option>' % (group.id, group.name)) in str(response))
-
-  # Create a new regular user (duplicate name)
-  response = c.post('/useradmin/users/new', dict(username="test", password1="test", password2="test"))
-  assert_equal({ 'username': ["User with this Username already exists."]}, response.context["form"].errors)
-
-  # Create a new regular user (for real)
-  response = c.post('/useradmin/users/new', dict(username=FUNNY_NAME,
-                                           password1="test",
-                                           password2="test",
-                                           is_active="True"))
-  response = c.get('/useradmin/')
-  assert_true(FUNNY_NAME_QUOTED in response.content)
-  assert_true(len(response.context["users"]) > 1)
-  assert_true("Hue Users" in response.content)
-  # Validate profile is created.
-  assert_true(UserProfile.objects.filter(user__username=FUNNY_NAME).exists())
-
-  # Need to give access to the user for the rest of the test
-  group = Group.objects.create(name="test-group")
-  perm = HuePermission.objects.get(app='useradmin', action='access')
-  GroupPermission.objects.create(group=group, hue_permission=perm)
-
-  # Verify that we can modify user groups through the user admin pages
-  response = c.post('/useradmin/users/new', dict(username="group_member", password1="test", password2="test", groups=[group.pk]))
-  User.objects.get(username='group_member')
-  assert_true(User.objects.get(username='group_member').groups.filter(name='test-group').exists())
-  response = c.post('/useradmin/users/edit/group_member', dict(username="group_member", groups=[]))
-  assert_false(User.objects.get(username='group_member').groups.filter(name='test-group').exists())
-
-  # Check permissions by logging in as the new user
-  c_reg = make_logged_in_client(username=FUNNY_NAME, password="test")
-  test_user = User.objects.get(username=FUNNY_NAME)
-  test_user.groups.add(Group.objects.get(name="test-group"))
-  test_user.save()
-
-  # Regular user should be able to modify oneself
-  response = c_reg.post('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,),
-                        dict(username = FUNNY_NAME,
-                             first_name = "Hello",
-                             is_active = True,
-                             groups=[group.id for group in test_user.groups.all()]), follow=True)
-  assert_equal(response.status_code, 200)
-  response = c_reg.get('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,), follow=True)
-  assert_equal(response.status_code, 200)
-  assert_equal("Hello", response.context["form"].instance.first_name)
-  funny_user = User.objects.get(username=FUNNY_NAME)
-  # Can't edit other people.
-  response = c_reg.post("/useradmin/users/delete", {u'user_ids': [funny_user.id]})
-  assert_true("You must be a superuser" in response.content,
-              "Regular user can't edit other people")
-
-  # Revert to regular "test" user, that has superuser powers.
-  c_su = make_logged_in_client()
-  # Inactivate FUNNY_NAME
-  c_su.post('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,),
-                        dict(username = FUNNY_NAME,
-                             first_name = "Hello",
-                             is_active = False))
-  # Now make sure FUNNY_NAME can't log back in
-  response = c_reg.get('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,))
-  assert_true(response.status_code == 302 and "login" in response["location"],
-              "Inactivated user gets redirected to login page")
-
-  # Delete that regular user
-  funny_profile = get_profile(test_user)
-  response = c_su.post('/useradmin/users/delete', {u'user_ids': [funny_user.id]})
-  assert_equal(302, response.status_code)
-  assert_false(User.objects.filter(username=FUNNY_NAME).exists())
-  assert_false(UserProfile.objects.filter(id=funny_profile.id).exists())
-
-  # Bulk delete users
-  u1 = User.objects.create(username='u1', password="u1")
-  u2 = User.objects.create(username='u2', password="u2")
-  assert_equal(User.objects.filter(username__in=['u1', 'u2']).count(), 2)
-  response = c_su.post('/useradmin/users/delete', {u'user_ids': [u1.id, u2.id]})
-  assert_equal(User.objects.filter(username__in=['u1', 'u2']).count(), 0)
-
-  # Make sure that user deletion works if the user has never performed a request.
-  funny_user = User.objects.create(username=FUNNY_NAME, password='test')
-  assert_true(User.objects.filter(username=FUNNY_NAME).exists())
-  assert_false(UserProfile.objects.filter(user__username=FUNNY_NAME).exists())
-  response = c_su.post('/useradmin/users/delete', {u'user_ids': [funny_user.id]})
-  assert_equal(302, response.status_code)
-  assert_false(User.objects.filter(username=FUNNY_NAME).exists())
-  assert_false(UserProfile.objects.filter(user__username=FUNNY_NAME).exists())
-
-  # You shouldn't be able to create a user without a password
-  response = c_su.post('/useradmin/users/new', dict(username="test"))
-  assert_true("You must specify a password when creating a new user." in response.content)
-
-@attr('requires_hadoop')
-def test_ensure_home_directory():
-  raise SkipTest
-
-  reset_all_users()
-  reset_all_groups()
-
-  useradmin.conf.PASSWORD_POLICY.IS_ENABLED.set_for_testing(False)
-  reset_password_policy()
-
-  # Cluster and client for home directory creation
-  cluster = pseudo_hdfs4.shared_cluster()
-  c = make_logged_in_client(cluster.superuser, is_superuser=True, groupname='test1')
-  cluster.fs.setuser(cluster.superuser)
-
-  # Create a user with a home directory
-  assert_false(cluster.fs.exists('/user/test1'))
-  response = c.post('/useradmin/users/new', dict(username="test1", password1='test', password2='test', ensure_home_directory=True))
-  assert_true(cluster.fs.exists('/user/test1'))
-  dir_stat = cluster.fs.stats('/user/test1')
-  assert_equal('test1', dir_stat.user)
-  assert_equal('test1', dir_stat.group)
-  assert_equal('40755', '%o' % dir_stat.mode)
-
-  # Create a user, then add their home directory
-  assert_false(cluster.fs.exists('/user/test2'))
-  response = c.post('/useradmin/users/new', dict(username="test2", password1='test', password2='test'))
-  assert_false(cluster.fs.exists('/user/test2'))
-  response = c.post('/useradmin/users/edit/%s' % "test2", dict(username="test2", password1='test', password2='test', password_old="test", ensure_home_directory=True))
-  assert_true(cluster.fs.exists('/user/test2'))
-  dir_stat = cluster.fs.stats('/user/test2')
-  assert_equal('test2', dir_stat.user)
-  assert_equal('test2', dir_stat.group)
-  assert_equal('40755', '%o' % dir_stat.mode)
-
-def test_list_for_autocomplete():
-  reset_all_users()
-  reset_all_groups()
-
-  # Now the autocomplete has access to all the users and groups
-  c1 = make_logged_in_client('test_list_for_autocomplete', is_superuser=False, groupname='test_list_for_autocomplete')
-  c2_same_group = make_logged_in_client('test_list_for_autocomplete2', is_superuser=False, groupname='test_list_for_autocomplete')
-  c3_other_group = make_logged_in_client('test_list_for_autocomplete3', is_superuser=False, groupname='test_list_for_autocomplete_other_group')
-
-  # c1 is in the same group as c2
-  response = c1.get(reverse('useradmin.views.list_for_autocomplete'), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
-  content = json.loads(response.content)
-
-  users = [smart_unicode(user['username']) for user in content['users']]
-  groups = [smart_unicode(user['name']) for user in content['groups']]
-
-  assert_equal([u'test_list_for_autocomplete2', u'test_list_for_autocomplete3'], users)
-  assert_true(u'test_list_for_autocomplete' in groups, groups)
-  assert_true(u'test_list_for_autocomplete_other_group' in groups, groups)
-
-  # c2 is in the same group as c1
-  response = c2_same_group.get(reverse('useradmin.views.list_for_autocomplete'), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
-  content = json.loads(response.content)
-
-  users = [smart_unicode(user['username']) for user in content['users']]
-  groups = [smart_unicode(user['name']) for user in content['groups']]
-
-  assert_equal([u'test_list_for_autocomplete', u'test_list_for_autocomplete3'], users)
-  assert_true(u'test_list_for_autocomplete' in groups, groups)
-  assert_true(u'test_list_for_autocomplete_other_group' in groups, groups)
-
-  # c3 is alone except for groups
-  response = c3_other_group.get(reverse('useradmin.views.list_for_autocomplete'), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
-  content = json.loads(response.content)
-
-  users = [smart_unicode(user['username']) for user in content['users']]
-  groups = [smart_unicode(user['name']) for user in content['groups']]
-
-  assert_equal([u'test_list_for_autocomplete', u'test_list_for_autocomplete2'], users)
-  assert_true(u'test_list_for_autocomplete' in groups, groups)
-  assert_true(u'test_list_for_autocomplete_other_group' in groups, groups)
+class BaseUserAdminTests(object):
+  @classmethod
+  def setUpClass(cls):
+    cls._class_resets = [
+        useradmin.conf.DEFAULT_USER_GROUP.set_for_testing(None),
+    ]
+
+  @classmethod
+  def tearDownClass(cls):
+    for reset in cls._class_resets:
+      reset()
+
+  def setUp(self):
+    reset_all_users()
+    reset_all_groups()
+
+  def tearDown(self):
+    pass
+
+
+class TestUserAdmin(BaseUserAdminTests):
+  def test_group_permissions(self):
+    # Get ourselves set up with a user and a group
+    c = make_logged_in_client(username="test", is_superuser=True)
+    Group.objects.create(name="test-group")
+    test_user = User.objects.get(username="test")
+    test_user.groups.add(Group.objects.get(name="test-group"))
+    test_user.save()
+
+    # Make sure that a superuser can always access applications
+    response = c.get('/useradmin/users')
+    assert_true('Hue Users' in response.content)
+
+    assert_true(len(GroupPermission.objects.all()) == 0)
+    c.post('/useradmin/groups/edit/test-group',
+           dict(name="test-group",
+           members=[User.objects.get(username="test").pk],
+           permissions=[HuePermission.objects.get(app='useradmin',action='access').pk],
+           save="Save"), follow=True)
+    assert_true(len(GroupPermission.objects.all()) == 1)
+
+    # Now test that we have limited access
+    c1 = make_logged_in_client(username="nonadmin", is_superuser=False)
+    response = c1.get('/useradmin/users')
+    assert_true('You do not have permission to access the Useradmin application.' in response.content)
+
+    # Add the non-admin to a group that should grant permissions to the app
+    test_user = User.objects.get(username="nonadmin")
+    test_user.groups.add(Group.objects.get(name='test-group'))
+    test_user.save()
+
+    # Check that we have access now
+    response = c1.get('/useradmin/users')
+    assert_true(get_profile(test_user).has_hue_permission('access','useradmin'))
+    assert_true('Hue Users' in response.content)
+
+    # Make sure we can't modify permissions
+    response = c1.get('/useradmin/permissions/edit/useradmin/access')
+    assert_true('must be a superuser to change permissions' in response.content)
+
+    # And revoke access from the group
+    c.post('/useradmin/permissions/edit/useradmin/access',
+           dict(app='useradmin',
+           priv='access',
+           groups=[],
+           save="Save"), follow=True)
+    assert_true(len(GroupPermission.objects.all()) == 0)
+    assert_false(get_profile(test_user).has_hue_permission('access','useradmin'))
+
+    # We should no longer have access to the app
+    response = c1.get('/useradmin/users')
+    assert_true('You do not have permission to access the Useradmin application.' in response.content)
+
+  def test_default_group(self):
+    resets = [
+      useradmin.conf.DEFAULT_USER_GROUP.set_for_testing('test_default')
+    ]
+
+    try:
+      get_default_user_group()
+
+      c = make_logged_in_client(username='test', is_superuser=True)
+
+      # Create default group if it doesn't already exist.
+      assert_true(Group.objects.filter(name='test_default').exists())
+
+      # Try deleting the default group
+      assert_true(Group.objects.filter(name='test_default').exists())
+      response = c.post('/useradmin/groups/delete', {'group_names': ['test_default']})
+      assert_true('default user group may not be deleted' in response.content)
+      assert_true(Group.objects.filter(name='test_default').exists())
+
+      # Change the name of the default group, and try deleting again
+      resets.append(useradmin.conf.DEFAULT_USER_GROUP.set_for_testing('new_default'))
+
+      response = c.post('/useradmin/groups/delete' , {'group_names': ['test_default']})
+      assert_false(Group.objects.filter(name='test_default').exists())
+      assert_true(Group.objects.filter(name='new_default').exists())
+    finally:
+      for reset in resets:
+        reset()
+
+
+  def test_get_profile(self):
+    # Ensure profiles are created after get_profile is called.
+    c = make_logged_in_client(username='test', password='test', is_superuser=True)
+    assert_equal(0, UserProfile.objects.count())
+    p = get_profile(User.objects.get(username='test'))
+    assert_equal(1, UserProfile.objects.count())
+
+
+  def test_group_admin(self):
+    c = make_logged_in_client(username="test", is_superuser=True)
+    response = c.get('/useradmin/groups')
+    # No groups just yet
+    assert_true(len(response.context["groups"]) == 0)
+    assert_true("Hue Groups" in response.content)
+
+    # Create a group
+    response = c.get('/useradmin/groups/new')
+    assert_equal('/useradmin/groups/new', response.context['action'])
+    c.post('/useradmin/groups/new', dict(name="testgroup"))
+
+    # We should have an empty group in the DB now
+    assert_true(len(Group.objects.all()) == 1)
+    assert_true(Group.objects.filter(name="testgroup").exists())
+    assert_true(len(Group.objects.get(name="testgroup").user_set.all()) == 0)
+
+    # And now, just for kicks, let's try adding a user
+    response = c.post('/useradmin/groups/edit/testgroup',
+                      dict(name="testgroup",
+                      members=[User.objects.get(username="test").pk],
+                      save="Save"), follow=True)
+    assert_true(len(Group.objects.get(name="testgroup").user_set.all()) == 1)
+    assert_true(Group.objects.get(name="testgroup").user_set.filter(username="test").exists())
+
+    # Test some permissions
+    c2 = make_logged_in_client(username="nonadmin", is_superuser=False)
+
+    # Need to give access to the user for the rest of the test
+    group = Group.objects.create(name="access-group")
+    perm = HuePermission.objects.get(app='useradmin', action='access')
+    GroupPermission.objects.create(group=group, hue_permission=perm)
+    test_user = User.objects.get(username="nonadmin")
+    test_user.groups.add(Group.objects.get(name="access-group"))
+    test_user.save()
+
+    # Make sure non-superusers can't do bad things
+    response = c2.get('/useradmin/groups/new')
+    assert_true("You must be a superuser" in response.content)
+    response = c2.get('/useradmin/groups/edit/testgroup')
+    assert_true("You must be a superuser" in response.content)
+
+    response = c2.post('/useradmin/groups/new', dict(name="nonsuperuser"))
+    assert_true("You must be a superuser" in response.content)
+    response = c2.post('/useradmin/groups/edit/testgroup',
+                      dict(name="nonsuperuser",
+                      members=[User.objects.get(username="test").pk],
+                      save="Save"), follow=True)
+    assert_true("You must be a superuser" in response.content)
+
+    # Should be one group left, because we created the other group
+    response = c.post('/useradmin/groups/delete', {'group_names': ['testgroup']})
+    assert_true(len(Group.objects.all()) == 1)
+
+    group_count = len(Group.objects.all())
+    response = c.post('/useradmin/groups/new', dict(name="with space"))
+    assert_equal(len(Group.objects.all()), group_count + 1)
+
+  def test_user_admin_password_policy(self):
+    # Set up password policy
+    password_hint = password_error_msg = ("The password must be at least 8 characters long, "
+                                          "and must contain both uppercase and lowercase letters, "
+                                          "at least one number, and at least one special character.")
+    password_rule = "^(?=.*?[A-Z])(?=(.*[a-z]){1,})(?=(.*[\d]){1,})(?=(.*[\W_]){1,}).{8,}$"
+
+    resets = [
+      useradmin.conf.PASSWORD_POLICY.IS_ENABLED.set_for_testing(True),
+      useradmin.conf.PASSWORD_POLICY.PWD_RULE.set_for_testing(password_rule),
+      useradmin.conf.PASSWORD_POLICY.PWD_HINT.set_for_testing(password_hint),
+      useradmin.conf.PASSWORD_POLICY.PWD_ERROR_MESSAGE.set_for_testing(password_error_msg),
+    ]
+
+    try:
+      reset_password_policy()
+
+      # Test first-ever login with password policy enabled
+      c = Client()
+
+      response = c.get('/accounts/login/')
+      assert_equal(200, response.status_code)
+      assert_true(response.context['first_login_ever'])
+
+      response = c.post('/accounts/login/', dict(username="test_first_login", password="foo"))
+      assert_true(response.context['first_login_ever'])
+      assert_equal([password_error_msg], response.context["form"]["password"].errors)
+
+      response = c.post('/accounts/login/', dict(username="test_first_login", password="foobarTest1["), follow=True)
+      assert_equal(200, response.status_code)
+      assert_true(User.objects.get(username="test_first_login").is_superuser)
+      assert_true(User.objects.get(username="test_first_login").check_password("foobarTest1["))
+
+      c.get('/accounts/logout')
+
+      # Test changing a user's password
+      c = make_logged_in_client('superuser', is_superuser=True)
+
+      # Test password hint is displayed
+      response = c.get('/useradmin/users/edit/superuser')
+      assert_true(password_hint in response.content)
+
+      # Password is less than 8 characters
+      response = c.post('/useradmin/users/edit/superuser',
+                        dict(username="superuser",
+                             is_superuser=True,
+                             password1="foo",
+                             password2="foo"))
+      assert_equal([password_error_msg], response.context["form"]["password1"].errors)
+
+      # Password is more than 8 characters long but does not have a special character
+      response = c.post('/useradmin/users/edit/superuser',
+                        dict(username="superuser",
+                             is_superuser=True,
+                             password1="foobarTest1",
+                             password2="foobarTest1"))
+      assert_equal([password_error_msg], response.context["form"]["password1"].errors)
+
+      # Password1 and Password2 are valid but they do not match
+      response = c.post('/useradmin/users/edit/superuser',
+                        dict(username="superuser",
+                             is_superuser=True,
+                             password1="foobarTest1??",
+                             password2="foobarTest1?",
+                             password_old="foobarTest1[",
+                             is_active=True))
+      assert_equal(["Passwords do not match."], response.context["form"]["password2"].errors)
+
+      # Password is valid now
+      c.post('/useradmin/users/edit/superuser',
+             dict(username="superuser",
+                  is_superuser=True,
+                  password1="foobarTest1[",
+                  password2="foobarTest1[",
+                  password_old="test",
+                  is_active=True))
+      assert_true(User.objects.get(username="superuser").is_superuser)
+      assert_true(User.objects.get(username="superuser").check_password("foobarTest1["))
+
+      # Test creating a new user
+      response = c.get('/useradmin/users/new')
+      assert_true(password_hint in response.content)
+
+      # Password is more than 8 characters long but does not have a special character
+      response = c.post('/useradmin/users/new',
+                        dict(username="test_user",
+                             is_superuser=False,
+                             password1="foo",
+                             password2="foo"))
+      assert_equal({'password1': [password_error_msg], 'password2': [password_error_msg]},
+                   response.context["form"].errors)
+
+      # Password is more than 8 characters long but does not have a special character
+      response = c.post('/useradmin/users/new',
+                        dict(username="test_user",
+                             is_superuser=False,
+                             password1="foobarTest1",
+                             password2="foobarTest1"))
+
+      assert_equal({'password1': [password_error_msg], 'password2': [password_error_msg]},
+                   response.context["form"].errors)
+
+      # Password1 and Password2 are valid but they do not match
+      response = c.post('/useradmin/users/new',
+                        dict(username="test_user",
+                             is_superuser=False,
+                             password1="foobarTest1[",
+                             password2="foobarTest1?"))
+      assert_equal({'password2': ["Passwords do not match."]}, response.context["form"].errors)
+
+      # Password is valid now
+      c.post('/useradmin/users/new',
+             dict(username="test_user",
+                  is_superuser=False,
+                  password1="foobarTest1[",
+                  password2="foobarTest1[", is_active=True))
+      assert_false(User.objects.get(username="test_user").is_superuser)
+      assert_true(User.objects.get(username="test_user").check_password("foobarTest1["))
+    finally:
+      for reset in resets:
+        reset()
+
+
+  def test_user_admin(self):
+    FUNNY_NAME = '~`!@#$%^&*()_-+={}[]|\;"<>?/,.'
+    FUNNY_NAME_QUOTED = urllib.quote(FUNNY_NAME)
+
+    resets = [
+      useradmin.conf.DEFAULT_USER_GROUP.set_for_testing('test_default'),
+      useradmin.conf.PASSWORD_POLICY.IS_ENABLED.set_for_testing(False),
+    ]
+
+    try:
+      reset_password_policy()
+
+      c = make_logged_in_client('test', is_superuser=True)
+      user = User.objects.get(username='test')
+
+      # Test basic output.
+      response = c.get('/useradmin/')
+      assert_true(len(response.context["users"]) > 0)
+      assert_true("Hue Users" in response.content)
+
+      # Test editing a superuser
+      # Just check that this comes back
+      response = c.get('/useradmin/users/edit/test')
+      # Edit it, to add a first and last name
+      response = c.post('/useradmin/users/edit/test',
+                        dict(username="test",
+                             first_name=u"Inglés",
+                             last_name=u"Español",
+                             is_superuser="True",
+                             is_active="True"),
+                        follow=True)
+      assert_true("User information updated" in response.content,
+                  "Notification should be displayed in: %s" % response.content)
+      # Edit it, can't change username
+      response = c.post('/useradmin/users/edit/test',
+                        dict(username="test2",
+                             first_name=u"Inglés",
+                             last_name=u"Español",
+                             is_superuser="True",
+                             is_active="True"),
+                        follow=True)
+      assert_true("You cannot change a username" in response.content)
+      # Now make sure that those were materialized
+      response = c.get('/useradmin/users/edit/test')
+      assert_equal(smart_unicode("Inglés"), response.context["form"].instance.first_name)
+      assert_true("Español" in response.content)
+      # Shouldn't be able to demote to non-superuser
+      response = c.post('/useradmin/users/edit/test', dict(username="test",
+                            first_name=u"Inglés", last_name=u"Español",
+                            is_superuser=False, is_active=True))
+      assert_true("You cannot remove" in response.content,
+                  "Shouldn't be able to remove the last superuser")
+      # Shouldn't be able to delete oneself
+      response = c.post('/useradmin/users/delete', {u'user_ids': [user.id]})
+      assert_true("You cannot remove yourself" in response.content,
+                  "Shouldn't be able to delete the last superuser")
+
+      # Let's try changing the password
+      response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", is_superuser=True, password1="foo", password2="foobar"))
+      assert_equal(["Passwords do not match."], response.context["form"]["password2"].errors, "Should have complained about mismatched password")
+      # Old password not confirmed
+      response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", password1="foo", password2="foo", is_active=True, is_superuser=True))
+      assert_equal(["The old password does not match the current password."], response.context["form"]["password_old"].errors, "Should have complained about old password")
+      # Good now
+      response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", password1="foo", password2="foo", password_old="test", is_active=True, is_superuser=True))
+      assert_true(User.objects.get(username="test").is_superuser)
+      assert_true(User.objects.get(username="test").check_password("foo"))
+      # Change it back!
+      response = c.post('/useradmin/users/edit/test', dict(username="test", first_name="Tom", last_name="Tester", password1="test", password2="test", password_old="foo", is_active="True", is_superuser="True"))
+      assert_true(User.objects.get(username="test").check_password("test"))
+      assert_true(make_logged_in_client(username = "test", password = "test"), "Check that we can still login.")
+
+      # Check new user form for default group
+      group = get_default_user_group()
+      response = c.get('/useradmin/users/new')
+      assert_true(response)
+      assert_true(('<option value="%s" selected="selected">%s</option>' % (group.id, group.name)) in str(response))
+
+      # Create a new regular user (duplicate name)
+      response = c.post('/useradmin/users/new', dict(username="test", password1="test", password2="test"))
+      assert_equal({ 'username': ["User with this Username already exists."]}, response.context["form"].errors)
+
+      # Create a new regular user (for real)
+      response = c.post('/useradmin/users/new', dict(username=FUNNY_NAME,
+                                               password1="test",
+                                               password2="test",
+                                               is_active="True"))
+      response = c.get('/useradmin/')
+      assert_true(FUNNY_NAME_QUOTED in response.content)
+      assert_true(len(response.context["users"]) > 1)
+      assert_true("Hue Users" in response.content)
+      # Validate profile is created.
+      assert_true(UserProfile.objects.filter(user__username=FUNNY_NAME).exists())
+
+      # Need to give access to the user for the rest of the test
+      group = Group.objects.create(name="test-group")
+      perm = HuePermission.objects.get(app='useradmin', action='access')
+      GroupPermission.objects.create(group=group, hue_permission=perm)
+
+      # Verify that we can modify user groups through the user admin pages
+      response = c.post('/useradmin/users/new', dict(username="group_member", password1="test", password2="test", groups=[group.pk]))
+      User.objects.get(username='group_member')
+      assert_true(User.objects.get(username='group_member').groups.filter(name='test-group').exists())
+      response = c.post('/useradmin/users/edit/group_member', dict(username="group_member", groups=[]))
+      assert_false(User.objects.get(username='group_member').groups.filter(name='test-group').exists())
+
+      # Check permissions by logging in as the new user
+      c_reg = make_logged_in_client(username=FUNNY_NAME, password="test")
+      test_user = User.objects.get(username=FUNNY_NAME)
+      test_user.groups.add(Group.objects.get(name="test-group"))
+      test_user.save()
+
+      # Regular user should be able to modify oneself
+      response = c_reg.post('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,),
+                            dict(username = FUNNY_NAME,
+                                 first_name = "Hello",
+                                 is_active = True,
+                                 groups=[group.id for group in test_user.groups.all()]), follow=True)
+      assert_equal(response.status_code, 200)
+      response = c_reg.get('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,), follow=True)
+      assert_equal(response.status_code, 200)
+      assert_equal("Hello", response.context["form"].instance.first_name)
+      funny_user = User.objects.get(username=FUNNY_NAME)
+      # Can't edit other people.
+      response = c_reg.post("/useradmin/users/delete", {u'user_ids': [funny_user.id]})
+      assert_true("You must be a superuser" in response.content,
+                  "Regular user can't edit other people")
+
+      # Revert to regular "test" user, that has superuser powers.
+      c_su = make_logged_in_client()
+      # Inactivate FUNNY_NAME
+      c_su.post('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,),
+                            dict(username = FUNNY_NAME,
+                                 first_name = "Hello",
+                                 is_active = False))
+      # Now make sure FUNNY_NAME can't log back in
+      response = c_reg.get('/useradmin/users/edit/%s' % (FUNNY_NAME_QUOTED,))
+      assert_true(response.status_code == 302 and "login" in response["location"],
+                  "Inactivated user gets redirected to login page")
+
+      # Delete that regular user
+      funny_profile = get_profile(test_user)
+      response = c_su.post('/useradmin/users/delete', {u'user_ids': [funny_user.id]})
+      assert_equal(302, response.status_code)
+      assert_false(User.objects.filter(username=FUNNY_NAME).exists())
+      assert_false(UserProfile.objects.filter(id=funny_profile.id).exists())
+
+      # Bulk delete users
+      u1 = User.objects.create(username='u1', password="u1")
+      u2 = User.objects.create(username='u2', password="u2")
+      assert_equal(User.objects.filter(username__in=['u1', 'u2']).count(), 2)
+      response = c_su.post('/useradmin/users/delete', {u'user_ids': [u1.id, u2.id]})
+      assert_equal(User.objects.filter(username__in=['u1', 'u2']).count(), 0)
+
+      # Make sure that user deletion works if the user has never performed a request.
+      funny_user = User.objects.create(username=FUNNY_NAME, password='test')
+      assert_true(User.objects.filter(username=FUNNY_NAME).exists())
+      assert_false(UserProfile.objects.filter(user__username=FUNNY_NAME).exists())
+      response = c_su.post('/useradmin/users/delete', {u'user_ids': [funny_user.id]})
+      assert_equal(302, response.status_code)
+      assert_false(User.objects.filter(username=FUNNY_NAME).exists())
+      assert_false(UserProfile.objects.filter(user__username=FUNNY_NAME).exists())
+
+      # You shouldn't be able to create a user without a password
+      response = c_su.post('/useradmin/users/new', dict(username="test"))
+      assert_true("You must specify a password when creating a new user." in response.content)
+    finally:
+      for reset in resets:
+        reset()
+
+
+
+  def test_list_for_autocomplete(self):
+    # Now the autocomplete has access to all the users and groups
+    c1 = make_logged_in_client('test_list_for_autocomplete', is_superuser=False, groupname='test_list_for_autocomplete')
+    c2_same_group = make_logged_in_client('test_list_for_autocomplete2', is_superuser=False, groupname='test_list_for_autocomplete')
+    c3_other_group = make_logged_in_client('test_list_for_autocomplete3', is_superuser=False, groupname='test_list_for_autocomplete_other_group')
+
+    # c1 is in the same group as c2
+    response = c1.get(reverse('useradmin.views.list_for_autocomplete'), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+    content = json.loads(response.content)
+
+    users = [smart_unicode(user['username']) for user in content['users']]
+    groups = [smart_unicode(user['name']) for user in content['groups']]
+
+    assert_equal([u'test_list_for_autocomplete2', u'test_list_for_autocomplete3'], users)
+    assert_true(u'test_list_for_autocomplete' in groups, groups)
+    assert_true(u'test_list_for_autocomplete_other_group' in groups, groups)
+
+    # c2 is in the same group as c1
+    response = c2_same_group.get(reverse('useradmin.views.list_for_autocomplete'), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+    content = json.loads(response.content)
+
+    users = [smart_unicode(user['username']) for user in content['users']]
+    groups = [smart_unicode(user['name']) for user in content['groups']]
+
+    assert_equal([u'test_list_for_autocomplete', u'test_list_for_autocomplete3'], users)
+    assert_true(u'test_list_for_autocomplete' in groups, groups)
+    assert_true(u'test_list_for_autocomplete_other_group' in groups, groups)
+
+    # c3 is alone except for groups
+    response = c3_other_group.get(reverse('useradmin.views.list_for_autocomplete'), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+    content = json.loads(response.content)
+
+    users = [smart_unicode(user['username']) for user in content['users']]
+    groups = [smart_unicode(user['name']) for user in content['groups']]
+
+    assert_equal([u'test_list_for_autocomplete', u'test_list_for_autocomplete2'], users)
+    assert_true(u'test_list_for_autocomplete' in groups, groups)
+    assert_true(u'test_list_for_autocomplete_other_group' in groups, groups)
+
+
+class TestUserAdminWithHadoop(BaseUserAdminTests):
+  requires_hadoop = True
+
+  def test_ensure_home_directory(self):
+    raise SkipTest
+
+    resets = [
+      useradmin.conf.PASSWORD_POLICY.IS_ENABLED.set_for_testing(False),
+    ]
+
+    try:
+      reset_password_policy()
+
+      # Cluster and client for home directory creation
+      cluster = pseudo_hdfs4.shared_cluster()
+      c = make_logged_in_client(cluster.superuser, is_superuser=True, groupname='test1')
+      cluster.fs.setuser(cluster.superuser)
+
+      # Create a user with a home directory
+      assert_false(cluster.fs.exists('/user/test1'))
+      response = c.post('/useradmin/users/new', dict(username="test1", password1='test', password2='test', ensure_home_directory=True))
+      assert_true(cluster.fs.exists('/user/test1'))
+      dir_stat = cluster.fs.stats('/user/test1')
+      assert_equal('test1', dir_stat.user)
+      assert_equal('test1', dir_stat.group)
+      assert_equal('40755', '%o' % dir_stat.mode)
+
+      # Create a user, then add their home directory
+      assert_false(cluster.fs.exists('/user/test2'))
+      response = c.post('/useradmin/users/new', dict(username="test2", password1='test', password2='test'))
+      assert_false(cluster.fs.exists('/user/test2'))
+      response = c.post('/useradmin/users/edit/%s' % "test2", dict(username="test2", password1='test', password2='test', password_old="test", ensure_home_directory=True))
+      assert_true(cluster.fs.exists('/user/test2'))
+      dir_stat = cluster.fs.stats('/user/test2')
+      assert_equal('test2', dir_stat.user)
+      assert_equal('test2', dir_stat.group)
+      assert_equal('40755', '%o' % dir_stat.mode)
+    finally:
+      for reset in resets:
+        reset()
 
 class MockLdapConnection(object):
   def __init__(self, ldap_config, ldap_url, username, password, ldap_cert):

+ 3 - 1
desktop/core/src/desktop/lib/test_utils.py

@@ -40,7 +40,9 @@ def add_permission(username, groupname, permname, appname):
 
 def add_to_group(username, groupname=None):
     if groupname is None:
-        groupname = get_default_user_group().name
+        group = get_default_user_group()
+        assert group is not None
+        groupname = group.name
 
     user = User.objects.get(username=username)
     group, created = Group.objects.get_or_create(name=groupname)

Một số tệp đã không được hiển thị bởi vì quá nhiều tập tin thay đổi trong này khác