|
16 | 16 | from django.test import TransactionTestCase |
17 | 17 | from django.urls import reverse |
18 | 18 | from django.utils import timezone |
| 19 | +from django.utils.html import strip_spaces_between_tags |
19 | 20 | from django.utils.timesince import timesince |
20 | 21 | from freezegun import freeze_time |
21 | 22 |
|
|
42 | 43 | from openwisp_users.tests.utils import TestOrganizationMixin |
43 | 44 | from openwisp_utils.tests import capture_any_output |
44 | 45 |
|
| 46 | +from . import ( |
| 47 | + _test_batch_email_notification_email_body, |
| 48 | + _test_batch_email_notification_email_html, |
| 49 | +) |
| 50 | + |
45 | 51 | User = get_user_model() |
46 | 52 |
|
47 | 53 | Notification = load_model('Notification') |
@@ -136,10 +142,16 @@ def test_superuser_notifications_disabled(self): |
136 | 142 | organization_id=target_obj.organization.pk, |
137 | 143 | type='default', |
138 | 144 | ) |
139 | | - self.assertTrue(notification_preference.email) |
| 145 | + # Database field is set to None |
| 146 | + self.assertEqual(notification_preference.email, None) |
| 147 | + # The fallback is taked from notification type |
| 148 | + self.assertTrue(notification_preference.email_notification) |
140 | 149 | notification_preference.web = False |
| 150 | + notification_preference.full_clean() |
141 | 151 | notification_preference.save() |
142 | 152 | notification_preference.refresh_from_db() |
| 153 | + # The database field has been updated to override the default |
| 154 | + # value in notification type. |
143 | 155 | self.assertEqual(notification_preference.email, False) |
144 | 156 | self._create_notification() |
145 | 157 | self.assertEqual(notification_queryset.count(), 0) |
@@ -953,77 +965,101 @@ def test_notification_for_unverified_email(self): |
953 | 965 | # we don't send emails to unverified email addresses |
954 | 966 | self.assertEqual(len(mail.outbox), 0) |
955 | 967 |
|
956 | | - @patch('openwisp_notifications.tasks.send_batched_email_notifications.apply_async') |
957 | | - def test_send_batched_email_notifications_no_instance_id(self, mock_send_email): |
958 | | - # Ensure no emails are sent if instance_id is None or empty |
| 968 | + @patch('logging.Logger.error') |
| 969 | + def test_send_batched_email_notifications_no_instance_id(self, mocked_logger): |
| 970 | + # Ensure no emails are sent if instance_id is invalid |
959 | 971 | tasks.send_batched_email_notifications(None) |
960 | 972 | self.assertEqual(len(mail.outbox), 0) |
| 973 | + mocked_logger.assert_called_once_with( |
| 974 | + 'Failed to send batched email notifications:' |
| 975 | + f' User with ID {None} not found in the database.' |
| 976 | + ) |
| 977 | + mocked_logger.reset_mock() |
961 | 978 |
|
962 | | - tasks.send_batched_email_notifications("") |
| 979 | + invalid_id = uuid4() |
| 980 | + tasks.send_batched_email_notifications(str(invalid_id)) |
963 | 981 | self.assertEqual(len(mail.outbox), 0) |
| 982 | + mocked_logger.assert_called_once_with( |
| 983 | + 'Failed to send batched email notifications:' |
| 984 | + f' User with ID {invalid_id} not found in the database.' |
| 985 | + ) |
964 | 986 |
|
965 | 987 | @patch('openwisp_notifications.tasks.send_batched_email_notifications.apply_async') |
966 | 988 | def test_send_batched_email_notifications_single_notification( |
967 | 989 | self, mock_send_email |
968 | 990 | ): |
969 | | - # Ensure no batch email template is used for a single batched notification |
| 991 | + # The first notification will always be sent immediately |
970 | 992 | self._create_notification() |
971 | | - n = self._create_notification().pop()[1][0] |
972 | | - |
973 | | - tasks.send_batched_email_notifications(self.admin.id) |
974 | | - |
975 | | - self.assertEqual(len(mail.outbox), 2) |
976 | | - self.assertIn(n.message, mail.outbox[0].body) |
| 993 | + mail.outbox.clear() |
| 994 | + # There's only one notification in the batch, it should |
| 995 | + # not use summary of batched email. |
| 996 | + self._create_notification() |
| 997 | + # The task is mocked to prevent immediate execution in test environment. |
| 998 | + # In tests, Celery runs in EAGER mode which would execute tasks immediately, |
| 999 | + # preventing us from testing the batching behavior properly. |
| 1000 | + tasks.send_batched_email_notifications(str(self.admin.id)) |
| 1001 | + self.assertEqual(len(mail.outbox), 1) |
| 1002 | + self.assertNotIn('new notifications since', mail.outbox[0].body) |
977 | 1003 |
|
978 | | - # @override_settings(TIME_ZONE='UTC') |
979 | 1004 | @patch('openwisp_notifications.tasks.send_batched_email_notifications.apply_async') |
980 | 1005 | def test_batch_email_notification(self, mock_send_email): |
981 | 1006 | fixed_datetime = timezone.localtime( |
982 | 1007 | datetime(2024, 7, 26, 11, 40, tzinfo=timezone.utc) |
983 | 1008 | ) |
984 | 1009 | datetime_str = fixed_datetime.strftime('%B %-d, %Y, %-I:%M %p %Z') |
985 | 1010 |
|
| 1011 | + # Add multiple notifications with slightly different timestamps |
| 1012 | + # to maintain consistent order in generated email text across test runs |
| 1013 | + for _ in range(3): |
| 1014 | + with freeze_time(fixed_datetime): |
| 1015 | + # Create notifications with URL (from self.notification_options) |
| 1016 | + self._create_notification() |
| 1017 | + fixed_datetime += timedelta( |
| 1018 | + microseconds=100 |
| 1019 | + ) # Increment time for ordering consistency |
| 1020 | + # Notification without URL |
| 1021 | + self.notification_options.pop('url') |
986 | 1022 | with freeze_time(fixed_datetime): |
987 | | - for _ in range(5): |
988 | | - notify.send(recipient=self.admin, **self.notification_options) |
989 | | - |
990 | | - # Check if only one mail is sent initially |
991 | | - self.assertEqual(len(mail.outbox), 1) |
992 | | - |
993 | | - # Call the task |
994 | | - tasks.send_batched_email_notifications(self.admin.id) |
995 | | - |
996 | | - # Check if the rest of the notifications are sent in a batch |
997 | | - self.assertEqual(len(mail.outbox), 2) |
998 | | - |
999 | | - expected_subject = f'[example.com] 4 new notifications since {datetime_str}' |
1000 | | - expected_body = f""" |
1001 | | -[example.com] 4 new notifications since {datetime_str} |
1002 | | -
|
1003 | | -
|
1004 | | -- Test Notification |
1005 | | - Description: Test Notification |
1006 | | - Date & Time: {datetime_str} |
1007 | | - URL: https://localhost:8000/admin |
1008 | | -
|
1009 | | -- Test Notification |
1010 | | - Description: Test Notification |
1011 | | - Date & Time: {datetime_str} |
1012 | | - URL: https://localhost:8000/admin |
| 1023 | + self._create_notification() |
| 1024 | + fixed_datetime += timedelta(microseconds=100) |
| 1025 | + # Notification with a type and target object |
| 1026 | + self.notification_options.update( |
| 1027 | + {'type': 'default', 'target': self._get_org_user()} |
| 1028 | + ) |
| 1029 | + with freeze_time(fixed_datetime): |
| 1030 | + default = self._create_notification().pop()[1][0] |
1013 | 1031 |
|
1014 | | -- Test Notification |
1015 | | - Description: Test Notification |
1016 | | - Date & Time: {datetime_str} |
1017 | | - URL: https://localhost:8000/admin |
| 1032 | + # Check if only one mail is sent initially |
| 1033 | + self.assertEqual(len(mail.outbox), 1) |
1018 | 1034 |
|
1019 | | -- Test Notification |
1020 | | - Description: Test Notification |
1021 | | - Date & Time: {datetime_str} |
1022 | | - URL: https://localhost:8000/admin |
1023 | | - """ |
| 1035 | + # Call the task |
| 1036 | + tasks.send_batched_email_notifications(self.admin.id) |
1024 | 1037 |
|
1025 | | - self.assertEqual(mail.outbox[1].subject, expected_subject) |
1026 | | - self.assertEqual(mail.outbox[1].body.strip(), expected_body.strip()) |
| 1038 | + # Check if the rest of the notifications are sent in a batch |
| 1039 | + self.assertEqual(len(mail.outbox), 2) |
| 1040 | + email = mail.outbox[1] |
| 1041 | + expected_subject = f'[example.com] 4 new notifications since {datetime_str}' |
| 1042 | + self.assertEqual(email.subject, expected_subject) |
| 1043 | + self.assertEqual( |
| 1044 | + email.body.strip(), |
| 1045 | + _test_batch_email_notification_email_body.format( |
| 1046 | + datetime_str=datetime_str, |
| 1047 | + notification_id=default.id, |
| 1048 | + ).strip(), |
| 1049 | + ) |
| 1050 | + html_email = email.alternatives[0][0] |
| 1051 | + self.maxDiff = None |
| 1052 | + self.assertIn( |
| 1053 | + strip_spaces_between_tags( |
| 1054 | + _test_batch_email_notification_email_html.format( |
| 1055 | + datetime_str=datetime_str, |
| 1056 | + notification_id=default.id, |
| 1057 | + ) |
| 1058 | + .replace('\n', '') |
| 1059 | + .replace(' ', ' ') |
| 1060 | + ), |
| 1061 | + strip_spaces_between_tags(html_email), |
| 1062 | + ) |
1027 | 1063 |
|
1028 | 1064 | @patch('openwisp_notifications.tasks.send_batched_email_notifications.apply_async') |
1029 | 1065 | def test_batch_email_notification_with_call_to_action(self, mock_send_email): |
|
0 commit comments