|
1 | 1 | from django.test import TestCase |
| 2 | +from django.urls import reverse |
| 3 | +from django.contrib.auth import get_user_model |
| 4 | +from rest_framework.test import APIClient |
| 5 | +from rest_framework import status |
| 6 | +from unittest.mock import patch, MagicMock |
2 | 7 |
|
3 | | -# Create your tests here. |
| 8 | +from apps.notifications.models import Notification |
| 9 | +from apps.notifications.utils import create_notification, deliver_notification, send_notification_to_channel_layer |
| 10 | +from apps.notifications.tasks import process_scheduled_notifications |
| 11 | + |
| 12 | +""" |
| 13 | + MagicMock() helps create mock objects for testing. It's particularly useful when you need to simulate the |
| 14 | + behavior of complex objects or external dependencies without actually using them in your tests. |
| 15 | + |
| 16 | + https://docs.python.org/3/library/unittest.mock.html |
| 17 | + """ |
| 18 | + |
| 19 | +User = get_user_model() |
| 20 | + |
| 21 | + |
| 22 | +class NotificationTestCase(TestCase): |
| 23 | + def setUp(self): |
| 24 | + # create a test user |
| 25 | + self.user = User.objects.create_user( |
| 26 | + username="testuser", |
| 27 | + password="testpassword", |
| 28 | + first_name="Test", |
| 29 | + last_name="User", |
| 30 | + email="test@example.com" |
| 31 | + ) |
| 32 | + |
| 33 | + # create a test notification |
| 34 | + self.notification = Notification.objects.create( |
| 35 | + user=self.user, |
| 36 | + notification_title='Test Notification', |
| 37 | + notification_message='This is a test notification', |
| 38 | + notification_type='info', |
| 39 | + info_category='general' |
| 40 | + ) |
| 41 | + |
| 42 | + # setup API client |
| 43 | + self.client = APIClient() |
| 44 | + # sets up authentication to test API client without going through the normal authentication process |
| 45 | + self.client.force_authenticate(user=self.user) |
| 46 | + |
| 47 | + # models.py tests |
| 48 | + def test_notification_model(self): |
| 49 | + self.assertEqual(self.notification.notification_title, 'Test Notification') |
| 50 | + self.assertEqual(self.notification.notification_message, 'This is a test notification') |
| 51 | + self.assertEqual(self.notification.user, self.user) |
| 52 | + self.assertFalse(self.notification.is_read) |
| 53 | + self.assertFalse(self.notification.is_scheduled) |
| 54 | + self.assertEqual(self.notification.info_subcategory, 'General Information') |
| 55 | + |
| 56 | + def test_mark_as_read(self): |
| 57 | + self.assertFalse(self.notification.is_read) |
| 58 | + self.notification.mark_as_read() |
| 59 | + self.assertTrue(self.notification.is_read) |
| 60 | + |
| 61 | + # utility.py function tests |
| 62 | + def test_create_notification_function(self): |
| 63 | + notif = create_notification( |
| 64 | + user=self.user, |
| 65 | + title='Utility Test', |
| 66 | + message='Testing utility function', |
| 67 | + notification_type='WARNING' |
| 68 | + ) |
| 69 | + |
| 70 | + self.assertEqual(notif.notification_title, 'Utility Test') |
| 71 | + self.assertEqual(notif.notification_type, 'warning') |
| 72 | + |
| 73 | + def test_send_notification_to_channel_layer(self): |
| 74 | + from apps.notifications.serializers import NotificationSerializer |
| 75 | + |
| 76 | + serializer = NotificationSerializer(self.notification) |
| 77 | + expected_data = serializer.data |
| 78 | + |
| 79 | + # creating mock channel layer |
| 80 | + mock_channel_layer = MagicMock() |
| 81 | + |
| 82 | + # -- AI Generated |
| 83 | + # Mock async_to_sync to simply call the function directly |
| 84 | + with patch('asgiref.sync.async_to_sync', lambda f: f): |
| 85 | + send_notification_to_channel_layer( |
| 86 | + mock_channel_layer, |
| 87 | + self.user.id, |
| 88 | + self.notification |
| 89 | + ) |
| 90 | + # -- |
| 91 | + # verify the channel layer was called correctly |
| 92 | + mock_channel_layer.group_send.assert_called_once() |
| 93 | + |
| 94 | + # check group name |
| 95 | + args = mock_channel_layer.group_send.call_args[0] |
| 96 | + self.assertEqual(args[0], f'notifications_{self.user.id}') |
| 97 | + |
| 98 | + # check message structure |
| 99 | + message = args[1] |
| 100 | + self.assertEqual(message['type'], 'notification_event') |
| 101 | + self.assertEqual(message['data'], expected_data) |
| 102 | + |
| 103 | + # test with pre-serialized data |
| 104 | + mock_channel_layer.reset_mock() |
| 105 | + send_notification_to_channel_layer( |
| 106 | + mock_channel_layer, |
| 107 | + self.user.id, |
| 108 | + expected_data # pass the data directly instead of notification object |
| 109 | + ) |
| 110 | + |
| 111 | + mock_channel_layer.group_send.assert_called_once() |
| 112 | + second_message = mock_channel_layer.group_send.call_args[0][1] |
| 113 | + self.assertEqual(second_message['data'], expected_data) |
| 114 | + |
| 115 | + @patch('channels.layers.get_channel_layer') |
| 116 | + @patch('asgiref.sync.async_to_sync') |
| 117 | + def test_deliver_notification(self, mock_async_to_sync, mock_get_channel_layer): |
| 118 | + # setup mocks |
| 119 | + mock_channel_layer = MagicMock() |
| 120 | + mock_group_send = MagicMock() |
| 121 | + mock_async_to_sync.return_value = mock_group_send |
| 122 | + mock_get_channel_layer.return_value = mock_channel_layer |
| 123 | + |
| 124 | + # call the function |
| 125 | + result = deliver_notification(self.notification) |
| 126 | + |
| 127 | + # verify results |
| 128 | + self.assertTrue(result) |
| 129 | + mock_get_channel_layer.assert_called_once() |
| 130 | + mock_async_to_sync.assert_called_once_with(mock_channel_layer.group_send) |
| 131 | + |
| 132 | + # API endpoint tests |
| 133 | + def test_list_notifications(self): |
| 134 | + url = reverse('notification-list') |
| 135 | + response = self.client.get(url) |
| 136 | + |
| 137 | + self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 138 | + self.assertEqual(len(response.data), 1) |
| 139 | + |
| 140 | + @patch('apps.notifications.utils.deliver_notification') |
| 141 | + def test_create_notification_api(self, mock_deliver): |
| 142 | + mock_deliver.return_value = True |
| 143 | + |
| 144 | + url = reverse('notification-list') |
| 145 | + data = { |
| 146 | + 'user': self.user.id, |
| 147 | + 'notification_title': 'API Test', |
| 148 | + 'notification_message': 'Testing API creation', |
| 149 | + 'notification_type': 'success', |
| 150 | + 'info_category': 'general' |
| 151 | + } |
| 152 | + |
| 153 | + response = self.client.post(url, data) |
| 154 | + |
| 155 | + self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
| 156 | + mock_deliver.assert_called_once() |
| 157 | + |
| 158 | + def test_mark_notification_read_api(self): |
| 159 | + url = reverse('notification-mark-read', args=[self.notification.id]) |
| 160 | + response = self.client.patch(url) |
| 161 | + |
| 162 | + self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 163 | + |
| 164 | + self.notification.refresh_from_db() |
| 165 | + self.assertTrue(self.notification.is_read) |
| 166 | + |
| 167 | + def test_mark_all_read_api(self): |
| 168 | + Notification.objects.create( |
| 169 | + user=self.user, |
| 170 | + notification_title='Another Test', |
| 171 | + notification_message='Another test message', |
| 172 | + notification_type='warning' |
| 173 | + ) |
| 174 | + |
| 175 | + url = reverse('notification-mark-all-read') |
| 176 | + response = self.client.post(url) |
| 177 | + |
| 178 | + self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 179 | + self.assertEqual(response.data['status'], '2 notifications marked as read') |
| 180 | + |
| 181 | + # check all notifications are read |
| 182 | + self.assertEqual(Notification.objects.filter(user=self.user, is_read=True).count(), 2) |
| 183 | + |
| 184 | + def test_clear_all_notifications(self): |
| 185 | + url = reverse('notification-clear-all') |
| 186 | + response = self.client.delete(url) |
| 187 | + |
| 188 | + self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) |
| 189 | + self.assertEqual(Notification.objects.filter(user=self.user).count(), 0) |
| 190 | + |
| 191 | + # web socket tests using mocks |
| 192 | + """ VERY SIMPLE TEST AS OF NOW. MAY NEED UPDATING IN THE FUTURE """ |
| 193 | + |
| 194 | + @patch('apps.notifications.utils.send_notification_to_channel_layer') |
| 195 | + def test_websocket_notification_sending(self, mock_send): |
| 196 | + # -- AI GENERATED |
| 197 | + # Setup mocks |
| 198 | + mock_send.return_value = None # Function doesn't return anything |
| 199 | + |
| 200 | + # Create a mock channel layer |
| 201 | + mock_channel_layer = MagicMock() |
| 202 | + |
| 203 | + # Patch get_channel_layer to return our mock |
| 204 | + with patch('channels.layers.get_channel_layer', return_value=mock_channel_layer): |
| 205 | + # Call the function |
| 206 | + result = deliver_notification(self.notification) |
| 207 | + |
| 208 | + # Verify result and mock calls |
| 209 | + self.assertTrue(result) |
| 210 | + mock_send.assert_called_once_with( |
| 211 | + mock_channel_layer, |
| 212 | + self.notification.user.id, |
| 213 | + self.notification |
| 214 | + ) |
| 215 | + # -- |
| 216 | + |
| 217 | + # task.py test(s) |
| 218 | + """ |
| 219 | + ts pmo!!!!!!! i CANNOT get this one working for some reason. even when putting it through AI. |
| 220 | + will test scheduled notifs on my 4th PR since i am working with periodic reminders. |
| 221 | + may have to re-implement scheduled notifs, but we'll see :< |
| 222 | + """ |
| 223 | + # @patch('apps.notifications.utils.deliver_notification') |
| 224 | + # def test_process_scheduled_notifications(self, mock_deliver): |
| 225 | + # mock_deliver.return_value = True |
| 226 | + # |
| 227 | + # # -- AI Generated |
| 228 | + # # First, clear any existing scheduled notifications for clean testing |
| 229 | + # Notification.objects.filter(scheduled_time__isnull=False).delete() |
| 230 | + # |
| 231 | + # # Create a very specific past time for scheduled_time |
| 232 | + # past_time = timezone.now() - timezone.timedelta(hours=1) |
| 233 | + # |
| 234 | + # # Create our test notification |
| 235 | + # scheduled_notif = Notification.objects.create( |
| 236 | + # user=self.user, |
| 237 | + # notification_title='Scheduled Test', |
| 238 | + # notification_message='Testing scheduled notifications', |
| 239 | + # scheduled_time=past_time |
| 240 | + # ) |
| 241 | + # |
| 242 | + # # Verify we have only one scheduled notification |
| 243 | + # count = Notification.objects.filter(scheduled_time__isnull=False).count() |
| 244 | + # self.assertEqual(count, 1, "Should have exactly one scheduled notification") |
| 245 | + # |
| 246 | + # # Run the task |
| 247 | + # result = process_scheduled_notifications() |
| 248 | + # |
| 249 | + # # Check the result text for any number of processed notifications |
| 250 | + # self.assertIn("Processed", result) |
| 251 | + # self.assertIn("scheduled notifications", result) |
| 252 | + # |
| 253 | + # # More importantly, check that our notification was delivered |
| 254 | + # mock_deliver.assert_called_with(scheduled_notif) |
| 255 | + # |
| 256 | + # # And that it was marked as sent |
| 257 | + # scheduled_notif.refresh_from_db() |
| 258 | + # self.assertIsNotNone(scheduled_notif.sent_at) |
| 259 | + # # -- |
0 commit comments