import base64
import uuid
from collections.abc import Mapping
from datetime import timedelta
from typing import Any
from unittest import mock, skipUnless

import aioapns
import firebase_admin.messaging as firebase_messaging
import orjson
import requests
import responses
import time_machine
from django.conf import settings
from django.db.models import F, Q
from django.test import override_settings
from django.utils.crypto import get_random_string
from django.utils.timezone import now
from dns.resolver import NoAnswer as DNSNoAnswer
from firebase_admin import exceptions as firebase_exceptions
from requests.exceptions import ConnectionError
from typing_extensions import override

from analytics.models import RealmCount
from zerver.actions.message_flags import do_mark_stream_messages_as_read, do_update_message_flags
from zerver.actions.user_groups import check_add_user_group
from zerver.actions.user_settings import do_regenerate_api_key
from zerver.lib.avatar import absolute_avatar_url, get_avatar_for_inaccessible_user
from zerver.lib.exceptions import JsonableError
from zerver.lib.push_notifications import (
    DeviceToken,
    InvalidRemotePushDeviceTokenError,
    UserPushIdentityCompat,
    get_apns_badge_count,
    get_apns_badge_count_future,
    get_apns_context,
    get_base_payload,
    get_message_payload,
    get_message_payload_apns,
    get_message_payload_gcm,
    get_mobile_push_content,
    handle_push_notification,
    parse_fcm_options,
    send_android_push_notification_to_user,
    send_apple_push_notification,
    send_notifications_to_bouncer,
)
from zerver.lib.remote_server import (
    PushNotificationBouncerError,
    PushNotificationBouncerRetryLaterError,
    PushNotificationBouncerServerError,
    get_realms_info_for_push_bouncer,
    prepare_for_registration_transfer_challenge,
    send_to_push_bouncer,
)
from zerver.lib.response import json_response_from_error
from zerver.lib.send_email import FromAddress
from zerver.lib.test_classes import BouncerTestCase, PushNotificationTestCase, ZulipTestCase
from zerver.lib.test_helpers import (
    activate_push_notification_service,
    mock_queue_publish,
    reset_email_visibility_to_everyone_in_zulip_realm,
)
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.models import (
    Message,
    PushDeviceToken,
    RealmAuditLog,
    Recipient,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
)
from zerver.models.clients import get_client
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import get_realm
from zerver.models.recipients import get_or_create_direct_message_group
from zerver.models.scheduled_jobs import NotificationTriggers
from zerver.models.streams import get_stream
from zilencer.auth import (
    REMOTE_SERVER_TAKEOVER_TOKEN_VALIDITY_SECONDS,
    generate_registration_transfer_verification_secret,
)
from zilencer.models import RemoteZulipServerAuditLog
from zilencer.views import DevicesToCleanUpDict, get_deleted_devices

if settings.ZILENCER_ENABLED:
    from zilencer.models import (
        RemotePushDeviceToken,
        RemoteRealm,
        RemoteRealmAuditLog,
        RemoteZulipServer,
    )
    from zilencer.views import update_remote_realm_data_for_server


