trigger tasks on SshPublicKey save and delete

- implement save and delete methods in osusers.models.SshPublicKey that trigger
  set_file_ssh_authorized_keys
- add new test methods to osusers.tests.test_models.SshPublicKeyTest
This commit is contained in:
Jan Dittberner 2015-01-31 23:00:33 +01:00
parent 0080fe7e78
commit 7dd4c78345
2 changed files with 41 additions and 0 deletions

View file

@ -37,6 +37,7 @@ from ldaptasks.tasks import (
from fileservertasks.tasks import ( from fileservertasks.tasks import (
delete_file_mail_userdir, delete_file_mail_userdir,
delete_file_sftp_userdir, delete_file_sftp_userdir,
set_file_ssh_authorized_keys,
setup_file_mail_userdir, setup_file_mail_userdir,
setup_file_sftp_userdir, setup_file_sftp_userdir,
) )
@ -579,3 +580,24 @@ class SshPublicKey(TimeStampedModel):
return "{algorithm} {data} {comment}".format( return "{algorithm} {data} {comment}".format(
algorithm=self.algorithm, data=self.data, comment=self.comment algorithm=self.algorithm, data=self.data, comment=self.comment
).strip() ).strip()
def save(self, **kwargs):
key = super(SshPublicKey, self).save(**kwargs)
TaskResult.objects.create_task_result(
set_file_ssh_authorized_keys.delay(
self.user.username, [
str(key) for key in
SshPublicKey.objects.filter(user=self.user)]),
'set_file_ssh_authorized_keys'
)
return key
def delete(self, **kwargs):
super(SshPublicKey, self).delete(**kwargs)
TaskResult.objects.create_task_result(
set_file_ssh_authorized_keys.delay(
self.user.username, [
str(key) for key in
SshPublicKey.objects.filter(user=self.user)]),
'set_file_ssh_authorized_keys'
)

View file

@ -469,6 +469,7 @@ class SshPublicKeyTest(TestCaseWithCeleryTasks):
super(SshPublicKeyTest, self).setUp() super(SshPublicKeyTest, self).setUp()
customer = Customer.objects.create_user('test') customer = Customer.objects.create_user('test')
self.user = User.objects.create_user(customer) self.user = User.objects.create_user(customer)
TaskResult.objects.all().delete()
def test__str__rfc4716(self): def test__str__rfc4716(self):
res = SshPublicKey.objects.create_ssh_public_key( res = SshPublicKey.objects.create_ssh_public_key(
@ -489,3 +490,21 @@ class SshPublicKeyTest(TestCaseWithCeleryTasks):
res = SshPublicKey.objects.create_ssh_public_key( res = SshPublicKey.objects.create_ssh_public_key(
self.user, EXAMPLE_KEY_4_OPENSSH) self.user, EXAMPLE_KEY_4_OPENSSH)
self.assertEqual(str(res), EXAMPLE_KEY_4_OPENSSH) self.assertEqual(str(res), EXAMPLE_KEY_4_OPENSSH)
def test_call_tasks_on_save(self):
SshPublicKey.objects.create_ssh_public_key(
self.user, EXAMPLE_KEY_4_OPENSSH)
taskresults = TaskResult.objects.all()
self.assertEqual(len(taskresults), 1)
self.assertEqual(
taskresults[0].task_name, 'set_file_ssh_authorized_keys')
def test_call_tasks_on_delete(self):
key = SshPublicKey.objects.create_ssh_public_key(
self.user, EXAMPLE_KEY_4_OPENSSH)
TaskResult.objects.all().delete()
key.delete()
taskresults = TaskResult.objects.all()
self.assertEqual(len(taskresults), 1)
self.assertEqual(
taskresults[0].task_name, 'set_file_ssh_authorized_keys')