diff --git a/notifications/apps.py b/notifications/apps.py index f00baa1..b5e0343 100644 --- a/notifications/apps.py +++ b/notifications/apps.py @@ -1,6 +1,7 @@ """ Django notifications apps file """ # -*- coding: utf-8 -*- from django.apps import AppConfig +from django.core.signals import setting_changed from django.utils.translation import gettext_lazy as _ @@ -10,10 +11,15 @@ class Config(AppConfig): default_auto_field = "django.db.models.AutoField" def ready(self) -> None: + from notifications.settings import ( # pylint: disable=import-outside-toplevel + reload_notification_settings, + ) from notifications.signals import ( # pylint: disable=import-outside-toplevel notify, notify_handler, ) notify.connect(notify_handler, dispatch_uid="notifications.models.notification") + setting_changed.connect(reload_notification_settings, dispatch_uid="notifications.models.notification") + return super().ready() diff --git a/notifications/settings.py b/notifications/settings.py index 8288ccd..2c80c5f 100644 --- a/notifications/settings.py +++ b/notifications/settings.py @@ -4,10 +4,6 @@ from typing import Any, TypedDict, Union from django.conf import settings -# Import from `django.core.signals` instead of the official location -# `django.test.signals` to avoid importing the test module unnecessarily. -from django.core.signals import setting_changed - NOTIFICATION_DEFAULTS = { "PAGINATE_BY": 20, "USE_JSONFIELD": False, @@ -86,6 +82,4 @@ def reload_notification_settings(*args: Any, **kwargs: Any): # pylint: disable= notification_settings.reload() -setting_changed.connect(reload_notification_settings) - __all__ = ("notification_settings",) diff --git a/notifications/signals.py b/notifications/signals.py index c14bbd5..165e8cd 100644 --- a/notifications/signals.py +++ b/notifications/signals.py @@ -11,22 +11,22 @@ from swapper import load_model from notifications.models.base import NotificationLevel from notifications.settings import notification_settings -EXTRA_DATA = notification_settings.USE_JSONFIELD +Notification = load_model("notifications", "Notification") -def notify_handler(verb, **kwargs): +def notify_handler(sender, **kwargs): """ Handler function to create Notification instance upon action signal call. """ # Pull the options out of kwargs kwargs.pop("signal", None) + actor = sender recipient = kwargs.pop("recipient") - actor = kwargs.pop("sender") + verb = kwargs.pop("verb") optional_objs = [(kwargs.pop(opt, None), opt) for opt in ("target", "action_object")] public = bool(kwargs.pop("public", True)) description = kwargs.pop("description", None) timestamp = kwargs.pop("timestamp", timezone.now()) - Notification = load_model("notifications", "Notification") # pylint: disable=invalid-name level = kwargs.pop("level", NotificationLevel.INFO) actor_for_concrete_model = kwargs.pop("actor_for_concrete_model", True) @@ -63,7 +63,7 @@ def notify_handler(verb, **kwargs): ContentType.objects.get_for_model(obj, for_concrete_model=for_concrete_model), ) - if kwargs and EXTRA_DATA: + if kwargs and notification_settings.USE_JSONFIELD: # set kwargs as model column if available for key in list(kwargs.keys()): if hasattr(newnotify, key): diff --git a/notifications/tests/test_signals.py b/notifications/tests/test_signals.py new file mode 100644 index 0000000..45255c2 --- /dev/null +++ b/notifications/tests/test_signals.py @@ -0,0 +1,127 @@ +import pytest +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Group +from django.test import override_settings # noqa +from django.utils.timezone import now +from swapper import load_model + +from notifications.models.base import NotificationLevel +from notifications.signals import notify +from notifications.tests.factories.users import ( + ActorFactory, + RecipientFactory, + TargetFactory, +) + +Notification = load_model("notifications", "Notification") +User = get_user_model() + + +@pytest.mark.django_db +def test_send(): + actor = ActorFactory() + recipient = RecipientFactory() + target = TargetFactory() + action_object = TargetFactory() + timestamp = now() + notify.send( + sender=actor, + recipient=recipient, + verb="poked", + public=False, + description="Testing", + timestamp=timestamp, + level=NotificationLevel.ERROR, + target=target, + action_object=action_object, + ) + + assert Notification.objects.count() == 1 + notification = Notification.objects.first() + assert notification.actor == actor + assert notification.recipient == recipient + assert notification.verb == "poked" + assert notification.public is False + assert notification.description == "Testing" + assert notification.timestamp == timestamp + assert notification.level == NotificationLevel.ERROR + assert notification.target == target + assert notification.action_object == action_object + + +@pytest.mark.django_db +def test_send_to_multiple_recipients(): + actor = ActorFactory() + group = Group.objects.create(name="group1") + recipient1 = RecipientFactory() + recipient2 = RecipientFactory() + recipient1.groups.add(group) + + notify.send(sender=actor, recipient=[recipient1, recipient2], verb="poked you") + assert Notification.objects.filter(actor_object_id=actor.id, recipient=recipient1, verb="poked you").count() == 1 + assert Notification.objects.filter(actor_object_id=actor.id, recipient=recipient2, verb="poked you").count() == 1 + + recipients = User.objects.filter(username__startswith="recipient") + notify.send(sender=actor, recipient=recipients, verb="poked you") + assert Notification.objects.filter(actor_object_id=actor.id, recipient=recipient1, verb="poked you").count() == 2 + assert Notification.objects.filter(actor_object_id=actor.id, recipient=recipient2, verb="poked you").count() == 2 + + notify.send(sender=actor, recipient=group, verb="poked you") + assert Notification.objects.filter(actor_object_id=actor.id, recipient=recipient1, verb="poked you").count() == 3 + assert Notification.objects.filter(actor_object_id=actor.id, recipient=recipient2, verb="poked you").count() == 2 + + +@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"USE_JSONFIELD": True}) +@pytest.mark.django_db +def test_extra_data(): + actor = ActorFactory() + recipient = RecipientFactory() + target = TargetFactory() + action_object = TargetFactory() + timestamp = now() + notify.send( + sender=actor, + recipient=recipient, + verb="poked", + public=False, + description="Testing", + timestamp=timestamp, + level=NotificationLevel.ERROR, + target=target, + action_object=action_object, + extra1=True, + extra2="Hello", + extra3=3.1415, + extra4=[1, 2], + extra5={1: 2, "bla": True}, + ) + + assert Notification.objects.count() == 1 + notification = Notification.objects.first() + assert notification.data["extra1"] is True + assert notification.data["extra2"] == "Hello" + assert notification.data["extra3"] == 3.1415 + assert notification.data["extra4"] == [1, 2] + assert notification.data["extra5"]["1"] == 2 + assert notification.data["extra5"]["bla"] is True + + +@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"USE_JSONFIELD": False}) +@pytest.mark.django_db +def test_extra_data_disabled(): + actor = ActorFactory() + recipient = RecipientFactory() + notify.send( + sender=actor, + recipient=recipient, + verb="poked", + extra1=True, + extra2="Hello", + extra3=3.1415, + extra4=[1, 2], + extra5={1: 2, "bla": True}, + ) + + assert Notification.objects.count() == 1 + notification = Notification.objects.first() + assert notification.data is None