class SendTestPushNotificationEndpointTest(BouncerTestCase):
    @activate_push_notification_service()
    @responses.activate
    def test_send_test_push_notification_api_invalid_token(self) -> None:
        # What happens when the mobile device isn't registered with its server,
        # and makes a request to this API:
        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, "Device not recognized")
        self.assertEqual(orjson.loads(result.content)["code"], "INVALID_PUSH_DEVICE_TOKEN")

        # What response the server receives when it makes a request to the bouncer
        # to the /test_notification endpoint:
        payload = {
            "realm_uuid": str(user.realm.uuid),
            "user_uuid": str(user.uuid),
            "user_id": user.id,
            "token": "invalid",
            "token_kind": PushDeviceToken.FCM,
            "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, "Device not recognized by the push bouncer")
        self.assertEqual(orjson.loads(result.content)["code"], "INVALID_REMOTE_PUSH_DEVICE_TOKEN")

        # Finally, test the full scenario where the mobile device is registered with its
        # server, but for some reason the server failed to register it with the bouncer.

        token = "111222"
        token_kind = PushDeviceToken.FCM
        # We create a PushDeviceToken object, but no RemotePushDeviceToken object, to simulate
        # a missing registration on the bouncer.
        PushDeviceToken.objects.create(user=user, token=token, kind=token_kind)

        # As verified above, this is the response the server receives from the bouncer in this kind of case.
        # We have to simulate it with a response mock.
        error_response = json_response_from_error(InvalidRemotePushDeviceTokenError())
        responses.add(
            responses.POST,
            f"{settings.ZULIP_SERVICES_URL}/api/v1/remotes/push/test_notification",
            body=error_response.content,
            status=error_response.status_code,
        )

        result = self.api_post(
            user, "/api/v1/mobile_push/test_notification", {"token": token}, subdomain="zulip"
        )
        self.assert_json_error(result, "Device not recognized by the push bouncer")
        self.assertEqual(orjson.loads(result.content)["code"], "INVALID_REMOTE_PUSH_DEVICE_TOKEN")

    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.FCM
        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_name": "Zulip Dev",
            "realm_uri": "http://zulip.testserver",
            "realm_url": "http://zulip.testserver",
            "user_id": user.id,
            "event": "test",
            "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 Zulip Dev (http://zulip.testserver).",
            },
            "sound": "default",
            "custom": {
                "zulip": {
                    "server": "testserver",
                    "realm_id": user.realm_id,
                    "realm_name": "Zulip Dev",
                    "realm_uri": "http://zulip.testserver",
                    "realm_url": "http://zulip.testserver",
                    "user_id": user.id,
                    "event": "test",
                }
            },
        }
        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)

    @activate_push_notification_service()
    @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 = self.server
        remote_realm = RemoteRealm.objects.get(server=server, uuid=user.realm.uuid)

        token = "111222"
        token_kind = PushDeviceToken.FCM
        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_name": "Zulip Dev",
            "realm_uri": "http://zulip.testserver",
            "realm_url": "http://zulip.testserver",
            "user_id": user.id,
            "event": "test",
            "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)

        remote_realm.refresh_from_db()
        self.assertEqual(remote_realm.last_request_datetime, time_now)


