backend: Support user IDs for sending typing notifications.

This is a part of our efforts surrounding #9474.

zerver/lib/actions.py

@@ -1624,15 +1624,21 @@ def check_send_typing_notification(sender: UserProfile, notification_to: Sequenc
 # check_typing_notification:
 # Returns typing notification ready for sending with do_send_typing_notification on success
 # or the error message (string) on error.
-def check_typing_notification(sender: UserProfile, notification_to: Sequence[str],
+def check_typing_notification(sender: UserProfile,
+                              notification_to: Union[Sequence[str], Sequence[int]],
                               operator: str) -> Dict[str, Any]:
     if len(notification_to) == 0:
         raise JsonableError(_('Missing parameter: \'to\' (recipient)'))
     elif operator not in ('start', 'stop'):
         raise JsonableError(_('Invalid \'op\' value (should be start or stop)'))
+
     try:
-        recipient = recipient_for_emails(notification_to, False,
-                                         sender, sender)
+        if isinstance(notification_to[0], str):
+            emails = cast(Sequence[str], notification_to)
+            recipient = recipient_for_emails(emails, False, sender, sender)
+        elif isinstance(notification_to[0], int):
+            user_ids = cast(Sequence[int], notification_to)
+            recipient = recipient_for_user_ids(user_ids, sender)
     except ValidationError as e:
         assert isinstance(e.messages[0], str)
         raise JsonableError(e.messages[0])

zerver/tests/test_typing.py

@@ -78,6 +78,43 @@ class TypingNotificationRecipientsTest(ZulipTestCase):
         self.assertEqual(event['type'], 'typing')
         self.assertEqual(event['op'], 'start')
 
+    def test_single_recipient_by_user_id(self) -> None:
+        """
+        Sending typing notification to a single recipient (using user IDs)
+        is successful
+        """
+        sender = self.example_user('hamlet')
+        recipient_user = self.example_user('othello')
+        expected_recipients = set([sender, recipient_user])
+        expected_recipient_emails = set([user.email for user in expected_recipients])
+        expected_recipient_ids = set([user.id for user in expected_recipients])
+
+        events = []  # type: List[Mapping[str, Any]]
+        with tornado_redirected_to_list(events):
+            result = self.api_post(
+                sender.email,
+                '/api/v1/typing',
+                {
+                    'to': ujson.dumps([recipient_user.id]),
+                    'op': 'start'
+                }
+            )
+
+        self.assert_json_success(result)
+        self.assertEqual(len(events), 1)
+
+        event = events[0]['event']
+        event_recipient_emails = set(user['email'] for user in event['recipients'])
+        event_user_ids = set(events[0]['users'])
+        event_recipient_user_ids = set(user['user_id'] for user in event['recipients'])
+
+        self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
+        self.assertEqual(expected_recipient_ids, event_user_ids)
+        self.assertEqual(event['sender']['email'], sender.email)
+        self.assertEqual(event_recipient_emails, expected_recipient_emails)
+        self.assertEqual(event['type'], 'typing')
+        self.assertEqual(event['op'], 'start')
+
     def test_multiple_recipients(self) -> None:
         """
         Sending typing notification to a single recipient is successful
@@ -107,6 +144,36 @@ class TypingNotificationRecipientsTest(ZulipTestCase):
         self.assertEqual(event['type'], 'typing')
         self.assertEqual(event['op'], 'start')
 
+    def test_multiple_recipients_by_user_ids(self) -> None:
+        """
+        Sending typing notification to multiple recipients (using user IDs)
+        is successful
+        """
+        sender = self.example_user('hamlet')
+        recipient_users = [self.example_user('othello'), self.example_user('cordelia')]
+        expected_recipients = set(recipient_users) | set([sender])
+        expected_recipient_emails = set([user.email for user in expected_recipients])
+        expected_recipient_ids = set([user.id for user in expected_recipients])
+        events = []  # type: List[Mapping[str, Any]]
+        with tornado_redirected_to_list(events):
+            result = self.api_post(sender.email, '/api/v1/typing',
+                                   {'to': ujson.dumps([user.id for user in recipient_users]),
+                                    'op': 'start'})
+        self.assert_json_success(result)
+        self.assertEqual(len(events), 1)
+
+        event = events[0]['event']
+        event_recipient_emails = set(user['email'] for user in event['recipients'])
+        event_user_ids = set(events[0]['users'])
+        event_recipient_user_ids = set(user['user_id'] for user in event['recipients'])
+
+        self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
+        self.assertEqual(expected_recipient_ids, event_user_ids)
+        self.assertEqual(event['sender']['email'], sender.email)
+        self.assertEqual(event_recipient_emails, expected_recipient_emails)
+        self.assertEqual(event['type'], 'typing')
+        self.assertEqual(event['op'], 'start')
+
 class TypingStartedNotificationTest(ZulipTestCase):
     def test_send_notification_to_self_event(self) -> None:
         """
@@ -136,6 +203,40 @@ class TypingStartedNotificationTest(ZulipTestCase):
         self.assertEqual(event['type'], 'typing')
         self.assertEqual(event['op'], 'start')
 
+    def test_send_notification_to_self_by_user_id_event(self) -> None:
+        """
+        Sending typing notification to yourself (using user IDs)
+        is successful.
+        """
+        user = self.example_user('hamlet')
+        email = user.email
+        expected_recipient_emails = set([email])
+        expected_recipient_ids = set([user.id])
+        events = []  # type: List[Mapping[str, Any]]
+        with tornado_redirected_to_list(events):
+            result = self.api_post(
+                email,
+                '/api/v1/typing',
+                {
+                    'to': ujson.dumps([user.id]),
+                    'op': 'start'
+                }
+            )
+        self.assert_json_success(result)
+        self.assertEqual(len(events), 1)
+
+        event = events[0]['event']
+        event_recipient_emails = set(user['email'] for user in event['recipients'])
+        event_user_ids = set(events[0]['users'])
+        event_recipient_user_ids = set(user['user_id'] for user in event['recipients'])
+
+        self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
+        self.assertEqual(expected_recipient_ids, event_user_ids)
+        self.assertEqual(event_recipient_emails, expected_recipient_emails)
+        self.assertEqual(event['sender']['email'], email)
+        self.assertEqual(event['type'], 'typing')
+        self.assertEqual(event['op'], 'start')
+
     def test_send_notification_to_another_user_event(self) -> None:
         """
         Sending typing notification to another user