|
|
@@ -17,9 +17,9 @@
|
|
|
|
|
|
from nose.tools import assert_true, assert_false, assert_equal
|
|
|
|
|
|
+from django.conf import settings
|
|
|
from django.contrib.auth.models import User
|
|
|
from django.test.client import Client
|
|
|
-from django.conf import settings
|
|
|
|
|
|
from desktop import conf, middleware
|
|
|
from desktop.auth import backend
|
|
|
@@ -35,19 +35,44 @@ from useradmin.views import import_ldap_groups
|
|
|
|
|
|
class TestLoginWithHadoop(PseudoHdfsTestBase):
|
|
|
|
|
|
- def setUp(self):
|
|
|
+ reset = []
|
|
|
+ test_username = 'test_login_with_hadoop'
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def setup_class(cls):
|
|
|
# Simulate first login ever
|
|
|
User.objects.all().delete()
|
|
|
+ PseudoHdfsTestBase.setup_class()
|
|
|
+
|
|
|
+ cls.auth_backends = settings.AUTHENTICATION_BACKENDS
|
|
|
+ settings.AUTHENTICATION_BACKENDS = ('desktop.auth.backend.AllowFirstUserDjangoBackend',)
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def teardown_class(cls):
|
|
|
+ settings.AUTHENTICATION_BACKENDS = cls.auth_backends
|
|
|
+
|
|
|
+ def setUp(self):
|
|
|
self.c = Client()
|
|
|
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(['desktop.auth.backend.AllowFirstUserDjangoBackend']) )
|
|
|
+
|
|
|
+ def tearDown(self):
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
+ for finish in self.reset:
|
|
|
+ finish()
|
|
|
+
|
|
|
+ if self.fs.exists("/user/%s" % self.test_username):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/%s" % self.test_username)
|
|
|
+
|
|
|
def test_login(self):
|
|
|
response = self.c.get('/accounts/login/')
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_true(response.context['first_login_ever'])
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', dict(username="foo", password="foo"))
|
|
|
+ response = self.c.post('/accounts/login/', dict(username=self.test_username, password="foo"))
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
- assert_true(self.fs.exists("/user/foo"))
|
|
|
+ assert_true(self.fs.exists("/user/%s" % self.test_username))
|
|
|
|
|
|
response = self.c.get('/accounts/login/')
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
@@ -61,59 +86,66 @@ class TestLoginWithHadoop(PseudoHdfsTestBase):
|
|
|
# Create home directory as a file in order to fail in the home creation later
|
|
|
cluster = pseudo_hdfs4.shared_cluster()
|
|
|
fs = cluster.fs
|
|
|
- assert_false(cluster.fs.exists("/user/test-hue-foo2"))
|
|
|
+ assert_false(cluster.fs.exists("/user/%s" % self.test_username))
|
|
|
+ fs.do_as_superuser(fs.create, "/user/%s" % self.test_username)
|
|
|
|
|
|
- try:
|
|
|
- fs.do_as_superuser(fs.create, "/user/test-hue-foo2")
|
|
|
-
|
|
|
- response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "test-hue-foo2",
|
|
|
+ response = self.c.post('/accounts/login/', {
|
|
|
+ 'username': self.test_username,
|
|
|
'password': "test-hue-foo2",
|
|
|
}, follow=True)
|
|
|
|
|
|
- assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
- assert_true('/beeswax' in response.content, response.content)
|
|
|
- # Custom login process should not do 'http-equiv="refresh"' but call the correct view
|
|
|
- # 'Could not create home directory.' won't show up because the messages are consumed before
|
|
|
- finally:
|
|
|
- fs.do_as_superuser(fs.remove, '/user/test-hue-foo2')
|
|
|
+ assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
+ assert_true('/beeswax' in response.content, response.content)
|
|
|
+ # Custom login process should not do 'http-equiv="refresh"' but call the correct view
|
|
|
+ # 'Could not create home directory.' won't show up because the messages are consumed before
|
|
|
|
|
|
|
|
|
class TestLdapLogin(PseudoHdfsTestBase):
|
|
|
+
|
|
|
reset = []
|
|
|
+ test_username = 'test_ldap_login'
|
|
|
|
|
|
@classmethod
|
|
|
def setup_class(cls):
|
|
|
+ # Simulate first login ever
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
PseudoHdfsTestBase.setup_class()
|
|
|
|
|
|
- cls.backend = django_auth_ldap_backend.LDAPBackend
|
|
|
+ cls.ldap_backend = django_auth_ldap_backend.LDAPBackend
|
|
|
django_auth_ldap_backend.LDAPBackend = MockLdapBackend
|
|
|
|
|
|
+ # Override auth backend, settings are only loaded from conf at initialization so we can't use set_for_testing
|
|
|
+ cls.auth_backends = settings.AUTHENTICATION_BACKENDS
|
|
|
+ settings.AUTHENTICATION_BACKENDS = ('desktop.auth.backend.LdapBackend',)
|
|
|
+
|
|
|
# Need to recreate LdapBackend class with new monkey patched base class
|
|
|
reload(backend)
|
|
|
|
|
|
- cls.old_backends = settings.AUTHENTICATION_BACKENDS
|
|
|
- settings.AUTHENTICATION_BACKENDS = ("desktop.auth.backend.LdapBackend",)
|
|
|
-
|
|
|
@classmethod
|
|
|
def teardown_class(cls):
|
|
|
- django_auth_ldap_backend.LDAPBackend = cls.backend
|
|
|
+ django_auth_ldap_backend.LDAPBackend = cls.ldap_backend
|
|
|
+
|
|
|
+ settings.AUTHENTICATION_BACKENDS = cls.auth_backends
|
|
|
|
|
|
- # Need to recreate LdapBackend class with old base class
|
|
|
reload(backend)
|
|
|
|
|
|
- settings.AUTHENTICATION_BACKENDS = cls.old_backends
|
|
|
+ def setUp(self):
|
|
|
+ self.c = Client()
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(['desktop.auth.backend.LdapBackend']) )
|
|
|
+ self.reset.append(conf.LDAP.LDAP_URL.set_for_testing('does not matter'))
|
|
|
|
|
|
def tearDown(self):
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
for finish in self.reset:
|
|
|
finish()
|
|
|
|
|
|
- def setUp(self):
|
|
|
- # Simulate first login ever
|
|
|
- User.objects.all().delete()
|
|
|
- self.c = Client()
|
|
|
+ if self.fs.exists("/user/%s" % self.test_username):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/%s" % self.test_username)
|
|
|
|
|
|
- self.reset.append(conf.LDAP.LDAP_URL.set_for_testing('does not matter'))
|
|
|
+ if self.fs.exists("/user/curly"):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/curly")
|
|
|
|
|
|
def test_login(self):
|
|
|
response = self.c.get('/accounts/login/')
|
|
|
@@ -121,11 +153,11 @@ class TestLdapLogin(PseudoHdfsTestBase):
|
|
|
assert_false(response.context['first_login_ever'])
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "ldap1",
|
|
|
+ 'username': self.test_username,
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
- assert_true(self.fs.exists("/user/ldap1"))
|
|
|
+ assert_true(self.fs.exists("/user/%s" % self.test_username))
|
|
|
|
|
|
response = self.c.get('/accounts/login/')
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
@@ -139,48 +171,44 @@ class TestLdapLogin(PseudoHdfsTestBase):
|
|
|
# Create home directory as a file in order to fail in the home creation later
|
|
|
cluster = pseudo_hdfs4.shared_cluster()
|
|
|
fs = cluster.fs
|
|
|
- assert_false(cluster.fs.exists("/user/test-hue-ldap2"))
|
|
|
-
|
|
|
- try:
|
|
|
- fs.do_as_superuser(fs.create, "/user/test-hue-ldap2")
|
|
|
+ assert_false(cluster.fs.exists("/user/%s" % self.test_username))
|
|
|
+ fs.do_as_superuser(fs.create, "/user/%s" % self.test_username)
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "test-hue-ldap2",
|
|
|
- 'password': "test-hue-ldap2",
|
|
|
- }, follow=True)
|
|
|
- assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
- assert_true('/beeswax' in response.content, response.content)
|
|
|
- # Custom login process should not do 'http-equiv="refresh"' but call the correct view
|
|
|
- # 'Could not create home directory.' won't show up because the messages are consumed before
|
|
|
- finally:
|
|
|
- fs.do_as_superuser(fs.remove, '/user/test-hue-ldap2')
|
|
|
+ response = self.c.post('/accounts/login/', {
|
|
|
+ 'username': self.test_username,
|
|
|
+ 'password': "test-hue-ldap2",
|
|
|
+ }, follow=True)
|
|
|
+ assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
+ assert_true('/beeswax' in response.content, response.content)
|
|
|
+ # Custom login process should not do 'http-equiv="refresh"' but call the correct view
|
|
|
+ # 'Could not create home directory.' won't show up because the messages are consumed before
|
|
|
|
|
|
def test_login_ignore_case(self):
|
|
|
self.reset.append(conf.LDAP.IGNORE_USERNAME_CASE.set_for_testing(True))
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "LDAP1",
|
|
|
+ 'username': self.test_username.upper(),
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('ldap1', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
self.c.logout()
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "ldap1",
|
|
|
+ 'username': self.test_username,
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('ldap1', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
def test_login_force_lower_case(self):
|
|
|
self.reset.append(conf.LDAP.FORCE_USERNAME_LOWERCASE.set_for_testing(True))
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "LDAP1",
|
|
|
+ 'username': self.test_username.upper(),
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
@@ -189,34 +217,34 @@ class TestLdapLogin(PseudoHdfsTestBase):
|
|
|
self.c.logout()
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "ldap1",
|
|
|
+ 'username': self.test_username,
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('ldap1', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
def test_login_force_lower_case_and_ignore_case(self):
|
|
|
self.reset.append(conf.LDAP.IGNORE_USERNAME_CASE.set_for_testing(True))
|
|
|
self.reset.append(conf.LDAP.FORCE_USERNAME_LOWERCASE.set_for_testing(True))
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "LDAP1",
|
|
|
+ 'username': self.test_username.upper(),
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('ldap1', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
self.c.logout()
|
|
|
|
|
|
response = self.c.post('/accounts/login/', {
|
|
|
- 'username': "ldap1",
|
|
|
+ 'username': self.test_username,
|
|
|
'password': "ldap1"
|
|
|
})
|
|
|
assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('ldap1', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
def test_import_groups_on_login(self):
|
|
|
self.reset.append(conf.LDAP.SYNC_GROUPS_ON_LOGIN.set_for_testing(True))
|
|
|
@@ -235,35 +263,56 @@ class TestLdapLogin(PseudoHdfsTestBase):
|
|
|
assert_equal(3, User.objects.all()[0].groups.all().count(), User.objects.all()[0].groups.all())
|
|
|
|
|
|
|
|
|
-class TestRemoteUserLogin(object):
|
|
|
+class TestRemoteUserLogin(PseudoHdfsTestBase):
|
|
|
+
|
|
|
reset = []
|
|
|
+ test_username = "test_remote_user_login"
|
|
|
|
|
|
- def setUp(self):
|
|
|
+ @classmethod
|
|
|
+ def setup_class(cls):
|
|
|
+ # Simulate first login ever
|
|
|
User.objects.all().delete()
|
|
|
- self.reset.append( conf.AUTH.BACKEND.set_for_testing('desktop.auth.backend.RemoteUserDjangoBackend') )
|
|
|
- self.reset.append( conf.AUTH.REMOTE_USER_HEADER.set_for_testing('REMOTE_USER') ) # Set for middlware
|
|
|
- self.backends = settings.AUTHENTICATION_BACKENDS
|
|
|
+
|
|
|
+ PseudoHdfsTestBase.setup_class()
|
|
|
+
|
|
|
+ cls.auth_backends = settings.AUTHENTICATION_BACKENDS
|
|
|
settings.AUTHENTICATION_BACKENDS = ('desktop.auth.backend.RemoteUserDjangoBackend',)
|
|
|
- self.remote_user_middleware_header = middleware.HueRemoteUserMiddleware.header
|
|
|
- middleware.HueRemoteUserMiddleware.header = conf.AUTH.REMOTE_USER_HEADER.get() # Set for middlware
|
|
|
+ cls.remote_user_middleware_header = middleware.HueRemoteUserMiddleware.header
|
|
|
+ middleware.HueRemoteUserMiddleware.header = conf.AUTH.REMOTE_USER_HEADER.get()
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def teardown_class(cls):
|
|
|
+ middleware.HueRemoteUserMiddleware.header = cls.remote_user_middleware_header
|
|
|
+ settings.AUTHENTICATION_BACKENDS = cls.auth_backends
|
|
|
+
|
|
|
+ def setUp(self):
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(['desktop.auth.backend.RemoteUserDjangoBackend']) )
|
|
|
+ self.reset.append( conf.AUTH.REMOTE_USER_HEADER.set_for_testing('REMOTE_USER') ) # Set for middleware
|
|
|
+
|
|
|
self.c = Client()
|
|
|
|
|
|
def tearDown(self):
|
|
|
- middleware.HueRemoteUserMiddleware.header = self.remote_user_middleware_header
|
|
|
- settings.AUTHENTICATION_BACKENDS = self.backends
|
|
|
for finish in self.reset:
|
|
|
finish()
|
|
|
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
+ if self.fs.exists("/user/%s" % self.test_username):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/%s" % self.test_username)
|
|
|
+
|
|
|
+ if self.fs.exists("/user/%s_%s" % (self.test_username, '2')):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/%s_%s" % (self.test_username, '2'))
|
|
|
+
|
|
|
def test_normal(self):
|
|
|
response = self.c.get('/accounts/login/')
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_false(response.context['first_login_ever'])
|
|
|
|
|
|
assert_equal(0, len(User.objects.all()))
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "foo3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('foo3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
def test_ignore_case(self):
|
|
|
self.reset.append( conf.AUTH.IGNORE_USERNAME_CASE.set_for_testing(True) )
|
|
|
@@ -272,25 +321,25 @@ class TestRemoteUserLogin(object):
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_false(response.context['first_login_ever'])
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "foo3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('foo3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "FOO3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username.upper()})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('foo3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "FOO4"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "%s_%s" % (self.test_username.upper(), '2')})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(2, len(User.objects.all().order_by('username')))
|
|
|
- assert_equal('foo4', User.objects.all()[1].username)
|
|
|
+ assert_equal("%s_%s" % (self.test_username, '2'), User.objects.all()[1].username)
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "foo4"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "%s_%s" % (self.test_username, '2')})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(2, len(User.objects.all()))
|
|
|
- assert_equal('foo4', User.objects.all().order_by('username')[1].username)
|
|
|
+ assert_equal("%s_%s" % (self.test_username, '2'), User.objects.all().order_by('username')[1].username)
|
|
|
|
|
|
def test_force_lower_case(self):
|
|
|
self.reset.append( conf.AUTH.FORCE_USERNAME_LOWERCASE.set_for_testing(True) )
|
|
|
@@ -299,23 +348,24 @@ class TestRemoteUserLogin(object):
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_false(response.context['first_login_ever'])
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "foo3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('foo3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "FOO3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username.upper()})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('foo3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username, User.objects.all()[0].username)
|
|
|
+
|
|
|
|
|
|
def test_ignore_case_and_force_lower_case(self):
|
|
|
reset = conf.AUTH.FORCE_USERNAME_LOWERCASE.set_for_testing(False)
|
|
|
try:
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "FOO3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username.upper()})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('FOO3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username.upper(), User.objects.all()[0].username)
|
|
|
finally:
|
|
|
reset()
|
|
|
|
|
|
@@ -323,32 +373,119 @@ class TestRemoteUserLogin(object):
|
|
|
self.reset.append( conf.AUTH.IGNORE_USERNAME_CASE.set_for_testing(True) )
|
|
|
|
|
|
# Previously existing users should not be forced to lower case.
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "FOO3"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": self.test_username.upper()})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(1, len(User.objects.all()))
|
|
|
- assert_equal('FOO3', User.objects.all()[0].username)
|
|
|
+ assert_equal(self.test_username.upper(), User.objects.all()[0].username)
|
|
|
|
|
|
# New users should be forced to lowercase.
|
|
|
- response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "FOO4"})
|
|
|
+ response = self.c.post('/accounts/login/', {}, **{"REMOTE_USER": "%s_%s" % (self.test_username.upper(), '2')})
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_equal(2, len(User.objects.all()))
|
|
|
- assert_equal('foo4', User.objects.all().order_by('username')[1].username)
|
|
|
+ assert_equal("%s_%s" % (self.test_username, '2'), User.objects.all().order_by('username')[1].username)
|
|
|
+
|
|
|
|
|
|
+class TestMultipleBackendLogin(PseudoHdfsTestBase):
|
|
|
|
|
|
-class TestLogin(object):
|
|
|
reset = []
|
|
|
+ test_username = "test_multiple_login"
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def setup_class(cls):
|
|
|
+ # Simulate first login ever
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
+ PseudoHdfsTestBase.setup_class()
|
|
|
+
|
|
|
+ cls.ldap_backend = django_auth_ldap_backend.LDAPBackend
|
|
|
+ django_auth_ldap_backend.LDAPBackend = MockLdapBackend
|
|
|
+
|
|
|
+ # Override auth backend, settings are only loaded from conf at initialization so we can't use set_for_testing
|
|
|
+ cls.auth_backends = settings.AUTHENTICATION_BACKENDS
|
|
|
+ settings.AUTHENTICATION_BACKENDS = ('desktop.auth.backend.LdapBackend','desktop.auth.backend.AllowFirstUserDjangoBackend')
|
|
|
+
|
|
|
+ # Need to recreate LdapBackend class with new monkey patched base class
|
|
|
+ reload(backend)
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def teardown_class(cls):
|
|
|
+ django_auth_ldap_backend.LDAPBackend = cls.ldap_backend
|
|
|
+
|
|
|
+ settings.AUTHENTICATION_BACKENDS = cls.auth_backends
|
|
|
+
|
|
|
+ reload(backend)
|
|
|
|
|
|
def setUp(self):
|
|
|
+ self.c = Client()
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(['desktop.auth.backend.LdapBackend','desktop.auth.backend.AllowFirstUserDjangoBackend']))
|
|
|
+ self.reset.append(conf.LDAP.LDAP_URL.set_for_testing('does not matter'))
|
|
|
+
|
|
|
+ def tearDown(self):
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
+ for finish in self.reset:
|
|
|
+ finish()
|
|
|
+
|
|
|
+ if self.fs.exists("/user/%s" % self.test_username):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/%s" % self.test_username)
|
|
|
+
|
|
|
+ def test_login_with_ldap(self):
|
|
|
+ ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
|
|
|
+
|
|
|
+ response = self.c.post('/accounts/login/', {
|
|
|
+ 'username': "curly",
|
|
|
+ 'password': "ldap1"
|
|
|
+ })
|
|
|
+ assert_equal(302, response.status_code, response.status_code)
|
|
|
+ assert_equal(1, len(User.objects.all()))
|
|
|
+
|
|
|
+ def test_fallback_to_db(self):
|
|
|
+ ldap_access.CACHED_LDAP_CONN = LdapTestConnection()
|
|
|
+
|
|
|
+ client = make_logged_in_client(username=self.test_username, password="test")
|
|
|
+ client.get('/accounts/logout')
|
|
|
+ user = User.objects.get(username=self.test_username)
|
|
|
+
|
|
|
+ response = self.c.post('/accounts/login/', dict(username=self.test_username, password="foo"))
|
|
|
+ assert_equal(302, response.status_code, "Expected ok redirect status.")
|
|
|
+ assert_true(self.fs.exists("/user/%s" % self.test_username))
|
|
|
+
|
|
|
+
|
|
|
+class TestLogin(PseudoHdfsTestBase):
|
|
|
+
|
|
|
+ reset = []
|
|
|
+ test_username = "test_login"
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def setup_class(cls):
|
|
|
# Simulate first login ever
|
|
|
User.objects.all().delete()
|
|
|
+
|
|
|
+ PseudoHdfsTestBase.setup_class()
|
|
|
+
|
|
|
+ cls.auth_backends = settings.AUTHENTICATION_BACKENDS
|
|
|
+ settings.AUTHENTICATION_BACKENDS = ('desktop.auth.backend.AllowFirstUserDjangoBackend',)
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def teardown_class(cls):
|
|
|
+ settings.AUTHENTICATION_BACKENDS = cls.auth_backends
|
|
|
+
|
|
|
+ def setUp(self):
|
|
|
self.c = Client()
|
|
|
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(['desktop.auth.backend.AllowFirstUserDjangoBackend']) )
|
|
|
+
|
|
|
def tearDown(self):
|
|
|
for finish in self.reset:
|
|
|
finish()
|
|
|
|
|
|
+ User.objects.all().delete()
|
|
|
+
|
|
|
+ if self.fs.exists("/user/%s" % self.test_username):
|
|
|
+ self.cluster.fs.do_as_superuser(self.cluster.fs.rmtree, "/user/%s" % self.test_username)
|
|
|
+
|
|
|
def test_bad_first_user(self):
|
|
|
- self.reset.append( conf.AUTH.BACKEND.set_for_testing("desktop.auth.backend.AllowFirstUserDjangoBackend") )
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(["desktop.auth.backend.AllowFirstUserDjangoBackend"]) )
|
|
|
|
|
|
response = self.c.get('/accounts/login/')
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
@@ -359,44 +496,44 @@ class TestLogin(object):
|
|
|
assert_true('This value may contain only letters, numbers and @/./+/-/_ characters.' in response.content, response)
|
|
|
|
|
|
def test_non_jframe_login(self):
|
|
|
- client = make_logged_in_client(username="test", password="test")
|
|
|
+ client = make_logged_in_client(username=self.test_username, password="test")
|
|
|
# Logout first
|
|
|
client.get('/accounts/logout')
|
|
|
# Login
|
|
|
- response = client.post('/accounts/login/', dict(username="test", password="test"), follow=True)
|
|
|
+ response = client.post('/accounts/login/', dict(username=self.test_username, password="test"), follow=True)
|
|
|
assert_true(any(["admin_wizard.mako" in _template.filename for _template in response.templates]), response.content) # Go to superuser wizard
|
|
|
|
|
|
def test_login_expiration(self):
|
|
|
""" Expiration test without superusers """
|
|
|
old_settings = settings.ADMINS
|
|
|
- self.reset.append( conf.AUTH.BACKEND.set_for_testing("desktop.auth.backend.AllowFirstUserDjangoBackend") )
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(["desktop.auth.backend.AllowFirstUserDjangoBackend"]) )
|
|
|
self.reset.append( conf.AUTH.EXPIRES_AFTER.set_for_testing(0) )
|
|
|
self.reset.append( conf.AUTH.EXPIRE_SUPERUSERS.set_for_testing(False) )
|
|
|
|
|
|
- client = make_logged_in_client(username="test", password="test")
|
|
|
+ client = make_logged_in_client(username=self.test_username, password="test")
|
|
|
client.get('/accounts/logout')
|
|
|
- user = User.objects.get(username="test")
|
|
|
+ user = User.objects.get(username=self.test_username)
|
|
|
|
|
|
# Login successfully
|
|
|
try:
|
|
|
user.is_superuser = True
|
|
|
user.save()
|
|
|
- response = client.post('/accounts/login/', dict(username="test", password="test"), follow=True)
|
|
|
+ response = client.post('/accounts/login/', dict(username=self.test_username, password="test"), follow=True)
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
|
|
|
client.get('/accounts/logout')
|
|
|
|
|
|
# Login fail
|
|
|
- settings.ADMINS = [('test', 'test@test.com')]
|
|
|
+ settings.ADMINS = [(self.test_username, 'test@test.com')]
|
|
|
user.is_superuser = False
|
|
|
user.save()
|
|
|
- response = client.post('/accounts/login/', dict(username="test", password="test"), follow=True)
|
|
|
+ response = client.post('/accounts/login/', dict(username=self.test_username, password="test"), follow=True)
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_true('Account deactivated. Please contact an <a href="mailto:test@test.com">administrator</a>' in response.content, response.content)
|
|
|
|
|
|
# Failure should report an inactive user without admin link
|
|
|
settings.ADMINS = []
|
|
|
- response = client.post('/accounts/login/', dict(username="test", password="test"), follow=True)
|
|
|
+ response = client.post('/accounts/login/', dict(username=self.test_username, password="test"), follow=True)
|
|
|
assert_equal(200, response.status_code, "Expected ok status.")
|
|
|
assert_true("Account deactivated. Please contact an administrator." in response.content, response.content)
|
|
|
finally:
|
|
|
@@ -404,18 +541,18 @@ class TestLogin(object):
|
|
|
|
|
|
def test_login_expiration_with_superusers(self):
|
|
|
""" Expiration test with superusers """
|
|
|
- self.reset.append( conf.AUTH.BACKEND.set_for_testing("desktop.auth.backend.AllowFirstUserDjangoBackend") )
|
|
|
+ self.reset.append( conf.AUTH.BACKEND.set_for_testing(["desktop.auth.backend.AllowFirstUserDjangoBackend"]) )
|
|
|
self.reset.append( conf.AUTH.EXPIRES_AFTER.set_for_testing(0) )
|
|
|
self.reset.append( conf.AUTH.EXPIRE_SUPERUSERS.set_for_testing(True) )
|
|
|
|
|
|
- client = make_logged_in_client(username="test", password="test")
|
|
|
+ client = make_logged_in_client(username=self.test_username, password="test")
|
|
|
client.get('/accounts/logout')
|
|
|
- user = User.objects.get(username="test")
|
|
|
+ user = User.objects.get(username=self.test_username)
|
|
|
|
|
|
# Login fail
|
|
|
user.is_superuser = True
|
|
|
user.save()
|
|
|
- response = client.post('/accounts/login/', dict(username="test", password="test"), follow=True)
|
|
|
+ response = client.post('/accounts/login/', dict(username=self.test_username, password="test"), follow=True)
|
|
|
assert_equal(200, response.status_code, "Expected unauthorized status.")
|
|
|
|
|
|
|