diff --git a/generic_notifications/__init__.py b/generic_notifications/__init__.py index 3116556..ee80822 100644 --- a/generic_notifications/__init__.py +++ b/generic_notifications/__init__.py @@ -1,9 +1,10 @@ import logging -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Iterable 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. @@ -99,3 +100,52 @@ 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, +) -> int: + """ + Send notifications to 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: + Number of notifications that were sent successfully + + Raises: + ValueError: If notification_type is not registered + """ + success_counter = 0 + for recipient in recipients: + notification = send_notification( + recipient=recipient, + notification_type=notification_type, + actor=actor, + target=target, + subject=subject, + text=text, + url=url, + metadata=metadata, + **kwargs, + ) + if notification is not None: + success_counter += 1 + return success_counter diff --git a/tests/test_utils.py b/tests/test_utils.py index 517c212..415fad2 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,8 +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) channel_names = [c.channel for c in channels] @@ -206,6 +204,242 @@ 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): + notification_count = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + subject="Test Subject", + text="Test message", + ) + + # 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.assertIsNone(notifications[1].target) + + # 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] + 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): + 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=url, + metadata=metadata, + ) + + # 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[1].target, self.actor) + + 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): + subject = "Test Subject" + text = "Test message" + + # Disable website channel for one user + self.notification_type.disable_channel(self.user1, WebsiteChannel) + + notification_count = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + 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 + 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 + 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) + + # 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) + + notification_count = send_notifications( + recipients=self.recipients, + notification_type=self.notification_type, + ) + + # 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 + class ForbiddenTestType(NotificationType): + key = "forbidden_test" + name = "Forbidden Test" + forbidden_channels = [WebsiteChannel] + + registry.register_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: + 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() + + subject = "Test Subject" + text = "Test message" + recipients = [user_no_email, self.user1] + + # Send notifications + notification_count = send_notifications( + recipients=recipients, + notification_type=self.notification_type, + subject=subject, + text=text, + ) + + # 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) + 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")