From bd0cd34f145c5c979c68ceebe54f7d98783910aa Mon Sep 17 00:00:00 2001 From: Cecilia Lau Date: Wed, 13 May 2026 15:59:01 -0400 Subject: [PATCH 1/3] add send_notifications util --- generic_notifications/__init__.py | 50 +++++++- tests/test_utils.py | 186 +++++++++++++++++++++++++++++- 2 files changed, 232 insertions(+), 4 deletions(-) diff --git a/generic_notifications/__init__.py b/generic_notifications/__init__.py index 3116556..dc2b5e2 100644 --- a/generic_notifications/__init__.py +++ b/generic_notifications/__init__.py @@ -1,5 +1,5 @@ import logging -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Iterable from django.db import transaction @@ -99,3 +99,51 @@ def send_notification( logger.error(f"Channel {channel_cls.key} failed to process notification {notification.id}: {e}") return notification + + +def send_notifications( + recipients: Iterable[Any], + notification_type: "type[NotificationType]", + actor: Any | None = None, + target: Any | None = None, + subject: str = "", + text: str = "", + url: str = "", + metadata: dict[str, Any] | None = None, + **kwargs, +): + """ + Send notifications multiple users through all registered channels. + + Args: + recipients: Users to send notification to + notification_type: Type of notification (must be registered in registry) + actor: User who triggered the notification (optional) + target: Object the notification is about (optional) + subject: Notification subject line (optional, can be generated by notification type) + text: Notification body text (optional, can be generated by notification type) + url: URL associated with the notification (optional) + metadata: Additional JSON-serializable data to store with the notification (optional) + **kwargs: Any additional fields for the notification model + + Returns: + List of Notification instances or None if channels are not enabled for that notification + + Raises: + ValueError: If notification_type is not registered + """ + notifications = [ + send_notification( + recipient, + notification_type, + actor, + target, + subject, + text, + url, + metadata, + **kwargs, + ) + for recipient in recipients + ] + return notifications diff --git a/tests/test_utils.py b/tests/test_utils.py index 517c212..8b336cd 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -4,10 +4,10 @@ from django.core import mail from django.test import TestCase, override_settings -from generic_notifications import send_notification +from generic_notifications import send_notification, send_notifications from generic_notifications.channels import EmailChannel, WebsiteChannel from generic_notifications.frequencies import RealtimeFrequency -from generic_notifications.models import Notification +from generic_notifications.models import Notification, NotificationChannel from generic_notifications.registry import registry from generic_notifications.types import NotificationType from generic_notifications.utils import get_notifications, get_unread_count, mark_notifications_as_read @@ -87,7 +87,6 @@ def test_send_notification_basic(self): self.assertIsNone(notification.target) # Verify both channels were created - from generic_notifications.models import NotificationChannel channels = NotificationChannel.objects.filter(notification=notification) self.assertEqual(channels.count(), 2) @@ -206,6 +205,187 @@ def test_send_notification_without_email_no_mail_sent(self): self.assertEqual(len(mail.outbox), 0) +class SendNotificationsTest(TestCase): + def setUp(self): + self.user1 = User.objects.create_user(username="test1", email="test1@example.com", password="testpass") + self.user2 = User.objects.create_user(username="test2", email="test2@example.com", password="testpass") + self.actor = User.objects.create_user(username="actor", email="actor@example.com", password="testpass") + self.recipients = [self.user1, self.user2] + + # Register test notification type + self.notification_type = TestNotificationType + registry.register_type(TestNotificationType) + + def tearDown(self): + mail.outbox.clear() + + @override_settings(DEFAULT_FROM_EMAIL="test@example.com") + def test_send_notifications_basic(self): + notifications = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + subject="Test Subject", + text="Test message", + ) + + self.assertIsInstance(notifications[0], Notification) + self.assertEqual(notifications[0].recipient, self.user1) + self.assertEqual(notifications[0].notification_type, "test_type") + self.assertEqual(notifications[0].subject, "Test Subject") + self.assertEqual(notifications[0].text, "Test message") + self.assertIsNone(notifications[0].actor) + self.assertIsNone(notifications[0].target) + self.assertIsInstance(notifications[1], Notification) + self.assertEqual(notifications[1].recipient, self.user2) + self.assertEqual(notifications[1].notification_type, "test_type") + self.assertEqual(notifications[1].subject, "Test Subject") + self.assertEqual(notifications[1].text, "Test message") + self.assertIsNone(notifications[1].actor) + self.assertIsNone(notifications[1].target) + + # Verify both channels were created + channels = NotificationChannel.objects.filter(notification=notifications[0]) + self.assertEqual(channels.count(), 2) + channel_names = [c.channel for c in channels] + self.assertIn("website", channel_names) + self.assertIn("email", channel_names) + channels = NotificationChannel.objects.filter(notification=notifications[1]) + self.assertEqual(channels.count(), 2) + channel_names = [c.channel for c in channels] + self.assertIn("website", channel_names) + self.assertIn("email", channel_names) + + # Verify emails were sent + self.assertEqual(len(mail.outbox), 2) + email = mail.outbox[0] + self.assertEqual(email.to, [self.user1.email]) + self.assertIn("Test Subject", email.subject) + email = mail.outbox[1] + self.assertEqual(email.to, [self.user2.email]) + self.assertIn("Test Subject", email.subject) + + def test_send_notifications_with_actor_and_target(self): + notifications = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + actor=self.actor, + target=self.actor, # Using actor as target for simplicity + url="/test/url", + metadata={"key": "value"}, + ) + + self.assertEqual(notifications[0].actor, self.actor) + self.assertEqual(notifications[0].target, self.actor) + self.assertEqual(notifications[0].url, "/test/url") + self.assertEqual(notifications[0].metadata, {"key": "value"}) + self.assertEqual(notifications[1].actor, self.actor) + self.assertEqual(notifications[1].target, self.actor) + self.assertEqual(notifications[1].url, "/test/url") + self.assertEqual(notifications[1].metadata, {"key": "value"}) + + def test_send_notifications_invalid_type(self): + class InvalidNotificationType(NotificationType): + key = "invalid_type" + name = "Invalid Type" + description = "" + + invalid_type = InvalidNotificationType + with self.assertRaises(ValueError) as cm: + send_notifications(recipients=self.recipients, notification_type=invalid_type) + + self.assertIn("not registered", str(cm.exception)) + + @override_settings(DEFAULT_FROM_EMAIL="test@example.com") + def test_send_notifications_with_disabled_channel(self): + # Disable website channel for one user + self.notification_type.disable_channel(self.user1, WebsiteChannel) + + notifications = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + subject="Test Subject", + text="Test message", + ) + + # Notification should only have email channel for user1 + channel_keys = notifications[0].get_channels() + self.assertNotIn("website", channel_keys) + self.assertIn("email", channel_keys) + self.assertEqual(len(channel_keys), 1) + + # Notification should both channels for user2 + channel_keys = notifications[1].get_channels() + self.assertIn("website", channel_keys) + self.assertIn("email", channel_keys) + self.assertEqual(len(channel_keys), 2) + + # Verify emails was sent + self.assertEqual(len(mail.outbox), 2) + + def test_send_notifications_with_all_channels_disabled(self): + # Disable both channels for user1 + self.notification_type.disable_channel(self.user1, WebsiteChannel) + self.notification_type.disable_channel(self.user1, EmailChannel) + + notifications = send_notifications(recipients=self.recipients, notification_type=self.notification_type) + + # Should return None when no channels are enabled + self.assertIsNone(notifications[0]) + self.assertIsNotNone(notifications[1]) + + def test_send_notifications_with_forbidden_channels(self): + # Create a notification type with forbidden channels + class ForbiddenTestType(NotificationType): + key = "forbidden_test" + name = "Forbidden Test" + forbidden_channels = [WebsiteChannel] + + registry.register_type(ForbiddenTestType) + + notifications = send_notifications(recipients=self.recipients, notification_type=ForbiddenTestType) + + # All notifications should only have email channel (website is forbidden) + for notification in notifications: + self.assertIsNotNone(notification) + channel_keys = notification.get_channels() + self.assertNotIn("website", channel_keys) + self.assertIn("email", channel_keys) + self.assertEqual(len(channel_keys), 1) + + def test_send_notifications_without_email_no_mail_sent(self): + """Test that no email is sent when recipient has no email address""" + # Create user without email + user_no_email = User.objects.create_user(username="nomail", password="testpass") + user_no_email.email = "" + user_no_email.save() + + # Send notifications + notifications = send_notifications( + recipients=[user_no_email, self.user1], + notification_type=self.notification_type, + subject="Test Subject", + text="Test message", + ) + + # Verify notification was created + self.assertIsNotNone(notifications[0]) + channels = NotificationChannel.objects.filter(notification=notifications[0]) + self.assertEqual(channels.count(), 1) + self.assertEqual(channels.first().channel, "website") + self.assertIsNotNone(notifications[1]) + channels = NotificationChannel.objects.filter(notification=notifications[1]) + self.assertEqual(channels.count(), 2) + channel_names = [c.channel for c in channels] + self.assertIn("website", channel_names) + self.assertIn("email", channel_names) + + # Verify only one email was sent + self.assertEqual(len(mail.outbox), 1) + email = mail.outbox[0] + self.assertEqual(email.to, [self.user1.email]) + self.assertIn("Test Subject", email.subject) + + class MarkNotificationsAsReadTest(TestCase): def setUp(self): self.user = User.objects.create_user(username="test", email="test@example.com", password="testpass") From f4f8d346fe1caf8649f642296d2d77db8ccf632f Mon Sep 17 00:00:00 2001 From: Cecilia Lau Date: Wed, 13 May 2026 16:34:40 -0400 Subject: [PATCH 2/3] fix typos/docs --- generic_notifications/__init__.py | 27 ++++++++++++++------------- tests/test_utils.py | 1 - 2 files changed, 14 insertions(+), 14 deletions(-) diff --git a/generic_notifications/__init__.py b/generic_notifications/__init__.py index dc2b5e2..fe76b59 100644 --- a/generic_notifications/__init__.py +++ b/generic_notifications/__init__.py @@ -1,9 +1,10 @@ import logging -from typing import TYPE_CHECKING, Any, Iterable +from typing import TYPE_CHECKING, Any, Iterable, List from django.db import transaction if TYPE_CHECKING: + from .models import Notification from .types import NotificationType @@ -17,7 +18,7 @@ def send_notification( url: str = "", metadata: dict[str, Any] | None = None, **kwargs, -): +) -> Notification | None: """ Send a notification to a user through all registered channels. @@ -111,9 +112,9 @@ def send_notifications( url: str = "", metadata: dict[str, Any] | None = None, **kwargs, -): +) -> List[Notification | None]: """ - Send notifications multiple users through all registered channels. + Send notifications to multiple users through all registered channels. Args: recipients: Users to send notification to @@ -127,21 +128,21 @@ def send_notifications( **kwargs: Any additional fields for the notification model Returns: - List of Notification instances or None if channels are not enabled for that notification + List of Notification instances Raises: ValueError: If notification_type is not registered """ notifications = [ send_notification( - recipient, - notification_type, - actor, - target, - subject, - text, - url, - metadata, + recipient=recipient, + notification_type=notification_type, + actor=actor, + target=target, + subject=subject, + text=text, + url=url, + metadata=metadata, **kwargs, ) for recipient in recipients diff --git a/tests/test_utils.py b/tests/test_utils.py index 8b336cd..024a4fb 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -87,7 +87,6 @@ def test_send_notification_basic(self): self.assertIsNone(notification.target) # Verify both channels were created - channels = NotificationChannel.objects.filter(notification=notification) self.assertEqual(channels.count(), 2) channel_names = [c.channel for c in channels] From 09cc4ff131d224984fea508a6ead7da13779ba16 Mon Sep 17 00:00:00 2001 From: Cecilia Lau Date: Thu, 14 May 2026 11:50:45 -0400 Subject: [PATCH 3/3] return number of notifications saved from send_notifications() --- generic_notifications/__init__.py | 17 ++-- tests/test_utils.py | 153 ++++++++++++++++++++---------- 2 files changed, 113 insertions(+), 57 deletions(-) diff --git a/generic_notifications/__init__.py b/generic_notifications/__init__.py index fe76b59..ee80822 100644 --- a/generic_notifications/__init__.py +++ b/generic_notifications/__init__.py @@ -1,5 +1,5 @@ import logging -from typing import TYPE_CHECKING, Any, Iterable, List +from typing import TYPE_CHECKING, Any, Iterable from django.db import transaction @@ -112,7 +112,7 @@ def send_notifications( url: str = "", metadata: dict[str, Any] | None = None, **kwargs, -) -> List[Notification | None]: +) -> int: """ Send notifications to multiple users through all registered channels. @@ -128,13 +128,14 @@ def send_notifications( **kwargs: Any additional fields for the notification model Returns: - List of Notification instances + Number of notifications that were sent successfully Raises: ValueError: If notification_type is not registered """ - notifications = [ - send_notification( + success_counter = 0 + for recipient in recipients: + notification = send_notification( recipient=recipient, notification_type=notification_type, actor=actor, @@ -145,6 +146,6 @@ def send_notifications( metadata=metadata, **kwargs, ) - for recipient in recipients - ] - return notifications + if notification is not None: + success_counter += 1 + return success_counter diff --git a/tests/test_utils.py b/tests/test_utils.py index 024a4fb..415fad2 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -220,29 +220,28 @@ def tearDown(self): @override_settings(DEFAULT_FROM_EMAIL="test@example.com") def test_send_notifications_basic(self): - notifications = send_notifications( + notification_count = send_notifications( recipients=self.recipients, notification_type=self.notification_type, subject="Test Subject", text="Test message", ) - self.assertIsInstance(notifications[0], Notification) - self.assertEqual(notifications[0].recipient, self.user1) - self.assertEqual(notifications[0].notification_type, "test_type") - self.assertEqual(notifications[0].subject, "Test Subject") - self.assertEqual(notifications[0].text, "Test message") - self.assertIsNone(notifications[0].actor) + # Verify both notifications were created + self.assertEqual(notification_count, 2) + + notifications = Notification.objects.filter( + recipient__in=self.recipients, + notification_type=self.notification_type.key, + subject="Test Subject", + text="Test message", + actor=None, + ) + self.assertEqual(notifications.count(), 2) self.assertIsNone(notifications[0].target) - self.assertIsInstance(notifications[1], Notification) - self.assertEqual(notifications[1].recipient, self.user2) - self.assertEqual(notifications[1].notification_type, "test_type") - self.assertEqual(notifications[1].subject, "Test Subject") - self.assertEqual(notifications[1].text, "Test message") - self.assertIsNone(notifications[1].actor) self.assertIsNone(notifications[1].target) - # Verify both channels were created + # Verify both channels were created for each notification channels = NotificationChannel.objects.filter(notification=notifications[0]) self.assertEqual(channels.count(), 2) channel_names = [c.channel for c in channels] @@ -264,23 +263,30 @@ def test_send_notifications_basic(self): self.assertIn("Test Subject", email.subject) def test_send_notifications_with_actor_and_target(self): - notifications = send_notifications( + url = "/test/url" + metadata = {"key": "value"} + notification_count = send_notifications( recipients=self.recipients, notification_type=self.notification_type, actor=self.actor, target=self.actor, # Using actor as target for simplicity - url="/test/url", - metadata={"key": "value"}, + url=url, + metadata=metadata, ) - self.assertEqual(notifications[0].actor, self.actor) + # Verify both notifications were created + self.assertEqual(notification_count, 2) + + notifications = Notification.objects.filter( + recipient__in=self.recipients, + notification_type=self.notification_type.key, + actor=self.actor, + url=url, + metadata=metadata, + ) + self.assertEqual(notifications.count(), 2) self.assertEqual(notifications[0].target, self.actor) - self.assertEqual(notifications[0].url, "/test/url") - self.assertEqual(notifications[0].metadata, {"key": "value"}) - self.assertEqual(notifications[1].actor, self.actor) self.assertEqual(notifications[1].target, self.actor) - self.assertEqual(notifications[1].url, "/test/url") - self.assertEqual(notifications[1].metadata, {"key": "value"}) def test_send_notifications_invalid_type(self): class InvalidNotificationType(NotificationType): @@ -296,24 +302,40 @@ class InvalidNotificationType(NotificationType): @override_settings(DEFAULT_FROM_EMAIL="test@example.com") def test_send_notifications_with_disabled_channel(self): + subject = "Test Subject" + text = "Test message" + # Disable website channel for one user self.notification_type.disable_channel(self.user1, WebsiteChannel) - notifications = send_notifications( + notification_count = send_notifications( recipients=self.recipients, notification_type=self.notification_type, - subject="Test Subject", - text="Test message", + subject=subject, + text=text, ) + # Verify both notifications were created + self.assertEqual(notification_count, 2) + + notifications = Notification.objects.filter( + recipient__in=self.recipients, + notification_type=self.notification_type.key, + subject=subject, + text=text, + ) + self.assertEqual(notifications.count(), 2) + # Notification should only have email channel for user1 - channel_keys = notifications[0].get_channels() + notification = notifications.get(recipient=self.user1) + channel_keys = notification.get_channels() self.assertNotIn("website", channel_keys) self.assertIn("email", channel_keys) self.assertEqual(len(channel_keys), 1) # Notification should both channels for user2 - channel_keys = notifications[1].get_channels() + notification = notifications.get(recipient=self.user2) + channel_keys = notification.get_channels() self.assertIn("website", channel_keys) self.assertIn("email", channel_keys) self.assertEqual(len(channel_keys), 2) @@ -326,11 +348,19 @@ def test_send_notifications_with_all_channels_disabled(self): self.notification_type.disable_channel(self.user1, WebsiteChannel) self.notification_type.disable_channel(self.user1, EmailChannel) - notifications = send_notifications(recipients=self.recipients, notification_type=self.notification_type) + notification_count = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + ) - # Should return None when no channels are enabled - self.assertIsNone(notifications[0]) - self.assertIsNotNone(notifications[1]) + # Verify only one notification was created + self.assertEqual(notification_count, 1) + + notifications = Notification.objects.filter( + recipient__in=self.recipients, + notification_type=self.notification_type.key, + ) + self.assertEqual(notifications.count(), 1) def test_send_notifications_with_forbidden_channels(self): # Create a notification type with forbidden channels @@ -341,11 +371,22 @@ class ForbiddenTestType(NotificationType): registry.register_type(ForbiddenTestType) - notifications = send_notifications(recipients=self.recipients, notification_type=ForbiddenTestType) + notification_count = send_notifications( + recipients=self.recipients, + notification_type=ForbiddenTestType, + ) + + # Verify both notifications were created + self.assertEqual(notification_count, 2) + + notifications = Notification.objects.filter( + recipient__in=self.recipients, + notification_type=ForbiddenTestType.key, + ) + self.assertEqual(notifications.count(), 2) # All notifications should only have email channel (website is forbidden) for notification in notifications: - self.assertIsNotNone(notification) channel_keys = notification.get_channels() self.assertNotIn("website", channel_keys) self.assertIn("email", channel_keys) @@ -358,25 +399,39 @@ def test_send_notifications_without_email_no_mail_sent(self): user_no_email.email = "" user_no_email.save() + subject = "Test Subject" + text = "Test message" + recipients = [user_no_email, self.user1] + # Send notifications - notifications = send_notifications( - recipients=[user_no_email, self.user1], + notification_count = send_notifications( + recipients=recipients, notification_type=self.notification_type, - subject="Test Subject", - text="Test message", + subject=subject, + text=text, ) - # Verify notification was created - self.assertIsNotNone(notifications[0]) - channels = NotificationChannel.objects.filter(notification=notifications[0]) - self.assertEqual(channels.count(), 1) - self.assertEqual(channels.first().channel, "website") - self.assertIsNotNone(notifications[1]) - channels = NotificationChannel.objects.filter(notification=notifications[1]) - self.assertEqual(channels.count(), 2) - channel_names = [c.channel for c in channels] - self.assertIn("website", channel_names) - self.assertIn("email", channel_names) + # Verify both notifications were created + self.assertEqual(notification_count, 2) + + notifications = Notification.objects.filter( + recipient__in=recipients, + notification_type=self.notification_type.key, + subject=subject, + text=text, + ) + self.assertEqual(notifications.count(), 2) + + # user_no_email should only have website channel + channel_keys = notifications.get(recipient=user_no_email).get_channels() + self.assertNotIn("email", channel_keys) + self.assertIn("website", channel_keys) + self.assertEqual(len(channel_keys), 1) + # user1 should have both channels + channel_keys = notifications.get(recipient=self.user1).get_channels() + self.assertIn("website", channel_keys) + self.assertIn("email", channel_keys) + self.assertEqual(len(channel_keys), 2) # Verify only one email was sent self.assertEqual(len(mail.outbox), 1)