diff --git a/openwisp_notifications/handlers.py b/openwisp_notifications/handlers.py index 6949ab28..a0f80f35 100644 --- a/openwisp_notifications/handlers.py +++ b/openwisp_notifications/handlers.py @@ -7,7 +7,7 @@ from django.db import transaction from django.db.models import Q from django.db.models.query import QuerySet -from django.db.models.signals import post_delete, post_save +from django.db.models.signals import post_delete, post_save, pre_save from django.dispatch import receiver from django.utils import timezone from django.utils.translation import gettext as _ @@ -300,13 +300,45 @@ def notification_setting_delete_org_user(instance, **kwargs): ) -@receiver(post_save, sender=User, dispatch_uid='user_notification_setting') -def update_superuser_notification_settings(instance, created, **kwargs): - transaction.on_commit( - lambda: tasks.update_superuser_notification_settings.delay( - instance.pk, instance.is_superuser, created +@receiver(pre_save, sender=User, dispatch_uid='superuser_demoted_notification_setting') +def update_superuser_notification_settings(instance, update_fields, **kwargs): + """ + If user is demoted from superuser status, then + remove notification settings for non-managed organizations. + + If user is promoted to superuser, then + create notification settings for all organizations. + """ + if update_fields is not None and 'is_superuser' not in update_fields: + # No-op if is_superuser field is not being updated. + # If update_fields is None, it means any field could be updated. + return + try: + db_instance = User.objects.only('is_superuser').get(pk=instance.pk) + except User.DoesNotExist: + # User is being created + return + # If user is demoted from superuser to non-superuser + if db_instance.is_superuser and not instance.is_superuser: + transaction.on_commit( + lambda: tasks.update_superuser_notification_settings.delay( + instance.pk, is_superuser=False + ) + ) + elif not db_instance.is_superuser and instance.is_superuser: + transaction.on_commit( + lambda: tasks.update_superuser_notification_settings.delay( + instance.pk, is_superuser=True + ) + ) + + +@receiver(post_save, sender=User, dispatch_uid='create_superuser_notification_settings') +def create_superuser_notification_settings(instance, created, **kwargs): + if created and instance.is_superuser: + transaction.on_commit( + lambda: tasks.create_superuser_notification_settings.delay(instance.pk) ) - ) @receiver( diff --git a/openwisp_notifications/tasks.py b/openwisp_notifications/tasks.py index 9869d46a..94a71fb4 100644 --- a/openwisp_notifications/tasks.py +++ b/openwisp_notifications/tasks.py @@ -83,24 +83,23 @@ def create_notification_settings(user, organizations, notification_types): @shared_task(base=OpenwispCeleryTask) -def update_superuser_notification_settings(instance_id, is_superuser, is_created): +def update_superuser_notification_settings(instance_id, is_superuser, is_created=False): """ - Adds notification setting for all notification types and organizations. - If a superuser gets demoted, flags it's notification settings as deleted. + NOTE: This task is deprecated and only kept for backward compatibility. + It will be removed in 1.2.0 release. """ - user = User.objects.get(pk=instance_id) - - # When a user is demoted from superuser status, - # only keep notification settings for organization they are member of. - if not (is_superuser or is_created): - NotificationSetting.objects.filter(user_id=instance_id).exclude( - organization__in=user.organizations_managed - ).update(deleted=True) - return - if not is_superuser: - return + superuser_demoted_notification_setting(instance_id) + else: + create_superuser_notification_settings(instance_id) + +@shared_task(base=OpenwispCeleryTask) +def create_superuser_notification_settings(user_id): + """ + Adds notification setting for all notification types and organizations. + """ + user = User.objects.get(pk=user_id) # Create notification settings for superuser create_notification_settings( user=user, @@ -109,6 +108,18 @@ def update_superuser_notification_settings(instance_id, is_superuser, is_created ) +@shared_task(base=OpenwispCeleryTask) +def superuser_demoted_notification_setting(user_id): + """ + Flags NotificationSettings as deleted for non-managed organizations + when a superuser is demoted to a non-superuser. + """ + user = User.objects.get(pk=user_id) + NotificationSetting.objects.filter(user_id=user_id).exclude( + organization__in=user.organizations_managed + ).update(deleted=True) + + @shared_task(base=OpenwispCeleryTask) def ns_register_unregister_notification_type( notification_type=None, delete_unregistered=True diff --git a/openwisp_notifications/tests/test_notification_setting.py b/openwisp_notifications/tests/test_notification_setting.py index 254ddb9f..621e2dc5 100644 --- a/openwisp_notifications/tests/test_notification_setting.py +++ b/openwisp_notifications/tests/test_notification_setting.py @@ -1,6 +1,9 @@ +from unittest.mock import patch + from django.db.models.signals import post_save from django.test import TransactionTestCase +from openwisp_notifications import tasks from openwisp_notifications.handlers import ( notification_type_registered_unregistered_handler, ) @@ -251,3 +254,37 @@ def test_deleted_notificationsetting_autocreated(self): self.assertEqual(ns_queryset.count(), 1) ns.refresh_from_db() self.assertEqual(ns.deleted, False) + + @patch.object(tasks, 'superuser_demoted_notification_setting') + @patch.object(tasks, 'create_superuser_notification_settings') + def test_task_not_called_on_user_login(self, created_mock, demoted_mock): + admin = self._create_admin() + org_user = self._create_staff_org_admin() + created_mock.delay.assert_called_once() + + created_mock.reset_mock() + with self.subTest('Test task not called if superuser status is unchanged'): + admin.username = 'new_admin' + admin.save() + created_mock.assert_not_called() + demoted_mock.assert_not_called() + + with self.subTest('Test task not called when superuser logs in'): + self.client.force_login(admin) + created_mock.assert_not_called() + demoted_mock.assert_not_called() + + with self.subTest('Test task not called when org user logs in'): + self.client.force_login(org_user.user) + created_mock.assert_not_called() + demoted_mock.assert_not_called() + + with self.subTest('Test task called when superuser status changed'): + admin.is_superuser = False + admin.save() + demoted_mock.assert_called_once() + created_mock.assert_not_called() + + admin.is_superuser = True + admin.save() + created_mock.assert_called_once()