class PushBouncerNotificationTest(BouncerTestCase):
    DEFAULT_SUBDOMAIN = ""

    def test_unregister_remote_push_user_params(self) -> None:
        token = "111222"
        token_kind = PushDeviceToken.FCM

        endpoint = "/api/v1/remotes/push/unregister"
        result = self.uuid_post(self.server_uuid, endpoint, {"token_kind": token_kind})
        self.assert_json_error(result, "Missing 'token' argument")
        result = self.uuid_post(self.server_uuid, endpoint, {"token": token})
        self.assert_json_error(result, "Missing 'token_kind' argument")

        # We need the root ('') subdomain to be in use for this next
        # test, since the push bouncer API is only available there:
        hamlet = self.example_user("hamlet")
        realm = get_realm("zulip")
        realm.string_id = ""
        realm.save()

        result = self.api_post(
            hamlet,
            endpoint,
            dict(user_id=15, token=token, token_kind=token_kind),
            subdomain="",
        )
        self.assert_json_error(result, "Must validate with valid Zulip server API key")

        # Try with deactivated remote servers
        self.server.deactivated = True
        self.server.save()
        result = self.uuid_post(self.server_uuid, endpoint, self.get_generic_payload("unregister"))
        self.assert_json_error_contains(
            result,
            "The mobile push notification service registration for your server has been deactivated",
            401,
        )

    def test_register_remote_push_user_params(self) -> None:
        token = "111222"
        user_id = 11
        token_kind = PushDeviceToken.FCM

        endpoint = "/api/v1/remotes/push/register"

        result = self.uuid_post(
            self.server_uuid, endpoint, {"user_id": user_id, "token_kind": token_kind}
        )
        self.assert_json_error(result, "Missing 'token' argument")
        result = self.uuid_post(self.server_uuid, endpoint, {"user_id": user_id, "token": token})
        self.assert_json_error(result, "Missing 'token_kind' argument")
        result = self.uuid_post(
            self.server_uuid, endpoint, {"token": token, "token_kind": token_kind}
        )
        self.assert_json_error(result, "Missing user_id or user_uuid")
        result = self.uuid_post(
            self.server_uuid, endpoint, {"user_id": user_id, "token": token, "token_kind": 17}
        )
        self.assert_json_error(result, "Invalid token type")

        hamlet = self.example_user("hamlet")

        # We need the root ('') subdomain to be in use for this next
        # test, since the push bouncer API is only available there:
        realm = get_realm("zulip")
        realm.string_id = ""
        realm.save()

        result = self.api_post(
            hamlet,
            endpoint,
            dict(user_id=user_id, token_kind=token_kind, token=token),
        )
        self.assert_json_error(result, "Must validate with valid Zulip server API key")

        result = self.uuid_post(
            self.server_uuid,
            endpoint,
            dict(user_id=user_id, token_kind=token_kind, token=token),
            subdomain="zulip",
        )
        self.assert_json_error(
            result, "Invalid subdomain for push notifications bouncer", status_code=401
        )

        # We do a bit of hackery here to the API_KEYS cache just to
        # make the code simple for sending an incorrect API key.
        self.API_KEYS[self.server_uuid] = "invalid"
        result = self.uuid_post(
            self.server_uuid, endpoint, dict(user_id=user_id, token_kind=token_kind, token=token)
        )
        self.assert_json_error(
            result,
            "Zulip server auth failure: key does not match role 6cde5f7a-1f7e-4978-9716-49f69ebfc9fe",
            status_code=401,
        )

        del self.API_KEYS[self.server_uuid]

        self.API_KEYS["invalid_uuid"] = "invalid"
        result = self.uuid_post(
            "invalid_uuid",
            endpoint,
            dict(user_id=user_id, token_kind=token_kind, token=token),
            subdomain="zulip",
        )
        self.assert_json_error(
            result,
            "Zulip server auth failure: invalid_uuid is not registered -- did you run `manage.py register_server`?",
            status_code=401,
        )
        del self.API_KEYS["invalid_uuid"]

        credentials_uuid = str(uuid.uuid4())
        credentials = "{}:{}".format(credentials_uuid, "invalid")
        api_auth = "Basic " + base64.b64encode(credentials.encode()).decode()
        result = self.client_post(
            endpoint,
            {"user_id": user_id, "token_kind": token_kind, "token": token},
            HTTP_AUTHORIZATION=api_auth,
        )
        self.assert_json_error(
            result,
            f"Zulip server auth failure: {credentials_uuid} is not registered -- did you run `manage.py register_server`?",
            status_code=401,
        )

        # Try with deactivated remote servers
        self.server.deactivated = True
        self.server.save()
        result = self.uuid_post(self.server_uuid, endpoint, self.get_generic_payload("register"))
        self.assert_json_error_contains(
            result,
            "The mobile push notification service registration for your server has been deactivated",
            401,
        )

    def test_register_require_ios_app_id(self) -> None:
        endpoint = "/api/v1/remotes/push/register"
        args = {"user_id": 11, "token": "1122"}

        result = self.uuid_post(
            self.server_uuid,
            endpoint,
            {**args, "token_kind": PushDeviceToken.APNS},
        )
        self.assert_json_error(result, "Missing ios_app_id")

        result = self.uuid_post(
            self.server_uuid,
            endpoint,
            {**args, "token_kind": PushDeviceToken.APNS, "ios_app_id": "example.app"},
        )
        self.assert_json_success(result)

        result = self.uuid_post(
            self.server_uuid,
            endpoint,
            {**args, "token_kind": PushDeviceToken.FCM},
        )
        self.assert_json_success(result)

    def test_register_validate_ios_app_id(self) -> None:
        endpoint = "/api/v1/remotes/push/register"
        args = {
            "user_id": 11,
            "token": "1122",
            "token_kind": PushDeviceToken.APNS,
            "ios_app_id": "'; tables --",
        }

        result = self.uuid_post(self.server_uuid, endpoint, args)
        self.assert_json_error(result, "ios_app_id has invalid format")

        args["ios_app_id"] = "com.zulip.apple"
        result = self.uuid_post(self.server_uuid, endpoint, args)
        self.assert_json_success(result)

    def test_register_device_deduplication(self) -> None:
        hamlet = self.example_user("hamlet")
        token = "111222"
        user_id = hamlet.id
        user_uuid = str(hamlet.uuid)
        token_kind = PushDeviceToken.FCM

        endpoint = "/api/v1/remotes/push/register"

        # First we create a legacy user_id registration.
        result = self.uuid_post(
            self.server_uuid,
            endpoint,
            {"user_id": user_id, "token_kind": token_kind, "token": token},
        )
        self.assert_json_success(result)

        registrations = list(RemotePushDeviceToken.objects.filter(token=token))
        self.assert_length(registrations, 1)
        self.assertEqual(registrations[0].user_id, user_id)
        self.assertEqual(registrations[0].user_uuid, None)

        # Register same user+device with uuid now. The old registration should be deleted
        # to avoid duplication.
        result = self.uuid_post(
            self.server_uuid,
            endpoint,
            {"user_id": user_id, "user_uuid": user_uuid, "token_kind": token_kind, "token": token},
        )
        registrations = list(RemotePushDeviceToken.objects.filter(token=token))
        self.assert_length(registrations, 1)
        self.assertEqual(registrations[0].user_id, None)
        self.assertEqual(str(registrations[0].user_uuid), user_uuid)

    def test_remote_push_user_endpoints(self) -> None:
        endpoints = [
            ("/api/v1/remotes/push/register", "register"),
            ("/api/v1/remotes/push/unregister", "unregister"),
        ]

        for endpoint, method in endpoints:
            payload = self.get_generic_payload(method)

            # Verify correct results are success
            result = self.uuid_post(self.server_uuid, endpoint, payload)
            self.assert_json_success(result)

            remote_tokens = RemotePushDeviceToken.objects.filter(token=payload["token"])
            token_count = 1 if method == "register" else 0
            self.assert_length(remote_tokens, token_count)

            # Try adding/removing tokens that are too big...
            broken_token = "x" * 5000  # too big
            payload["token"] = broken_token
            result = self.uuid_post(self.server_uuid, endpoint, payload)
            self.assert_json_error(result, "Empty or invalid length token")

    def test_send_notification_endpoint_sets_remote_realm_for_devices(self) -> None:
        hamlet = self.example_user("hamlet")
        server = self.server

        remote_realm = RemoteRealm.objects.get(server=server, uuid=hamlet.realm.uuid)

        android_token = RemotePushDeviceToken.objects.create(
            kind=RemotePushDeviceToken.FCM,
            token="aaaa",
            user_uuid=hamlet.uuid,
            server=server,
        )
        apple_token = RemotePushDeviceToken.objects.create(
            kind=RemotePushDeviceToken.APNS,
            token="bbbb",
            user_uuid=hamlet.uuid,
            server=server,
        )
        payload = {
            "user_id": hamlet.id,
            "user_uuid": str(hamlet.uuid),
            "realm_uuid": str(hamlet.realm.uuid),
            "gcm_payload": {},
            "apns_payload": {},
            "gcm_options": {},
        }
        with (
            mock.patch("zilencer.views.send_android_push_notification", return_value=1),
            mock.patch("zilencer.views.send_apple_push_notification", return_value=1),
            mock.patch(
                "corporate.lib.stripe.RemoteRealmBillingSession.current_count_for_billed_licenses",
                return_value=10,
            ),
            self.assertLogs("zilencer.views", level="INFO"),
        ):
            result = self.uuid_post(
                self.server_uuid,
                "/api/v1/remotes/push/notify",
                payload,
                content_type="application/json",
            )
        self.assert_json_success(result)

        android_token.refresh_from_db()
        apple_token.refresh_from_db()

        self.assertEqual(android_token.remote_realm, remote_realm)
        self.assertEqual(apple_token.remote_realm, remote_realm)

    def test_send_notification_endpoint(self) -> None:
        hamlet = self.example_user("hamlet")
        server = self.server
        remote_realm = RemoteRealm.objects.get(server=server, uuid=hamlet.realm.uuid)

        token = "aaaa"
        android_tokens = []
        uuid_android_tokens = []
        for i in ["aa", "bb"]:
            android_tokens.append(
                RemotePushDeviceToken.objects.create(
                    kind=RemotePushDeviceToken.FCM,
                    token=token + i,
                    user_id=hamlet.id,
                    server=server,
                )
            )

            # Create a duplicate, newer uuid-based registration for the same user to verify
            # the bouncer will handle that correctly, without triggering a duplicate notification,
            # and will delete the old, legacy registration.
            uuid_android_tokens.append(
                RemotePushDeviceToken.objects.create(
                    kind=RemotePushDeviceToken.FCM,
                    token=token + i,
                    user_uuid=str(hamlet.uuid),
                    server=server,
                )
            )

        apple_token = RemotePushDeviceToken.objects.create(
            kind=RemotePushDeviceToken.APNS,
            token=token,
            user_id=hamlet.id,
            server=server,
        )
        many_ids = ",".join(str(i) for i in range(1, 250))
        payload = {
            "user_id": hamlet.id,
            "user_uuid": str(hamlet.uuid),
            "realm_uuid": str(hamlet.realm.uuid),
            "gcm_payload": {"event": "remove", "zulip_message_ids": many_ids},
            "apns_payload": {
                "badge": 0,
                "custom": {"zulip": {"event": "remove", "zulip_message_ids": many_ids}},
            },
            "gcm_options": {},
        }

        time_sent = now()
        with (
            mock.patch(
                "zilencer.views.send_android_push_notification", return_value=2
            ) as android_push,
            mock.patch("zilencer.views.send_apple_push_notification", return_value=1) as apple_push,
            mock.patch(
                "corporate.lib.stripe.RemoteRealmBillingSession.current_count_for_billed_licenses",
                return_value=10,
            ),
            time_machine.travel(time_sent, tick=False),
            self.assertLogs("zilencer.views", level="INFO") as logger,
        ):
            result = self.uuid_post(
                self.server_uuid,
                "/api/v1/remotes/push/notify",
                payload,
                content_type="application/json",
            )
        data = self.assert_json_success(result)
        self.assertEqual(
            {
                "result": "success",
                "msg": "",
                "total_android_devices": 2,
                "total_apple_devices": 1,
                "deleted_devices": {"android_devices": [], "apple_devices": []},
                "realm": {"can_push": True, "expected_end_timestamp": None},
            },
            data,
        )
        self.assertEqual(
            logger.output,
            [
                "INFO:zilencer.views:"
                f"Deduplicating push registrations for server id:{server.id} user id:{hamlet.id} uuid:{hamlet.uuid} and tokens:{sorted(t.token for t in android_tokens)}",
                "INFO:zilencer.views:"
                f"Sending mobile push notifications for remote user 6cde5f7a-1f7e-4978-9716-49f69ebfc9fe:<id:{hamlet.id}><uuid:{hamlet.uuid}>: "
                "2 via FCM devices, 1 via APNs devices",
            ],
        )

        user_identity = UserPushIdentityCompat(user_id=hamlet.id, user_uuid=str(hamlet.uuid))
        apple_push.assert_called_once_with(
            user_identity,
            [apple_token],
            {
                "badge": 0,
                "custom": {
                    "zulip": {
                        "event": "remove",
                        "zulip_message_ids": ",".join(str(i) for i in range(50, 250)),
                    }
                },
            },
            remote=server,
        )
        android_push.assert_called_once_with(
            user_identity,
            uuid_android_tokens,
            {"event": "remove", "zulip_message_ids": ",".join(str(i) for i in range(50, 250))},
            {},
            remote=server,
        )

        remote_realm.refresh_from_db()
        server.refresh_from_db()
        self.assertEqual(remote_realm.last_request_datetime, time_sent)
        self.assertEqual(server.last_request_datetime, time_sent)

    def test_send_notification_endpoint_on_free_plans(self) -> None:
        hamlet = self.example_user("hamlet")
        remote_server = self.server
        RemotePushDeviceToken.objects.create(
            kind=RemotePushDeviceToken.FCM,
            token="aaaaaa",
            user_id=hamlet.id,
            server=remote_server,
        )

        current_time = now()
        message = Message(
            sender=hamlet,
            recipient=self.example_user("othello").recipient,
            realm_id=hamlet.realm_id,
            content="This is test content",
            rendered_content="This is test content",
            date_sent=current_time,
            sending_client=get_client("test"),
            is_channel_message=False,
        )
        message.save()

        # Test old zulip server case.
        self.assertIsNone(remote_server.last_api_feature_level)
        old_apns_payload = {
            "alert": {
                "title": "King Hamlet",
                "subtitle": "",
                "body": message.content,
            },
            "badge": 0,
            "sound": "default",
            "custom": {
                "zulip": {
                    "message_ids": [message.id],
                    "recipient_type": "private",
                    "sender_email": hamlet.email,
                    "sender_id": hamlet.id,
                    "server": settings.EXTERNAL_HOST,
                    "realm_id": hamlet.realm.id,
                    "realm_uri": hamlet.realm.url,
                    "realm_url": hamlet.realm.url,
                    "user_id": self.example_user("othello").id,
                }
            },
        }
        old_gcm_payload = {
            "user_id": self.example_user("othello").id,
            "event": "message",
            "alert": "New private message from King Hamlet",
            "zulip_message_id": message.id,
            "time": datetime_to_timestamp(message.date_sent),
            "content": message.content,
            "server": settings.EXTERNAL_HOST,
            "realm_id": hamlet.realm.id,
            "realm_uri": hamlet.realm.url,
            "realm_url": hamlet.realm.url,
            "sender_id": hamlet.id,
            "sender_email": hamlet.email,
            "sender_full_name": "King Hamlet",
            "sender_avatar_url": absolute_avatar_url(message.sender),
            "recipient_type": "private",
        }
        payload = {
            "user_id": hamlet.id,
            "gcm_payload": old_gcm_payload,
            "apns_payload": old_apns_payload,
            "gcm_options": {"priority": "high"},
        }

        result = self.uuid_post(
            self.server_uuid,
            "/api/v1/remotes/push/notify",
            payload,
            content_type="application/json",
        )
        self.assertEqual(orjson.loads(result.content)["code"], "INVALID_ZULIP_SERVER")

        remote_server.last_api_feature_level = 235
        remote_server.save()

        message_payload = get_message_payload(hamlet, message)
        gcm_payload, gcm_options = get_message_payload_gcm(message_payload, hamlet, message)
        apns_payload = get_message_payload_apns(
            message_payload, hamlet, message, NotificationTriggers.DIRECT_MESSAGE
        )
        payload = {
            "user_id": hamlet.id,
            "user_uuid": str(hamlet.uuid),
            "gcm_payload": gcm_payload,
            "apns_payload": apns_payload,
            "gcm_options": gcm_options,
        }

        # Test the case when there is no data about users.
        self.assertIsNone(remote_server.last_audit_log_update)
        result = self.uuid_post(
            self.server_uuid,
            "/api/v1/remotes/push/notify",
            payload,
            content_type="application/json",
        )
        self.assert_json_error(
            result,
            "Your plan doesn't allow sending push notifications. Reason provided by the server: Missing data",
        )
        self.assertEqual(orjson.loads(result.content)["code"], "PUSH_NOTIFICATIONS_DISALLOWED")

        human_counts = {
            str(UserProfile.ROLE_REALM_ADMINISTRATOR): 1,
            str(UserProfile.ROLE_REALM_OWNER): 1,
