Add tests for signal

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()

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",)

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):

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