add initial osusers unit tests
This commit is contained in:
parent
e8285518a3
commit
b9dd34d527
4 changed files with 97 additions and 27 deletions
|
@ -25,6 +25,10 @@ from .tasks import (
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
CANNOT_USE_PRIMARY_GROUP_AS_ADDITIONAL = _(
|
||||||
|
"You can not use a user's primary group.")
|
||||||
|
|
||||||
|
|
||||||
class TaskResult(TimeStampedModel, models.Model):
|
class TaskResult(TimeStampedModel, models.Model):
|
||||||
|
|
||||||
task_uuid = models.CharField(primary_key=True, max_length=64, blank=False)
|
task_uuid = models.CharField(primary_key=True, max_length=64, blank=False)
|
||||||
|
@ -47,7 +51,7 @@ class TaskResult(TimeStampedModel, models.Model):
|
||||||
|
|
||||||
def update_taskstatus(self):
|
def update_taskstatus(self):
|
||||||
if not self.is_finished:
|
if not self.is_finished:
|
||||||
asyncresult = AsyncResult(self.task_uuid, task_name=self.task_name)
|
asyncresult = AsyncResult(self.task_uuid)
|
||||||
self._set_result_fields(asyncresult)
|
self._set_result_fields(asyncresult)
|
||||||
self.save()
|
self.save()
|
||||||
|
|
||||||
|
@ -83,23 +87,26 @@ class Group(TimeStampedModel, models.Model):
|
||||||
def save(self, *args, **kwargs):
|
def save(self, *args, **kwargs):
|
||||||
super(Group, self).save(*args, **kwargs)
|
super(Group, self).save(*args, **kwargs)
|
||||||
GroupTaskResult.objects.create_grouptaskresult(
|
GroupTaskResult.objects.create_grouptaskresult(
|
||||||
self, create_ldap_group.delay(self.groupname, self.gid, self.descr)
|
self,
|
||||||
|
create_ldap_group.delay(self.groupname, self.gid, self.descr),
|
||||||
|
'create_ldap_group'
|
||||||
)
|
)
|
||||||
return self
|
return self
|
||||||
|
|
||||||
def delete(self, *args, **kwargs):
|
def delete(self, *args, **kwargs):
|
||||||
DeleteTaskResult.objects.create_deletetaskresult(
|
DeleteTaskResult.objects.create_deletetaskresult(
|
||||||
'group', self.groupname,
|
'group', self.groupname,
|
||||||
delete_ldap_group_if_empty.delay(self.groupname)
|
delete_ldap_group_if_empty.delay(self.groupname),
|
||||||
|
'delete_ldap_group'
|
||||||
)
|
)
|
||||||
super(Group, self).delete(*args, **kwargs)
|
super(Group, self).delete(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
class TaskResultManager(models.Manager):
|
class TaskResultManager(models.Manager):
|
||||||
|
|
||||||
def create(self, asyncresult):
|
def create(self, asyncresult, task_name):
|
||||||
result = self.model(
|
result = self.model(
|
||||||
task_uuid=asyncresult.task_id, task_name=asyncresult.task_name
|
task_uuid=asyncresult.task_id, task_name=task_name
|
||||||
)
|
)
|
||||||
result._set_result_fields(asyncresult)
|
result._set_result_fields(asyncresult)
|
||||||
return result
|
return result
|
||||||
|
@ -107,9 +114,11 @@ class TaskResultManager(models.Manager):
|
||||||
|
|
||||||
class DeleteTaskResultManager(TaskResultManager):
|
class DeleteTaskResultManager(TaskResultManager):
|
||||||
|
|
||||||
def create_deletetaskresult(self, modeltype, modelname, asyncresult):
|
def create_deletetaskresult(
|
||||||
|
self, modeltype, modelname, asyncresult, task_name
|
||||||
|
):
|
||||||
taskresult = super(DeleteTaskResultManager, self).create(
|
taskresult = super(DeleteTaskResultManager, self).create(
|
||||||
asyncresult)
|
asyncresult, task_name)
|
||||||
taskresult.modeltype = modeltype
|
taskresult.modeltype = modeltype
|
||||||
taskresult.modelname = modelname
|
taskresult.modelname = modelname
|
||||||
taskresult.save()
|
taskresult.save()
|
||||||
|
@ -126,9 +135,11 @@ class DeleteTaskResult(TaskResult):
|
||||||
|
|
||||||
class GroupTaskResultManager(TaskResultManager):
|
class GroupTaskResultManager(TaskResultManager):
|
||||||
|
|
||||||
def create_grouptaskresult(self, group, asyncresult, commit=False):
|
def create_grouptaskresult(
|
||||||
|
self, group, asyncresult, task_name, commit=False
|
||||||
|
):
|
||||||
taskresult = super(GroupTaskResultManager, self).create(
|
taskresult = super(GroupTaskResultManager, self).create(
|
||||||
asyncresult)
|
asyncresult, task_name)
|
||||||
taskresult.group = group
|
taskresult.group = group
|
||||||
taskresult.save()
|
taskresult.save()
|
||||||
return taskresult
|
return taskresult
|
||||||
|
@ -176,7 +187,8 @@ class UserManager(models.Manager):
|
||||||
group = Group.objects.create(groupname=username, gid=gid)
|
group = Group.objects.create(groupname=username, gid=gid)
|
||||||
GroupTaskResult.objects.create_grouptaskresult(
|
GroupTaskResult.objects.create_grouptaskresult(
|
||||||
group,
|
group,
|
||||||
create_ldap_group.delay(group.groupname, group.gid, group.descr)
|
create_ldap_group.delay(group.groupname, group.gid, group.descr),
|
||||||
|
'create_ldap_group'
|
||||||
)
|
)
|
||||||
user = self.create(username=username, group=group, uid=uid,
|
user = self.create(username=username, group=group, uid=uid,
|
||||||
homedir=homedir,
|
homedir=homedir,
|
||||||
|
@ -215,6 +227,7 @@ class User(TimeStampedModel, models.Model):
|
||||||
self.username, self.uid, self.group.gid, self.gecos,
|
self.username, self.uid, self.group.gid, self.gecos,
|
||||||
self.homedir, self.shell, password
|
self.homedir, self.shell, password
|
||||||
),
|
),
|
||||||
|
'create_ldap_user',
|
||||||
commit=True
|
commit=True
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -224,7 +237,8 @@ class User(TimeStampedModel, models.Model):
|
||||||
create_ldap_user.delay(
|
create_ldap_user.delay(
|
||||||
self.username, self.uid, self.group.gid, self.gecos,
|
self.username, self.uid, self.group.gid, self.gecos,
|
||||||
self.homedir, self.shell, password=None
|
self.homedir, self.shell, password=None
|
||||||
)
|
),
|
||||||
|
'create_ldap_user'
|
||||||
)
|
)
|
||||||
return super(User, self).save(*args, **kwargs)
|
return super(User, self).save(*args, **kwargs)
|
||||||
|
|
||||||
|
@ -236,15 +250,18 @@ class User(TimeStampedModel, models.Model):
|
||||||
'usergroup',
|
'usergroup',
|
||||||
'{0} in {1}'.format(self.username, group.groupname),
|
'{0} in {1}'.format(self.username, group.groupname),
|
||||||
remove_ldap_user_from_group.delay(
|
remove_ldap_user_from_group.delay(
|
||||||
self.username, group.groupname)
|
self.username, group.groupname),
|
||||||
|
'remove_ldap_user_from_group',
|
||||||
)
|
)
|
||||||
DeleteTaskResult.objects.create_deletetaskresult(
|
DeleteTaskResult.objects.create_deletetaskresult(
|
||||||
'user', self.username,
|
'user', self.username,
|
||||||
delete_ldap_user.delay(self.username)
|
delete_ldap_user.delay(self.username),
|
||||||
|
'delete_ldap_user'
|
||||||
)
|
)
|
||||||
DeleteTaskResult.objects.create_deletetaskresult(
|
DeleteTaskResult.objects.create_deletetaskresult(
|
||||||
'group', self.group.groupname,
|
'group', self.group.groupname,
|
||||||
delete_ldap_group_if_empty.delay(self.group.groupname)
|
delete_ldap_group_if_empty.delay(self.group.groupname),
|
||||||
|
'delete_ldap_group_if_empty'
|
||||||
)
|
)
|
||||||
self.group.delete()
|
self.group.delete()
|
||||||
super(User, self).delete(*args, **kwargs)
|
super(User, self).delete(*args, **kwargs)
|
||||||
|
@ -252,8 +269,10 @@ class User(TimeStampedModel, models.Model):
|
||||||
|
|
||||||
class UserTaskResultManager(TaskResultManager):
|
class UserTaskResultManager(TaskResultManager):
|
||||||
|
|
||||||
def create_usertaskresult(self, user, asyncresult, commit=False):
|
def create_usertaskresult(
|
||||||
taskresult = self.create(asyncresult)
|
self, user, asyncresult, task_name, commit=False
|
||||||
|
):
|
||||||
|
taskresult = self.create(asyncresult, task_name)
|
||||||
taskresult.user = user
|
taskresult.user = user
|
||||||
taskresult.save()
|
taskresult.save()
|
||||||
return taskresult
|
return taskresult
|
||||||
|
@ -336,15 +355,14 @@ class AdditionalGroup(TimeStampedModel, models.Model):
|
||||||
|
|
||||||
def clean(self):
|
def clean(self):
|
||||||
if self.user.group == self.group:
|
if self.user.group == self.group:
|
||||||
raise ValidationError(_(
|
raise ValidationError(CANNOT_USE_PRIMARY_GROUP_AS_ADDITIONAL)
|
||||||
"You can not use a user's primary group."))
|
|
||||||
|
|
||||||
def save(self, *args, **kwargs):
|
def save(self, *args, **kwargs):
|
||||||
res = add_ldap_user_to_group.delay(
|
GroupTaskResult.objects.create_grouptaskresult(
|
||||||
self.user.username, self.group.groupname)
|
self.group,
|
||||||
GroupTaskResult.objects.create(
|
add_ldap_user_to_group.delay(
|
||||||
group=self.group, task_uuid=res.id,
|
self.user.username, self.group.groupname),
|
||||||
task_name=res.task_name
|
'add_ldap_user_to_group'
|
||||||
)
|
)
|
||||||
super(AdditionalGroup, self).save(*args, **kwargs)
|
super(AdditionalGroup, self).save(*args, **kwargs)
|
||||||
|
|
||||||
|
@ -353,7 +371,8 @@ class AdditionalGroup(TimeStampedModel, models.Model):
|
||||||
'usergroup',
|
'usergroup',
|
||||||
str(self),
|
str(self),
|
||||||
remove_ldap_user_from_group.delay(
|
remove_ldap_user_from_group.delay(
|
||||||
self.user.username, self.group.groupname)
|
self.user.username, self.group.groupname),
|
||||||
|
'remove_ldap_user_from_group'
|
||||||
)
|
)
|
||||||
super(AdditionalGroup, self).delete(*args, **kwargs)
|
super(AdditionalGroup, self).delete(*args, **kwargs)
|
||||||
|
|
||||||
|
|
|
@ -1,3 +0,0 @@
|
||||||
from django.test import TestCase
|
|
||||||
|
|
||||||
# Create your tests here.
|
|
0
gnuviechadmin/osusers/tests/__init__.py
Normal file
0
gnuviechadmin/osusers/tests/__init__.py
Normal file
54
gnuviechadmin/osusers/tests/test_models.py
Normal file
54
gnuviechadmin/osusers/tests/test_models.py
Normal file
|
@ -0,0 +1,54 @@
|
||||||
|
from django.core.exceptions import ValidationError
|
||||||
|
from django.test import TestCase
|
||||||
|
from django.test.utils import override_settings
|
||||||
|
|
||||||
|
from osusers.models import (
|
||||||
|
CANNOT_USE_PRIMARY_GROUP_AS_ADDITIONAL,
|
||||||
|
AdditionalGroup,
|
||||||
|
Group,
|
||||||
|
Shadow,
|
||||||
|
User,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class AdditionalGroupTest(TestCase):
|
||||||
|
@override_settings(
|
||||||
|
CELERY_ALWAYS_EAGER=True,
|
||||||
|
CELERY_CACHE_BACKEND='memory',
|
||||||
|
BROKER_BACKEND='memory'
|
||||||
|
)
|
||||||
|
def test_clean_primary_group(self):
|
||||||
|
group1 = Group.objects.create(groupname='test1', gid=1000)
|
||||||
|
user = User.objects.create(
|
||||||
|
username='test', uid=1000, group=group1,
|
||||||
|
homedir='/home/test', shell='/bin/bash')
|
||||||
|
testsubj = AdditionalGroup(user=user, group=group1)
|
||||||
|
with self.assertRaises(ValidationError) as cm:
|
||||||
|
testsubj.clean()
|
||||||
|
self.assertEqual(
|
||||||
|
cm.exception.message, CANNOT_USE_PRIMARY_GROUP_AS_ADDITIONAL)
|
||||||
|
|
||||||
|
@override_settings(
|
||||||
|
CELERY_ALWAYS_EAGER=True,
|
||||||
|
CELERY_CACHE_BACKEND='memory',
|
||||||
|
BROKER_BACKEND='memory'
|
||||||
|
)
|
||||||
|
def test_clean_other_group(self):
|
||||||
|
group1 = Group(groupname='test1', gid=1000)
|
||||||
|
group2 = Group(groupname='test2', gid=1001)
|
||||||
|
user = User(username='test', uid=1000, group=group1,
|
||||||
|
homedir='/home/test', shell='/bin/bash')
|
||||||
|
testsubj = AdditionalGroup(user=user, group=group2)
|
||||||
|
testsubj.clean()
|
||||||
|
|
||||||
|
|
||||||
|
class GroupTest(TestCase):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class ShadowTest(TestCase):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class UserTest(TestCase):
|
||||||
|
pass
|
Loading…
Reference in a new issue