From 056ebf686c15e61773bde44b5fa0072873e67801 Mon Sep 17 00:00:00 2001 From: Tim Jahn Date: Wed, 17 Jan 2024 11:26:59 +0100 Subject: [PATCH] Add Unit Tests for GCMDeviceAdmin and fix FCM v1 tests --- tests/test_admin.py | 130 ++++++++++++++++ tests/test_gcm_push_payload.py | 45 +++--- tests/test_models.py | 271 ++++++++++++++++++++++----------- 3 files changed, 336 insertions(+), 110 deletions(-) create mode 100644 tests/test_admin.py diff --git a/tests/test_admin.py b/tests/test_admin.py new file mode 100644 index 00000000..4b5adfd3 --- /dev/null +++ b/tests/test_admin.py @@ -0,0 +1,130 @@ +from unittest import mock + + +from django.contrib.admin import AdminSite +from django.contrib import messages +from django.http import HttpRequest +from django.test import TestCase + +from firebase_admin.messaging import Message, BatchResponse, SendResponse, UnregisteredError + +from push_notifications.admin import GCMDeviceAdmin +from push_notifications.models import GCMDevice +from tests import responses + + +class GCMDeviceAdminTestCase(TestCase): + def test_send_bulk_messages_action(self): + request = HttpRequest() + + GCMDevice.objects.create(registration_id="abc", cloud_message_type="FCM") + queryset = GCMDevice.objects.all() + admin = GCMDeviceAdmin(GCMDevice, AdminSite()) + admin.message_user = mock.Mock() + + with mock.patch( + "firebase_admin.messaging.send_all", return_value=responses.FCM_SUCCESS + ) as p: + admin.send_messages(request, queryset, bulk=True) + + # one call + self.assertEqual(len(p.mock_calls), 1) + + call = p.call_args + kwargs = call[1] + + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) + + # only one message + call_messages = call[0][0] + self.assertEqual(len(call_messages), 1) + + message = call_messages[0] + self.assertIsInstance(message, Message) + self.assertEqual(message.token, "abc") + self.assertEqual(message.android.notification.body, "Test bulk notification") + + admin.message_user.assert_called_once_with( + request, "All messages were sent.", level=messages.SUCCESS + ) + + def test_send_single_message_action(self): + request = HttpRequest() + + GCMDevice.objects.create(registration_id="abc", cloud_message_type="FCM") + queryset = GCMDevice.objects.all() + admin = GCMDeviceAdmin(GCMDevice, AdminSite()) + admin.message_user = mock.Mock() + + with mock.patch( + "firebase_admin.messaging.send_all", return_value=responses.FCM_SUCCESS + ) as p: + admin.send_messages(request, queryset, bulk=False) + + # one call + self.assertEqual(len(p.mock_calls), 1) + + call = p.call_args + kwargs = call[1] + + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) + + # only one message + call_messages = call[0][0] + self.assertEqual(len(call_messages), 1) + + message = call_messages[0] + self.assertIsInstance(message, Message) + self.assertEqual(message.token, "abc") + self.assertEqual(message.android.notification.body, "Test single notification") + + admin.message_user.assert_called_once_with( + request, "All messages were sent.", level=messages.SUCCESS + ) + + def test_send_bulk_messages_action_fail(self): + request = HttpRequest() + + GCMDevice.objects.create(registration_id="abc", cloud_message_type="FCM") + queryset = GCMDevice.objects.all() + admin = GCMDeviceAdmin(GCMDevice, AdminSite()) + admin.message_user = mock.Mock() + + response = BatchResponse( + [SendResponse(resp={"name": "..."}, exception=UnregisteredError("error"),)] + ) + + with mock.patch( + "firebase_admin.messaging.send_all", return_value=response + ) as p: + admin.send_messages(request, queryset, bulk=True) + + # one call + self.assertEqual(len(p.mock_calls), 1) + + call = p.call_args + kwargs = call[1] + + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) + + # only one message + call_messages = call[0][0] + self.assertEqual(len(call_messages), 1) + + message = call_messages[0] + self.assertIsInstance(message, Message) + self.assertEqual(message.token, "abc") + self.assertEqual(message.android.notification.body, "Test bulk notification") + + admin.message_user.assert_called_once_with( + request, "Some messages could not be processed: UnregisteredError('error')", level=messages.ERROR + ) diff --git a/tests/test_gcm_push_payload.py b/tests/test_gcm_push_payload.py index e3447d70..657cdc9d 100644 --- a/tests/test_gcm_push_payload.py +++ b/tests/test_gcm_push_payload.py @@ -17,22 +17,21 @@ def test_fcm_push_payload(self): send_message("abc", message) - # one call - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] + self.assertEqual(p.call_count, 1) - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message - self.assertEqual(len(call.args[0]), 1) + messages = call[0][0] + self.assertEqual(len(messages), 1) - message = call.args[0][0] + message = messages[0] self.assertIsInstance(message, Message) self.assertEqual(message.token, "abc") self.assertEqual(message.android.notification.body, "Hello world") @@ -44,27 +43,25 @@ def test_fcm_push_payload_many(self): send_message(["abc", "123"], message) # one call - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] - - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) - messages_arg = call.args[0] + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # two message - self.assertEqual(len(messages_arg), 2) + messages = call[0][0] + self.assertEqual(len(messages), 2) - message_one = messages_arg[0] + message_one = messages[0] self.assertIsInstance(message_one, Message) self.assertEqual(message_one.token, "abc") self.assertEqual(message_one.android.notification.body, "Hello world") - message_two = messages_arg[1] + message_two = messages[1] self.assertIsInstance(message_two, Message) self.assertEqual( message_two.token,"123") self.assertEqual( message_two.android.notification.body, "Hello world") diff --git a/tests/test_models.py b/tests/test_models.py index 55f116ef..6a9996a7 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -44,24 +44,58 @@ def test_fcm_send_message(self): # one call self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message - self.assertEqual(len(call.args[0]), 1) + messages = call[0][0] + self.assertEqual(len(messages), 1) - message = call.args[0][0] + message = messages[0] self.assertIsInstance(message, Message) self.assertEqual(message.token, "abc") self.assertEqual(message.android.notification.body, "Hello world") + def test_fcm_send_message_with_fcm_message(self): + device = GCMDevice.objects.create(registration_id="abc", cloud_message_type="FCM") + with mock.patch( + "firebase_admin.messaging.send_all", return_value=responses.FCM_SUCCESS + ) as p: + message_to_send = messaging.Message( + notification=messaging.Notification( + title='Hello world', + body='What a beautiful day.' + ), + ) + device.send_message(message_to_send) + + # one call + self.assertEqual(len(p.mock_calls), 1) + + call = p.call_args + kwargs = call[1] + + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) + + # only one message + messages = call[0][0] + self.assertEqual(len(messages), 1) + + message = messages[0] + self.assertIsInstance(message, Message) + self.assertEqual(message.token, "abc") + self.assertEqual(message.notification.body, "What a beautiful day.") + self.assertEqual(message.notification.title, "Hello world") + def test_fcm_send_message_extra_data(self): device = GCMDevice.objects.create(registration_id="abc", cloud_message_type="FCM") with mock.patch( @@ -70,20 +104,19 @@ def test_fcm_send_message_extra_data(self): device.send_message("Hello world", extra={"foo": "bar"}) self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] - - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message - self.assertEqual(len(call.args[0]), 1) + messages = call[0][0] + self.assertEqual(len(messages), 1) - message = call.args[0][0] + message = messages[0] self.assertIsInstance(message, Message) self.assertEqual(message.token, "abc") self.assertEqual(message.android.notification.body, "Hello world") @@ -97,20 +130,19 @@ def test_fcm_send_message_extra_options(self): device.send_message("Hello world", collapse_key="test_key", foo="bar") self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] + call = p.call_args + kwargs = call[1] - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) - - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message - self.assertEqual(len(call.args[0]), 1) + messages = call[0][0] + self.assertEqual(len(messages), 1) - message = call.args[0][0] + message = messages[0] self.assertIsInstance(message, Message) self.assertEqual(message.token, "abc") self.assertEqual(message.android.notification.body, "Hello world") @@ -124,21 +156,20 @@ def test_fcm_send_message_extra_notification(self): ) as p: device.send_message("Hello world", extra={"icon": "test_icon"}, title="test") - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) - - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message - self.assertEqual(len(call.args[0]), 1) + messages = call[0][0] + self.assertEqual(len(messages), 1) - message = call.args[0][0] + message = messages[0] self.assertIsInstance(message, Message) self.assertEqual(message.token, "abc") self.assertEqual(message.android.notification.body, "Hello world") @@ -158,21 +189,20 @@ def test_fcm_send_message_extra_options_and_notification_and_data(self): collapse_key="test_key" ) - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] - - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message - self.assertEqual(len(call.args[0]), 1) + messages = call[0][0] + self.assertEqual(len(messages), 1) - message = call.args[0][0] + message = messages[0] self.assertIsInstance(message, Message) self.assertEqual(message.token, "abc") self.assertEqual(message.android.collapse_key, "test_key") @@ -189,22 +219,21 @@ def test_fcm_send_message_to_multiple_devices(self): ) as p: GCMDevice.objects.all().send_message("Hello world") - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] - - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # two messages - self.assertEqual(len(call.args[0]), 2) + messages = call[0][0] + self.assertEqual(len(messages), 2) - message_one = call.args[0][0] - message_two = call.args[0][1] + message_one = messages[0] + message_two = messages[1] self.assertIsInstance(message_one, Message) self.assertIsInstance(message_two, Message) self.assertEqual(message_one.token, "abc") @@ -212,6 +241,78 @@ def test_fcm_send_message_to_multiple_devices(self): self.assertEqual(message_one.android.notification.body, "Hello world") self.assertEqual(message_two.android.notification.body, "Hello world") + def test_fcm_send_message_to_multiple_devices_fcm_message(self): + self._create_fcm_devices(["abc", "abc1"]) + + with mock.patch( + "firebase_admin.messaging.send_all", return_value=responses.FCM_SUCCESS_MULTIPLE + ) as p: + message_to_send = messaging.Message( + notification=messaging.Notification( + title='Hello world', + body='What a beautiful day.' + ), + ) + GCMDevice.objects.all().send_message(message_to_send) + + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] + + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) + + # two messages + messages = call[0][0] + self.assertEqual(len(messages), 2) + + message_one = messages[0] + message_two = messages[1] + self.assertIsInstance(message_one, Message) + self.assertIsInstance(message_two, Message) + self.assertEqual(message_one.token, "abc") + self.assertEqual(message_two.token, "abc1") + self.assertEqual(message_one.notification.title, "Hello world") + self.assertEqual(message_one.notification.body, "What a beautiful day.") + self.assertEqual(message_two.notification.title, "Hello world") + self.assertEqual(message_two.notification.body, "What a beautiful day.") + + def test_gcm_send_message_does_not_send(self): + device = GCMDevice.objects.create(registration_id="abc", cloud_message_type="GCM") + + with mock.patch( + "firebase_admin.messaging.send_all", return_value=responses.FCM_SUCCESS_MULTIPLE + ) as p: + message_to_send = messaging.Message( + notification=messaging.Notification( + title='Hello world', + body='What a beautiful day.' + ), + ) + response = device.send_message(message_to_send) + self.assertIsNone(response) + self.assertEqual(p.call_count, 0) + + def test_gcm_send_multiple_message_does_not_send(self): + self._create_devices(["abc", "abc1"]) + + with mock.patch( + "firebase_admin.messaging.send_all", return_value=responses.FCM_SUCCESS_MULTIPLE + ) as p: + message_to_send = messaging.Message( + notification=messaging.Notification( + title='Hello world', + body='What a beautiful day.' + ), + ) + response = GCMDevice.objects.all().send_message(message_to_send).responses + + self.assertEqual(p.call_count, 0) + self.assertEqual(len(response), 0) + + def test_fcm_send_message_active_devices(self): GCMDevice.objects.create(registration_id="abc", active=True, cloud_message_type="FCM") GCMDevice.objects.create(registration_id="xyz", active=False, cloud_message_type="FCM") @@ -221,20 +322,19 @@ def test_fcm_send_message_active_devices(self): ) as p: GCMDevice.objects.all().send_message("Hello world") - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) - - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # only one message (one is inactive) - self.assertEqual(len(call.args[0]), 1) - message_one = call.args[0][0] + messages = call[0][0] + self.assertEqual(len(messages), 1) + message_one = messages[0] self.assertIsInstance(message_one, Message) self.assertEqual(message_one.token, "abc") @@ -248,22 +348,21 @@ def test_fcm_send_message_collapse_to_multiple_devices(self): ) as p: GCMDevice.objects.all().send_message("Hello world", collapse_key="test_key") - self.assertEqual(len(p.mock_calls), 1) - call = p.mock_calls[0] - - # only messages is args, dry_run and app are in kwargs - self.assertEqual(len(call.args), 1) + self.assertEqual(p.call_count, 1) + call = p.call_args + kwargs = call[1] - self.assertTrue("dry_run" in call.kwargs) - self.assertFalse(call.kwargs["dry_run"]) - self.assertTrue("app" in call.kwargs) - self.assertIsNone(call.kwargs["app"]) + self.assertTrue("dry_run" in kwargs) + self.assertFalse(kwargs["dry_run"]) + self.assertTrue("app" in kwargs) + self.assertIsNone(kwargs["app"]) # two messages - self.assertEqual(len(call.args[0]), 2) + messages = call[0][0] + self.assertEqual(len(messages), 2) - message_one = call.args[0][0] - message_two = call.args[0][1] + message_one = messages[0] + message_two = messages[1] self.assertIsInstance(message_one, Message) self.assertIsInstance(message_two, Message) self.assertEqual(message_one.token, "abc")