import mock import ujson from django.http import HttpRequest, HttpResponse from typing import Any, Callable, Dict, Tuple from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_helpers import POSTRequestMock from zerver.models import Recipient, Subscription, UserProfile, get_stream from zerver.tornado.event_queue import maybe_enqueue_notifications, \ get_client_descriptor, missedmessage_hook from zerver.tornado.views import get_events_backend class MissedMessageNotificationsTest(ZulipTestCase): """Tests the logic for when missed-message notifications should be triggered, based on user settings""" def check_will_notify(self, *args, **kwargs): # type: (*Any, **Any) -> Tuple[str, str] email_notice = None mobile_notice = None with mock.patch("zerver.tornado.event_queue.queue_json_publish") as mock_queue_publish: notified = maybe_enqueue_notifications(*args, **kwargs) for entry in mock_queue_publish.call_args_list: args = entry[0] if args[0] == "missedmessage_mobile_notifications": mobile_notice = args[1] if args[0] == "missedmessage_emails": email_notice = args[1] # Now verify the return value matches the queue actions if email_notice: self.assertTrue(notified['email_notified']) else: self.assertFalse(notified.get('email_notified', False)) if mobile_notice: self.assertTrue(notified['push_notified']) else: self.assertFalse(notified.get('push_notified', False)) return email_notice, mobile_notice def test_enqueue_notifications(self): # type: () -> None user_profile = self.example_user("hamlet") message_id = 32 # Boring message doesn't send a notice email_notice, mobile_notice = self.check_will_notify( user_profile.id, message_id, private_message=False, mentioned=False, stream_push_notify=False, stream_name=None, always_push_notify=False, idle=True) self.assertTrue(email_notice is None) self.assertTrue(mobile_notice is None) # Private message sends a notice email_notice, mobile_notice = self.check_will_notify( user_profile.id, message_id, private_message=True, mentioned=False, stream_push_notify=False, stream_name=None, always_push_notify=False, idle=True) self.assertTrue(email_notice is not None) self.assertTrue(mobile_notice is not None) # Mention sends a notice email_notice, mobile_notice = self.check_will_notify( user_profile.id, message_id, private_message=False, mentioned=True, stream_push_notify=False, stream_name=None, always_push_notify=False, idle=True) self.assertTrue(email_notice is not None) self.assertTrue(mobile_notice is not None) # stream_push_notify pushes but doesn't email email_notice, mobile_notice = self.check_will_notify( user_profile.id, message_id, private_message=False, mentioned=False, stream_push_notify=True, stream_name="Denmark", always_push_notify=False, idle=True) self.assertTrue(email_notice is None) self.assertTrue(mobile_notice is not None) # Private message doesn't send a notice if not idle email_notice, mobile_notice = self.check_will_notify( user_profile.id, message_id, private_message=True, mentioned=False, stream_push_notify=False, stream_name=None, always_push_notify=False, idle=False) self.assertTrue(email_notice is None) self.assertTrue(mobile_notice is None) # Private message sends push but not email if not idle but always_push_notify email_notice, mobile_notice = self.check_will_notify( user_profile.id, message_id, private_message=True, mentioned=False, stream_push_notify=False, stream_name=None, always_push_notify=True, idle=False) self.assertTrue(email_notice is None) self.assertTrue(mobile_notice is not None) def tornado_call(self, view_func, user_profile, post_data): # type: (Callable[[HttpRequest, UserProfile], HttpResponse], UserProfile, Dict[str, Any]) -> HttpResponse request = POSTRequestMock(post_data, user_profile) return view_func(request, user_profile) def test_end_to_end_missedmessage_hook(self): # type: () -> None """Tests what arguments missedmessage_hook passes into maybe_enqueue_notifications. Combined with the previous test, this ensures that the missedmessage_hook is correct""" user_profile = self.example_user('hamlet') email = user_profile.email self.login(email) result = self.tornado_call(get_events_backend, user_profile, {"apply_markdown": ujson.dumps(True), "event_types": ujson.dumps(["message"]), "user_client": "website", "dont_block": ujson.dumps(True), }) self.assert_json_success(result) queue_id = ujson.loads(result.content)["queue_id"] client_descriptor = get_client_descriptor(queue_id) with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue: # To test the missed_message hook, we first need to send a message msg_id = self.send_message(self.example_email("iago"), "Denmark", Recipient.STREAM) # Verify that nothing happens if you call it as not the # "last client descriptor", in which case the function # short-circuits, since the `missedmessage_hook` handler # for garbage-collection is only for the user's last queue. missedmessage_hook(user_profile.id, client_descriptor, False) mock_enqueue.assert_not_called() # Now verify that we called the appropriate enqueue function missedmessage_hook(user_profile.id, client_descriptor, True) mock_enqueue.assert_called_once() args_list = mock_enqueue.call_args_list[0][0] self.assertEqual(args_list, (user_profile.id, msg_id, False, False, False, "Denmark", False, True)) # Clear the event queue, before repeating with a private message client_descriptor.event_queue.pop() self.assertTrue(client_descriptor.event_queue.empty()) msg_id = self.send_message(self.example_email("iago"), [email], Recipient.PERSONAL) with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue: missedmessage_hook(user_profile.id, client_descriptor, True) mock_enqueue.assert_called_once() args_list = mock_enqueue.call_args_list[0][0] self.assertEqual(args_list, (user_profile.id, msg_id, True, False, False, None, False, True)) # Clear the event queue, now repeat with a mention client_descriptor.event_queue.pop() self.assertTrue(client_descriptor.event_queue.empty()) msg_id = self.send_message(self.example_email("iago"), "Denmark", Recipient.STREAM, content="@**King Hamlet** what's up?") with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue: # Clear the event queue, before repeating with a private message missedmessage_hook(user_profile.id, client_descriptor, True) mock_enqueue.assert_called_once() args_list = mock_enqueue.call_args_list[0][0] self.assertEqual(args_list, (user_profile.id, msg_id, False, True, False, "Denmark", False, True)) # Clear the event queue, now repeat with stream message with stream_push_notify stream = get_stream("Denmark", user_profile.realm) sub = Subscription.objects.get(user_profile=user_profile, recipient__type=Recipient.STREAM, recipient__type_id=stream.id) sub.push_notifications = True sub.save() client_descriptor.event_queue.pop() self.assertTrue(client_descriptor.event_queue.empty()) msg_id = self.send_message(self.example_email("iago"), "Denmark", Recipient.STREAM, content="what's up everyone?") with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue: # Clear the event queue, before repeating with a private message missedmessage_hook(user_profile.id, client_descriptor, True) mock_enqueue.assert_called_once() args_list = mock_enqueue.call_args_list[0][0] self.assertEqual(args_list, (user_profile.id, msg_id, False, False, True, "Denmark", False, True)) # Clean up the state sub.push_notifications = True sub.save()