push_notifs: Add endpoint for sending a test notification.

Fixes #23997

api_docs/changelog.md

@@ -20,6 +20,11 @@ format used by the Zulip server that they are interacting with.
 
 ## Changes in Zulip 8.0
 
+**Feature level 217**
+
+* [`POST /mobile_push/test_notification`](/api/test-notify): Added new endpoint
+  to send a test push notification to a mobile device or devices.
+
 **Feature level 216**:
 
 * `PATCH /realm`, [`POST register`](/api/register-queue),

tools/lib/capitalization.py

@@ -233,7 +233,7 @@ def check_banned_words(text: str) -> List[str]:
         if word in lower_cased_text:
             # Hack: Should move this into BANNED_WORDS framework; for
             # now, just hand-code the skips:
-            if "realm_name" in lower_cased_text:
+            if "realm_name" in lower_cased_text or "realm_uri" in lower_cased_text:
                 continue
             kwargs = dict(word=word, text=text, reason=reason)
             msg = "{word} found in '{text}'. {reason}".format(**kwargs)

version.py

@@ -33,7 +33,7 @@ DESKTOP_WARNING_VERSION = "5.9.3"
 # Changes should be accompanied by documentation explaining what the
 # new level means in api_docs/changelog.md, as well as "**Changes**"
 # entries in the endpoint's documentation in `zulip.yaml`.
-API_FEATURE_LEVEL = 216
+API_FEATURE_LEVEL = 217
 
 # Bump the minor PROVISION_VERSION to indicate that folks should provision
 # only when going from an old version of the code to a newer version. Bump

zerver/lib/push_notifications.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import base64
+import copy
 import logging
 import re
 from dataclasses import dataclass
@@ -1190,3 +1191,63 @@ def handle_push_notification(user_profile_id: int, missed_message: Dict[str, Any
     user_identity = UserPushIdentityCompat(user_id=user_profile.id)
     send_apple_push_notification(user_identity, apple_devices, apns_payload)
     send_android_push_notification(user_identity, android_devices, gcm_payload, gcm_options)
+
+
+def send_test_push_notification_directly_to_devices(
+    user_identity: UserPushIdentityCompat,
+    devices: Sequence[DeviceToken],
+    base_payload: Dict[str, Any],
+    remote: Optional["RemoteZulipServer"] = None,
+) -> None:
+    payload = copy.deepcopy(base_payload)
+    payload["event"] = "test-by-device-token"
+
+    apple_devices = [device for device in devices if device.kind == PushDeviceToken.APNS]
+    android_devices = [device for device in devices if device.kind == PushDeviceToken.GCM]
+    # Let's make the payloads separate objects to make sure mutating to make e.g. Android
+    # adjustments doesn't affect the Apple payload and vice versa.
+    apple_payload = copy.deepcopy(payload)
+    android_payload = copy.deepcopy(payload)
+
+    realm_uri = base_payload["realm_uri"]
+    apns_data = {
+        "alert": {
+            "title": _("Test notification"),
+            "body": _("This is a test notification from {realm_uri}.").format(realm_uri=realm_uri),
+        },
+        "sound": "default",
+        "custom": {"zulip": apple_payload},
+    }
+    send_apple_push_notification(user_identity, apple_devices, apns_data, remote=remote)
+
+    android_payload["time"] = datetime_to_timestamp(timezone_now())
+    gcm_options = {"priority": "high"}
+    send_android_push_notification(
+        user_identity, android_devices, android_payload, gcm_options, remote=remote
+    )
+
+
+def send_test_push_notification(user_profile: UserProfile, devices: List[PushDeviceToken]) -> None:
+    base_payload = get_base_payload(user_profile)
+    if uses_notification_bouncer():
+        for device in devices:
+            post_data = {
+                "user_uuid": str(user_profile.uuid),
+                "user_id": user_profile.id,
+                "token": device.token,
+                "token_kind": device.kind,
+                "base_payload": base_payload,
+            }
+
+            logger.info("Sending test push notification to bouncer: %r", post_data)
+            send_json_to_push_bouncer("POST", "push/test_notification", post_data)
+
+        return
+
+    # This server doesn't need the bouncer, so we send directly to the device.
+    user_identity = UserPushIdentityCompat(
+        user_id=user_profile.id, user_uuid=str(user_profile.uuid)
+    )
+    send_test_push_notification_directly_to_devices(
+        user_identity, devices, base_payload, remote=None
+    )

zerver/openapi/zulip.yaml

@@ -9142,6 +9142,54 @@ paths:
       responses:
         "200":
           $ref: "#/components/responses/SimpleSuccess"
+  /mobile_push/test_notification:
+    post:
+      operationId: test-notify
+      summary: Sends a test notifications to the user's mobile device(s)
+      tags: ["mobile"]
+      description: |
+        This endpoint allows a user to trigger a test push notification to their
+        selected mobile devices, or all their mobile devices.
+
+        **Changes**: New in Zulip 8.0 (feature level 217).
+      parameters:
+        - name: token
+          in: query
+          description: |
+            The push token for the device to send the test notification to.
+
+            If this parameter is not submitted, the test notification will be sent to all of the
+            user's devices registered on the server.
+
+            Mobile device clients should pass this parameter in order to trigger a notification
+            that is only delivered to this client.
+          schema:
+            type: string
+          example: "111222"
+          required: false
+      responses:
+        "200":
+          description: Success.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/JsonSuccess"
+        "400":
+          description: Bad request.
+          content:
+            application/json:
+              schema:
+                allOf:
+                  - $ref: "#/components/schemas/CodedError"
+                  - example:
+                      {
+                        "code": "BAD_REQUEST",
+                        "msg": "Token does not exist",
+                        "result": "error",
+                      }
+                    description: |
+                      An example JSON response for when a device with the specified token
+                      does not exist:
   /user_topics:
     post:
       operationId: update-user-topic

zerver/tests/test_openapi.py

@@ -1028,6 +1028,7 @@ class OpenAPIAttributesTest(ZulipTestCase):
             "drafts",
             "webhooks",
             "scheduled_messages",
+            "mobile",
         ]
         paths = OpenAPISpec(OPENAPI_SPEC_PATH).openapi()["paths"]
         for path, path_item in paths.items():

zerver/tests/test_push_notifications.py

@@ -40,6 +40,7 @@ from zerver.lib.push_notifications import (
     get_apns_badge_count,
     get_apns_badge_count_future,
     get_apns_context,
+    get_base_payload,
     get_message_payload_apns,
     get_message_payload_gcm,
     get_mobile_push_content,
@@ -111,6 +112,12 @@ class BouncerTestCase(ZulipTestCase):
         super().tearDown()
 
     def request_callback(self, request: PreparedRequest) -> Tuple[int, ResponseHeaders, bytes]:
+        kwargs = {}
+        if isinstance(request.body, bytes):
+            # send_json_to_push_bouncer sends the body as bytes containing json.
+            data = orjson.loads(request.body)
+            kwargs = dict(content_type="application/json")
+        else:
             assert isinstance(request.body, str) or request.body is None
             params: Dict[str, List[str]] = parse.parse_qs(request.body)
             # In Python 3, the values of the dict from `parse_qs` are
@@ -122,7 +129,7 @@ class BouncerTestCase(ZulipTestCase):
         assert settings.PUSH_NOTIFICATION_BOUNCER_URL is not None
         local_url = request.url.replace(settings.PUSH_NOTIFICATION_BOUNCER_URL, "")
         if request.method == "POST":
-            result = self.uuid_post(self.server_uuid, local_url, data, subdomain="")
+            result = self.uuid_post(self.server_uuid, local_url, data, subdomain="", **kwargs)
         elif request.method == "GET":
             result = self.uuid_get(self.server_uuid, local_url, data, subdomain="")
         return (result.status_code, result.headers, result.content)
@@ -142,6 +149,186 @@ class BouncerTestCase(ZulipTestCase):
         return {"user_id": user_id, "token": token, "token_kind": token_kind}
 
 
+class SendTestPushNotificationEndpointTest(BouncerTestCase):
+    def test_send_test_push_notification_api_invalid_token(self) -> None:
+        user = self.example_user("cordelia")
+        result = self.api_post(
+            user, "/api/v1/mobile_push/test_notification", {"token": "invalid"}, subdomain="zulip"
+        )
+        self.assert_json_error(result, "Token does not exist")
+
+        payload = {
+            "user_uuid": str(user.uuid),
+            "user_id": user.id,
+            "token": "invalid",
+            "token_kind": PushDeviceToken.GCM,
+            "base_payload": get_base_payload(user),
+        }
+        result = self.uuid_post(
+            self.server_uuid,
+            "/api/v1/remotes/push/test_notification",
+            payload,
+            subdomain="",
+            content_type="application/json",
+        )
+        self.assert_json_error(result, "Token does not exist")
+
+    def test_send_test_push_notification_api_no_bouncer_config(self) -> None:
+        """
+        Tests the endpoint on a server that doesn't use the bouncer, due to having its
+        own ability to send push notifications to devices directly.
+        """
+        user = self.example_user("cordelia")
+
+        android_token = "111222"
+        android_token_kind = PushDeviceToken.GCM
+        apple_token = "111223"
+        apple_token_kind = PushDeviceToken.APNS
+        android_device = PushDeviceToken.objects.create(
+            user=user, token=android_token, kind=android_token_kind
+        )
+        apple_device = PushDeviceToken.objects.create(
+            user=user, token=apple_token, kind=apple_token_kind
+        )
+
+        endpoint = "/api/v1/mobile_push/test_notification"
+        time_now = now()
+
+        # 1. First test for an android device.
+        # 2. Then test for an apple device.
+        # 3. Then test without submitting a specific token,
+        #    meaning both devices should get notified.
+
+        with mock.patch(
+            "zerver.lib.push_notifications.send_android_push_notification"
+        ) as mock_send_android_push_notification, time_machine.travel(time_now, tick=False):
+            result = self.api_post(user, endpoint, {"token": android_token}, subdomain="zulip")
+
+        expected_android_payload = {
+            "server": "testserver",
+            "realm_id": user.realm_id,
+            "realm_uri": "http://zulip.testserver",
+            "user_id": user.id,
+            "event": "test-by-device-token",
+            "time": datetime_to_timestamp(time_now),
+        }
+        expected_gcm_options = {"priority": "high"}
+        mock_send_android_push_notification.assert_called_once_with(
+            UserPushIdentityCompat(user_id=user.id, user_uuid=str(user.uuid)),
+            [android_device],
+            expected_android_payload,
+            expected_gcm_options,
+            remote=None,
+        )
+        self.assert_json_success(result)
+
+        with mock.patch(
+            "zerver.lib.push_notifications.send_apple_push_notification"
+        ) as mock_send_apple_push_notification, time_machine.travel(time_now, tick=False):
+            result = self.api_post(user, endpoint, {"token": apple_token}, subdomain="zulip")
+
+        expected_apple_payload = {
+            "alert": {
+                "title": "Test notification",
+                "body": "This is a test notification from http://zulip.testserver.",
+            },
+            "sound": "default",
+            "custom": {
+                "zulip": {
+                    "server": "testserver",
+                    "realm_id": user.realm_id,
+                    "realm_uri": "http://zulip.testserver",
+                    "user_id": user.id,
+                    "event": "test-by-device-token",
+                }
+            },
+        }
+        mock_send_apple_push_notification.assert_called_once_with(
+            UserPushIdentityCompat(user_id=user.id, user_uuid=str(user.uuid)),
+            [apple_device],
+            expected_apple_payload,
+            remote=None,
+        )
+        self.assert_json_success(result)
+
+        # Test without submitting a token value. Both devices should get notified.
+        with mock.patch(
+            "zerver.lib.push_notifications.send_apple_push_notification"
+        ) as mock_send_apple_push_notification, mock.patch(
+            "zerver.lib.push_notifications.send_android_push_notification"
+        ) as mock_send_android_push_notification, time_machine.travel(
+            time_now, tick=False
+        ):
+            result = self.api_post(user, endpoint, subdomain="zulip")
+
+        mock_send_android_push_notification.assert_called_once_with(
+            UserPushIdentityCompat(user_id=user.id, user_uuid=str(user.uuid)),
+            [android_device],
+            expected_android_payload,
+            expected_gcm_options,
+            remote=None,
+        )
+        mock_send_apple_push_notification.assert_called_once_with(
+            UserPushIdentityCompat(user_id=user.id, user_uuid=str(user.uuid)),
+            [apple_device],
+            expected_apple_payload,
+            remote=None,
+        )
+        self.assert_json_success(result)
+
+    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL="https://push.zulip.org.example.com")
+    @responses.activate
+    def test_send_test_push_notification_api_with_bouncer_config(self) -> None:
+        """
+        Tests the endpoint on a server that uses the bouncer. This will simulate the
+        end-to-end flow:
+        1. First we simulate a request from the mobile device to the remote server's
+        endpoint for a test notification.
+        2. As a result, the remote server makes a request to the bouncer to send that
+        notification.
+
+        We verify that the appropriate function for sending the notification to the
+        device is called on the bouncer as the ultimate result of the flow.
+        """
+
+        self.add_mock_response()
+
+        user = self.example_user("cordelia")
+        server = RemoteZulipServer.objects.get(uuid=self.server_uuid)
+
+        token = "111222"
+        token_kind = PushDeviceToken.GCM
+        PushDeviceToken.objects.create(user=user, token=token, kind=token_kind)
+        remote_device = RemotePushDeviceToken.objects.create(
+            server=server, user_uuid=str(user.uuid), token=token, kind=token_kind
+        )
+
+        endpoint = "/api/v1/mobile_push/test_notification"
+        time_now = now()
+        with mock.patch(
+            "zerver.lib.push_notifications.send_android_push_notification"
+        ) as mock_send_android_push_notification, time_machine.travel(time_now, tick=False):
+            result = self.api_post(user, endpoint, {"token": token}, subdomain="zulip")
+        expected_payload = {
+            "server": "testserver",
+            "realm_id": user.realm_id,
+            "realm_uri": "http://zulip.testserver",
+            "user_id": user.id,
+            "event": "test-by-device-token",
+            "time": datetime_to_timestamp(time_now),
+        }
+        expected_gcm_options = {"priority": "high"}
+        user_identity = UserPushIdentityCompat(user_id=user.id, user_uuid=str(user.uuid))
+        mock_send_android_push_notification.assert_called_once_with(
+            user_identity,
+            [remote_device],
+            expected_payload,
+            expected_gcm_options,
+            remote=server,
+        )
+        self.assert_json_success(result)
+
+
 class PushBouncerNotificationTest(BouncerTestCase):
     DEFAULT_SUBDOMAIN = ""
 

zerver/views/push_notifications.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from django.conf import settings
 from django.http import HttpRequest, HttpResponse
 from django.utils.translation import gettext as _
@@ -8,6 +10,7 @@ from zerver.lib.push_notifications import (
     add_push_device_token,
     b64_to_hex,
     remove_push_device_token,
+    send_test_push_notification,
 )
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
@@ -66,3 +69,24 @@ def remove_android_reg_id(
     validate_token(token, PushDeviceToken.GCM)
     remove_push_device_token(user_profile, token, PushDeviceToken.GCM)
     return json_success(request)
+
+
+@human_users_only
+@has_request_variables
+def send_test_push_notification_api(
+    request: HttpRequest, user_profile: UserProfile, token: Optional[str] = REQ(default=None)
+) -> HttpResponse:
+    # If a token is specified in the request, the test notification is supposed to be sent
+    # to that device. If no token is provided, the test notification should be sent to
+    # all devices registered for the user.
+    if token is not None:
+        try:
+            devices = [PushDeviceToken.objects.get(token=token, user=user_profile)]
+        except PushDeviceToken.DoesNotExist:
+            raise JsonableError(_("Token does not exist"))
+    else:
+        devices = list(PushDeviceToken.objects.filter(user=user_profile))
+
+    send_test_push_notification(user_profile, devices)
+
+    return json_success(request)

zilencer/urls.py

@@ -11,6 +11,7 @@ from zilencer.views import (
     remote_server_check_analytics,
     remote_server_notify_push,
     remote_server_post_analytics,
+    remote_server_send_test_notification,
     unregister_all_remote_push_devices,
     unregister_remote_push_device,
 )
@@ -23,6 +24,7 @@ push_bouncer_patterns = [
     remote_server_path("remotes/push/unregister", POST=unregister_remote_push_device),
     remote_server_path("remotes/push/unregister/all", POST=unregister_all_remote_push_devices),
     remote_server_path("remotes/push/notify", POST=remote_server_notify_push),
+    remote_server_path("remotes/push/test_notification", POST=remote_server_send_test_notification),
     # Push signup doesn't use the REST API, since there's no auth.
     path("remotes/server/register", register_remote_server),
     remote_server_path("remotes/server/deactivate", POST=deactivate_remote_server),

zilencer/views.py

@@ -15,6 +15,7 @@ from django.utils.timezone import now as timezone_now
 from django.utils.translation import gettext as _
 from django.utils.translation import gettext as err_
 from django.views.decorators.csrf import csrf_exempt
+from pydantic import BaseModel, ConfigDict
 
 from analytics.lib.counts import COUNT_STATS
 from corporate.lib.stripe import do_deactivate_remote_server
@@ -24,9 +25,11 @@ from zerver.lib.push_notifications import (
     UserPushIdentityCompat,
     send_android_push_notification,
     send_apple_push_notification,
+    send_test_push_notification_directly_to_devices,
 )
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
+from zerver.lib.typed_endpoint import JsonBodyPayload, typed_endpoint
 from zerver.lib.validator import (
     check_bool,
     check_capped_string,
@@ -281,6 +284,52 @@ def delete_duplicate_registrations(
     return deduplicated_registrations_to_return
 
 
+class TestNotificationPayload(BaseModel):
+    token: str
+    token_kind: int
+    user_id: int
+    user_uuid: str
+    base_payload: Dict[str, Any]
+
+    model_config = ConfigDict(extra="forbid")
+
+
+@typed_endpoint
+def remote_server_send_test_notification(
+    request: HttpRequest,
+    server: RemoteZulipServer,
+    *,
+    payload: JsonBodyPayload[TestNotificationPayload],
+) -> HttpResponse:
+    token = payload.token
+    token_kind = payload.token_kind
+
+    user_id = payload.user_id
+    user_uuid = payload.user_uuid
+
+    # The remote server only sends the base payload with basic user and server info,
+    # and the actual format of the test notification is defined on the bouncer, as that
+    # gives us the flexibility to modify it freely, without relying on other servers
+    # upgrading.
+    base_payload = payload.base_payload
+
+    # This is a new endpoint, so it can assume it will only be used by newer
+    # servers that will send user both UUID and ID.
+    user_identity = UserPushIdentityCompat(user_id=user_id, user_uuid=user_uuid)
+
+    try:
+        device = RemotePushDeviceToken.objects.get(
+            user_identity.filter_q(), token=token, kind=token_kind, server=server
+        )
+    except RemotePushDeviceToken.DoesNotExist:
+        raise JsonableError(err_("Token does not exist"))
+
+    send_test_push_notification_directly_to_devices(
+        user_identity, [device], base_payload, remote=server
+    )
+    return json_success(request)
+
+
 @has_request_variables
 def remote_server_notify_push(
     request: HttpRequest,

zproject/urls.py

@@ -92,6 +92,7 @@ from zerver.views.push_notifications import (
     add_apns_device_token,
     remove_android_reg_id,
     remove_apns_device_token,
+    send_test_push_notification_api,
 )
 from zerver.views.reactions import add_reaction, remove_reaction
 from zerver.views.read_receipts import read_receipts
@@ -380,6 +381,7 @@ v1_api_and_json_patterns = [
         "users/me/apns_device_token", POST=add_apns_device_token, DELETE=remove_apns_device_token
     ),
     rest_path("users/me/android_gcm_reg_id", POST=add_android_reg_id, DELETE=remove_android_reg_id),
+    rest_path("mobile_push/test_notification", POST=send_test_push_notification_api),
     # users/*/presence => zerver.views.presence.
     rest_path("users/me/presence", POST=update_active_status_backend),
     # It's important that this sit after users/me/presence so that