<commit_before>from contextlib import contextmanager
import datetime
import itertools
import requests
from unittest import mock
from unittest.mock import call
from typing import Any, Dict, List, Optional

import base64
import os
import ujson
import uuid

from django.test import override_settings
from django.conf import settings
from django.http import HttpResponse
from django.db import transaction
from django.db.models import F
from django.utils.crypto import get_random_string

from analytics.lib.counts import CountStat, LoggingCountStat
from analytics.models import InstallationCount, RealmCount
from zerver.models import (
    PushDeviceToken,
    Message,
    UserMessage,
    receives_offline_email_notifications,
    receives_offline_push_notifications,
    receives_online_notifications,
    receives_stream_notifications,
    get_client,
    get_realm,
    get_stream,
    Recipient,
    RealmAuditLog,
    Stream,
    Subscription,
)
from zerver.lib.actions import (
    do_delete_messages,
    do_mark_stream_messages_as_read,
    do_regenerate_api_key,
)
from zerver.lib.soft_deactivation import do_soft_deactivate_users
from zerver.lib.push_notifications import (
    absolute_avatar_url,
    b64_to_hex,
    datetime_to_timestamp,
    DeviceToken,
    get_apns_client,
    get_display_recipient,
    get_message_payload_apns,
    get_message_payload_gcm,
    get_mobile_push_content,
    handle_push_notification,
    handle_remove_push_notification,
    hex_to_b64,
    modernize_apns_payload,
    num_push_devices_for_user,
    parse_gcm_options,
    send_android_push_notification_to_user,
    send_apple_push_notification,
    send_notifications_to_bouncer,
    send_to_push_bouncer,
)
from zerver.lib.remote_server import send_analytics_to_remote_server, \
    build_analytics_data, PushNotificationBouncerException, PushNotificationBouncerRetryLaterError
from zerver.lib.request import JsonableError
from zerver.lib.test_classes import (
    TestCase, ZulipTestCase,
)

from zilencer.models import RemoteZulipServer, RemotePushDeviceToken, \
    RemoteRealmCount, RemoteInstallationCount, RemoteRealmAuditLog
from django.utils.timezone import now

ZERVER_DIR = os.path.dirname(os.path.dirname(__file__))

class BouncerTestCase(ZulipTestCase):
    def setUp(self) -> None:
        self.server_uuid = "1234-abcd"
        server = RemoteZulipServer(uuid=self.server_uuid,
                                   api_key="magic_secret_api_key",
                                   hostname="demo.example.com",
                                   last_updated=now())
        server.save()
        super().setUp()

    def tearDown(self) -> None:
        RemoteZulipServer.objects.filter(uuid=self.server_uuid).delete()
        super().tearDown()

    def bounce_request(self, *args: Any, **kwargs: Any) -> HttpResponse:
        """This method is used to carry out the push notification bouncer
        requests using the Django test browser, rather than python-requests.
        """
        # args[0] is method, args[1] is URL.
        local_url = args[1].replace(settings.PUSH_NOTIFICATION_BOUNCER_URL, "")
        if args[0] == "POST":
            result = self.uuid_post(
                self.server_uuid,
                local_url,
                kwargs['data'],
                subdomain='')

        elif args[0] == "GET":
            result = self.uuid_get(
                self.server_uuid,
                local_url,
                kwargs['data'],
                subdomain='')
        else:
            raise AssertionError("Unsupported method for bounce_request")
        return result

    def get_generic_payload(self, method: str='register') -> Dict[str, Any]:
        user_id = 10
        token = "111222"
        token_kind = PushDeviceToken.GCM

        return {'user_id': user_id,
                'token': token,
                'token_kind': token_kind}

class PushBouncerNotificationTest(BouncerTestCase):
    DEFAULT_SUBDOMAIN = ""

    def test_unregister_remote_push_user_params(self) -> None:
        token = "1<PASSWORD>"
        token_kind = PushDeviceToken.GCM

        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")

    def test_register_remote_push_user_paramas(self) -> None:
        token = "1<PASSWORD>"
        user_id = 11
        token_kind = PushDeviceToken.GCM

        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' argument")
        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')

        result = self.api_post(
            hamlet,
            endpoint,
            dict(user_id=user_id, token_kin=token_kind, token=token),
        )
        self.assert_json_error(result, "Account is not associated with this subdomain",
                               status_code=401)

        # 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 1234-abcd",
                               status_code=401)

        del self.API_KEYS[self.server_uuid]

        credentials = "%s:%s" % ("5<PASSWORD>-efgh", 'invalid')
        api_auth = 'Basic ' + base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
        result = self.client_post(endpoint, {'user_id': user_id,
                                             'token_kind': token_kind,
                                             'token': token},
                                  HTTP_AUTHORIZATION = api_auth)
        self.assert_json_error(result, "Zulip server auth failure: 5678-efgh is not registered",
                               status_code=401)

    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.assertEqual(len(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_remote_push_unregister_all(self) -> None:
        payload = self.get_generic_payload('register')

        # Verify correct results are success
        result = self.uuid_post(self.server_uuid,
                                '/api/v1/remotes/push/register', payload)
        self.assert_json_success(result)

        remote_tokens = RemotePushDeviceToken.objects.filter(token=payload['token'])
        self.assertEqual(len(remote_tokens), 1)
        result = self.uuid_post(self.server_uuid,
                                '/api/v1/remotes/push/unregister/all',
                                dict(user_id=10))
        self.assert_json_success(result)

        remote_tokens = RemotePushDeviceToken.objects.filter(token=payload['token'])
        self.assertEqual(len(remote_tokens), 0)

    def test_invalid_apns_token(self) -> None:
        endpoints = [
            ('/api/v1/remotes/push/register', 'apple-token'),
        ]

        for endpoint, method in endpoints:
            payload = {
                'user_id': 10,
                'token': '<PASSWORD>',
                'token_kind': PushDeviceToken.APNS,
            }
            result = self.uuid_post(self.server_uuid, endpoint, payload)
            self.assert_json_error(result, 'Invalid APNS token')

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_push_bouncer_api(self, mock_request: Any) -> None:
        """This is a variant of the below test_push_api, but using the full
        push notification bouncer flow
        """
        mock_request.side_effect = self.bounce_request
        user = self.example_user('cordelia')
        self.login_user(user)
        server = RemoteZulipServer.objects.get(uuid=self.server_uuid)

        endpoints = [
            ('/json/users/me/apns_device_token', 'apple-tokenaz', RemotePushDeviceToken.APNS),
            ('/json/users/me/android_gcm_reg_id', 'android-token', RemotePushDeviceToken.GCM),
        ]

        # Test error handling
        for endpoint, token, kind in endpoints:
            # Try adding/removing tokens that are too big...
            broken_token = "a" * 5000  # too big
            result = self.client_post(endpoint, {'token': broken_token,
                                                 'token_kind': kind},
                                      subdomain="zulip")
            self.assert_json_error(result, 'Empty or invalid length token')

            result = self.client_delete(endpoint, {'token': broken_token,
                                                   'token_kind': kind},
                                        subdomain="zulip")
            self.assert_json_error(result, 'Empty or invalid length token')

            # Try to remove a non-existent token...
            result = self.client_delete(endpoint, {'token': '<KEY>',
                                                   'token_kind': kind},
                                        subdomain="zulip")
            self.assert_json_error(result, 'Token does not exist')

            with mock.patch('zerver.lib.remote_server.requests.request',
                            side_effect=requests.ConnectionError):
                result = self.client_post(endpoint, {'token': token},
                                          subdomain="zulip")
                self.assert_json_error(
                    result, "ConnectionError while trying to connect to push notification bouncer", 502)

            with mock.patch('zerver.lib.remote_server.requests.request',
                            return_value=Result(status=500)):
                result = self.client_post(endpoint, {'token': token},
                                          subdomain="zulip")
                self.assert_json_error(result, "Received 500 from push notification bouncer", 502)

        # Add tokens
        for endpoint, token, kind in endpoints:
            # Test that we can push twice
            result = self.client_post(endpoint, {'token': token},
                                      subdomain="zulip")
            self.assert_json_success(result)

            result = self.client_post(endpoint, {'token': token},
                                      subdomain="zulip")
            self.assert_json_success(result)

            tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id, token=token,
                                                               server=server))
            self.assertEqual(len(tokens), 1)
            self.assertEqual(tokens[0].token, token)

        # User should have tokens for both devices now.
        tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                           server=server))
        self.assertEqual(len(tokens), 2)

        # Remove tokens
        for endpoint, token, kind in endpoints:
            result = self.client_delete(endpoint, {'token': token,
                                                   'token_kind': kind},
                                        subdomain="zulip")
            self.assert_json_success(result)
            tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id, token=token,
                                                               server=server))
            self.assertEqual(len(tokens), 0)

        # Re-add copies of those tokens
        for endpoint, token, kind in endpoints:
            result = self.client_post(endpoint, {'token': token},
                                      subdomain="zulip")
            self.assert_json_success(result)
        tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                           server=server))
        self.assertEqual(len(tokens), 2)

        # Now we want to remove them using the bouncer after an API key change.
        # First we test error handling in case of issues with the bouncer:
        with mock.patch('zerver.worker.queue_processors.clear_push_device_tokens',
                        side_effect=PushNotificationBouncerRetryLaterError("test")), \
                mock.patch('zerver.worker.queue_processors.retry_event') as mock_retry:
            do_regenerate_api_key(user, user)
            mock_retry.assert_called()

            # We didn't manage to communicate with the bouncer, to the tokens are still there:
            tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                               server=server))
            self.assertEqual(len(tokens), 2)

        # Now we successfully remove them:
        do_regenerate_api_key(user, user)
        tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                           server=server))
        self.assertEqual(len(tokens), 0)

class AnalyticsBouncerTest(BouncerTestCase):
    TIME_ZERO = datetime.datetime(1988, 3, 14, tzinfo=datetime.timezone.utc)

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_analytics_api(self, mock_request: Any) -> None:
        """This is a variant of the below test_push_api, but using the full
        push notification bouncer flow
        """
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        end_time = self.TIME_ZERO

        with mock.patch('zerver.lib.remote_server.requests.request',
                        side_effect=requests.ConnectionError), \
                mock.patch('zerver.lib.remote_server.logging.warning') as mock_warning:
            send_analytics_to_remote_server()
            mock_warning.assert_called_once_with(
                "ConnectionError while trying to connect to push notification bouncer")

        # Send any existing data over, so that we can start the test with a "clean" slate
        audit_log_max_id = RealmAuditLog.objects.all().order_by('id').last().id
        send_analytics_to_remote_server()
        self.assertEqual(mock_request.call_count, 2)
        remote_audit_log_count = RemoteRealmAuditLog.objects.count()
        self.assertEqual(RemoteRealmCount.objects.count(), 0)
        self.assertEqual(RemoteInstallationCount.objects.count(), 0)

        def check_counts(mock_request_call_count: int, remote_realm_count: int,
                         remote_installation_count: int, remote_realm_audit_log: int) -> None:
            self.assertEqual(mock_request.call_count, mock_request_call_count)
            self.assertEqual(RemoteRealmCount.objects.count(), remote_realm_count)
            self.assertEqual(RemoteInstallationCount.objects.count(), remote_installation_count)
            self.assertEqual(RemoteRealmAuditLog.objects.count(),
                             remote_audit_log_count + remote_realm_audit_log)

        # Create some rows we'll send to remote server
        realm_stat = LoggingCountStat('invites_sent::day', RealmCount, CountStat.DAY)
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time, value=5)
        InstallationCount.objects.create(
            property=realm_stat.property, end_time=end_time, value=5,
            # We set a subgroup here to work around:
            # https://github.com/zulip/zulip/issues/12362
            subgroup="test_subgroup")
        # Event type in SYNCED_BILLING_EVENTS -- should be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.USER_CREATED,
            event_time=end_time, extra_data='data')
        # Event type not in SYNCED_BILLING_EVENTS -- should not be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.REALM_LOGO_CHANGED,
            event_time=end_time, extra_data='data')
        self.assertEqual(RealmCount.objects.count(), 1)
        self.assertEqual(InstallationCount.objects.count(), 1)
        self.assertEqual(RealmAuditLog.objects.filter(id__gt=audit_log_max_id).count(), 2)

        send_analytics_to_remote_server()
        check_counts(4, 1, 1, 1)

        # Test having no new rows
        send_analytics_to_remote_server()
        check_counts(5, 1, 1, 1)

        # Test only having new RealmCount rows
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time + datetime.timedelta(days=1), value=6)
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time + datetime.timedelta(days=2), value=9)
        send_analytics_to_remote_server()
        check_counts(7, 3, 1, 1)

        # Test only having new InstallationCount rows
        InstallationCount.objects.create(
            property=realm_stat.property, end_time=end_time + datetime.timedelta(days=1), value=6)
        send_analytics_to_remote_server()
        check_counts(9, 3, 2, 1)

        # Test only having new RealmAuditLog rows
        # Non-synced event
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.REALM_LOGO_CHANGED,
            event_time=end_time, extra_data='data')
        send_analytics_to_remote_server()
        check_counts(10, 3, 2, 1)
        # Synced event
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.USER_REACTIVATED,
            event_time=end_time, extra_data='data')
        send_analytics_to_remote_server()
        check_counts(12, 3, 2, 2)

        (realm_count_data,
         installation_count_data,
         realmauditlog_data) = build_analytics_data(RealmCount.objects.all(),
                                                    InstallationCount.objects.all(),
                                                    RealmAuditLog.objects.all())
        result = self.uuid_post(
            self.server_uuid,
            '/api/v1/remotes/server/analytics',
            {'realm_counts': ujson.dumps(realm_count_data),
             'installation_counts': ujson.dumps(installation_count_data),
             'realmauditlog_rows': ujson.dumps(realmauditlog_data)},
            subdomain="")
        self.assert_json_error(result, "Data is out of order.")

        with mock.patch("zilencer.views.validate_incoming_table_data"):
            # We need to wrap a transaction here to avoid the
            # IntegrityError that will be thrown in here from breaking
            # the unittest transaction.
            with transaction.atomic():
                result = self.uuid_post(
                    self.server_uuid,
                    '/api/v1/remotes/server/analytics',
                    {'realm_counts': ujson.dumps(realm_count_data),
                     'installation_counts': ujson.dumps(installation_count_data),
                     'realmauditlog_rows': ujson.dumps(realmauditlog_data)},
                    subdomain="")
            self.assert_json_error(result, "Invalid data.")

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_analytics_api_invalid(self, mock_request: Any) -> None:
        """This is a variant of the below test_push_api, but using the full
        push notification bouncer flow
        """
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        end_time = self.TIME_ZERO

        realm_stat = LoggingCountStat('invalid count stat', RealmCount, CountStat.DAY)
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time, value=5)

        self.assertEqual(RealmCount.objects.count(), 1)

        self.assertEqual(RemoteRealmCount.objects.count(), 0)
        with mock.patch('zerver.lib.remote_server.logging.warning') as log_warning:
            send_analytics_to_remote_server()
            log_warning.assert_called_once()
        self.assertEqual(RemoteRealmCount.objects.count(), 0)

    # Servers on Zulip 2.0.6 and earlier only send realm_counts and installation_counts data,
    # and don't send realmauditlog_rows. Make sure that continues to work.
    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_old_two_table_format(self, mock_request: Any) -> None:
        mock_request.side_effect = self.bounce_request
        # Send fixture generated with Zulip 2.0 code
        send_to_push_bouncer('POST', 'server/analytics', {
            'realm_counts': '[{"id":1,"property":"invites_sent::day","subgroup":null,"end_time":574300800.0,"value":5,"realm":2}]',  # lint:ignore
            'installation_counts': '[]',
            'version': '"2.0.6+git"'})
        self.assertEqual(mock_request.call_count, 1)
        self.assertEqual(RemoteRealmCount.objects.count(), 1)
        self.assertEqual(RemoteInstallationCount.objects.count(), 0)
        self.assertEqual(RemoteRealmAuditLog.objects.count(), 0)

    # Make sure we aren't sending data we don't mean to, even if we don't store it.
    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_only_sending_intended_realmauditlog_data(self, mock_request: Any) -> None:
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        # Event type in SYNCED_BILLING_EVENTS -- should be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.USER_REACTIVATED,
            event_time=self.TIME_ZERO, extra_data='data')
        # Event type not in SYNCED_BILLING_EVENTS -- should not be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.REALM_LOGO_CHANGED,
            event_time=self.TIME_ZERO, extra_data='data')

        def check_for_unwanted_data(*args: Any) -> Any:
            if check_for_unwanted_data.first_call:  # type: ignore[attr-defined]
                check_for_unwanted_data.first_call = False  # type: ignore[attr-defined]
            else:
                # Test that we're respecting SYNCED_BILLING_EVENTS
                self.assertIn('"event_type":{}'.format(RealmAuditLog.USER_REACTIVATED), str(args))
                self.assertNotIn('"event_type":{}'.format(RealmAuditLog.REALM_LOGO_CHANGED), str(args))
                # Test that we're respecting REALMAUDITLOG_PUSHED_FIELDS
                self.assertIn('backfilled', str(args))
                self.assertNotIn('modified_user', str(args))
            return send_to_push_bouncer(*args)

        # send_analytics_to_remote_server calls send_to_push_bouncer twice.
        # We need to distinguish the first and second calls.
        check_for_unwanted_data.first_call = True  # type: ignore[attr-defined]
        with mock.patch('zerver.lib.remote_server.send_to_push_bouncer',
                        side_effect=check_for_unwanted_data):
            send_analytics_to_remote_server()

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_realmauditlog_data_mapping(self, mock_request: Any) -> None:
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        log_entry = RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, backfilled=True,
            event_type=RealmAuditLog.USER_REACTIVATED, event_time=self.TIME_ZERO, extra_data='data')
        send_analytics_to_remote_server()
        remote_log_entry = RemoteRealmAuditLog.objects.order_by('id').last()
        self.assertEqual(remote_log_entry.server.uuid, self.server_uuid)
        self.assertEqual(remote_log_entry.remote_id, log_entry.id)
        self.assertEqual(remote_log_entry.event_time, self.TIME_ZERO)
        self.assertEqual(remote_log_entry.backfilled, True)
        self.assertEqual(remote_log_entry.extra_data, 'data')
        self.assertEqual(remote_log_entry.event_type, RealmAuditLog.USER_REACTIVATED)

class PushNotificationTest(BouncerTestCase):
    def setUp(self) -> None:
        super().setUp()
        self.user_profile = self.example_user('hamlet')
        self.sending_client = get_client('test')
        self.sender = self.example_user('hamlet')

    def get_message(self, type: int, type_id: int=100) -> Message:
        recipient, _ = Recipient.objects.get_or_create(
            type_id=type_id,
            type=type,
        )

        message = Message(
            sender=self.sender,
            recipient=recipient,
            content='This is test content',
            rendered_content='This is test content',
            date_sent=now(),
            sending_client=self.sending_client,
        )
        message.set_topic_name('Test Topic')
        message.save()

        return message

    @contextmanager
    def mock_apns(self) -> mock.MagicMock:
        mock_apns = mock.Mock()
        with mock.patch('zerver.lib.push_notifications.get_apns_client') as mock_get:
            mock_get.return_value = mock_apns
            yield mock_apns

    def setup_apns_tokens(self) -> None:
        self.tokens = ['aaaa', 'bbbb']
        for token in self.tokens:
            PushDeviceToken.objects.create(
                kind=PushDeviceToken.APNS,
                token=hex_to_b64(token),
                user=self.user_profile,
                ios_app_id=settings.ZULIP_IOS_APP_ID)

        self.remote_tokens = ['cccc']
        for token in self.remote_tokens:
            RemotePushDeviceToken.objects.create(
                kind=RemotePushDeviceToken.APNS,
                token=hex_to_b64(token),
                user_id=self.user_profile.id,
                server=RemoteZulipServer.objects.get(uuid=self.server_uuid),
            )

    def setup_gcm_tokens(self) -> None:
        self.gcm_tokens = ['<PASSWORD>', '<PASSWORD>']
        for token in self.gcm_tokens:
            PushDeviceToken.objects.create(
                kind=PushDeviceToken.GCM,
                token=hex_to_b64(token),
                user=self.user_profile,<commit_msg>test_push_notifications: Fix mock_apns type.<commit_after>from contextlib import contextmanager
import datetime
import itertools
import requests
from unittest import mock
from unittest.mock import call
from typing import Any, Dict, Iterator, List, Optional

import base64
import os
import ujson
import uuid

from django.test import override_settings
from django.conf import settings
from django.http import HttpResponse
from django.db import transaction
from django.db.models import F
from django.utils.crypto import get_random_string

from analytics.lib.counts import CountStat, LoggingCountStat
from analytics.models import InstallationCount, RealmCount
from zerver.models import (
    PushDeviceToken,
    Message,
    UserMessage,
    receives_offline_email_notifications,
    receives_offline_push_notifications,
    receives_online_notifications,
    receives_stream_notifications,
    get_client,
    get_realm,
    get_stream,
    Recipient,
    RealmAuditLog,
    Stream,
    Subscription,
)
from zerver.lib.actions import (
    do_delete_messages,
    do_mark_stream_messages_as_read,
    do_regenerate_api_key,
)
from zerver.lib.soft_deactivation import do_soft_deactivate_users
from zerver.lib.push_notifications import (
    absolute_avatar_url,
    b64_to_hex,
    datetime_to_timestamp,
    DeviceToken,
    get_apns_client,
    get_display_recipient,
    get_message_payload_apns,
    get_message_payload_gcm,
    get_mobile_push_content,
    handle_push_notification,
    handle_remove_push_notification,
    hex_to_b64,
    modernize_apns_payload,
    num_push_devices_for_user,
    parse_gcm_options,
    send_android_push_notification_to_user,
    send_apple_push_notification,
    send_notifications_to_bouncer,
    send_to_push_bouncer,
)
from zerver.lib.remote_server import send_analytics_to_remote_server, \
    build_analytics_data, PushNotificationBouncerException, PushNotificationBouncerRetryLaterError
from zerver.lib.request import JsonableError
from zerver.lib.test_classes import (
    TestCase, ZulipTestCase,
)

from zilencer.models import RemoteZulipServer, RemotePushDeviceToken, \
    RemoteRealmCount, RemoteInstallationCount, RemoteRealmAuditLog
from django.utils.timezone import now

ZERVER_DIR = os.path.dirname(os.path.dirname(__file__))

class BouncerTestCase(ZulipTestCase):
    def setUp(self) -> None:
        self.server_uuid = "1234-abcd"
        server = RemoteZulipServer(uuid=self.server_uuid,
                                   api_key="magic_secret_api_key",
                                   hostname="demo.example.com",
                                   last_updated=now())
        server.save()
        super().setUp()

    def tearDown(self) -> None:
        RemoteZulipServer.objects.filter(uuid=self.server_uuid).delete()
        super().tearDown()

    def bounce_request(self, *args: Any, **kwargs: Any) -> HttpResponse:
        """This method is used to carry out the push notification bouncer
        requests using the Django test browser, rather than python-requests.
        """
        # args[0] is method, args[1] is URL.
        local_url = args[1].replace(settings.PUSH_NOTIFICATION_BOUNCER_URL, "")
        if args[0] == "POST":
            result = self.uuid_post(
                self.server_uuid,
                local_url,
                kwargs['data'],
                subdomain='')

        elif args[0] == "GET":
            result = self.uuid_get(
                self.server_uuid,
                local_url,
                kwargs['data'],
                subdomain='')
        else:
            raise AssertionError("Unsupported method for bounce_request")
        return result

    def get_generic_payload(self, method: str='register') -> Dict[str, Any]:
        user_id = 10
        token = "1<PASSWORD>"
        token_kind = PushDeviceToken.GCM

        return {'user_id': user_id,
                'token': token,
                'token_kind': token_kind}

class PushBouncerNotificationTest(BouncerTestCase):
    DEFAULT_SUBDOMAIN = ""

    def test_unregister_remote_push_user_params(self) -> None:
        token = "1<PASSWORD>"
        token_kind = PushDeviceToken.GCM

        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")

    def test_register_remote_push_user_paramas(self) -> None:
        token = "<PASSWORD>"
        user_id = 11
        token_kind = PushDeviceToken.GCM

        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' argument")
        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')

        result = self.api_post(
            hamlet,
            endpoint,
            dict(user_id=user_id, token_kin=token_kind, token=token),
        )
        self.assert_json_error(result, "Account is not associated with this subdomain",
                               status_code=401)

        # 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 1234-abcd",
                               status_code=401)

        del self.API_KEYS[self.server_uuid]

        credentials = "%s:%s" % ("<PASSWORD>", 'invalid')
        api_auth = 'Basic ' + base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
        result = self.client_post(endpoint, {'user_id': user_id,
                                             'token_kind': token_kind,
                                             'token': token},
                                  HTTP_AUTHORIZATION = api_auth)
        self.assert_json_error(result, "Zulip server auth failure: 5678-efgh is not registered",
                               status_code=401)

    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.assertEqual(len(remote_tokens), token_count)

            # Try adding/removing tokens that are too big...
            broken_token = "x" * <PASSWORD>  # 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_remote_push_unregister_all(self) -> None:
        payload = self.get_generic_payload('register')

        # Verify correct results are success
        result = self.uuid_post(self.server_uuid,
                                '/api/v1/remotes/push/register', payload)
        self.assert_json_success(result)

        remote_tokens = RemotePushDeviceToken.objects.filter(token=payload['token'])
        self.assertEqual(len(remote_tokens), 1)
        result = self.uuid_post(self.server_uuid,
                                '/api/v1/remotes/push/unregister/all',
                                dict(user_id=10))
        self.assert_json_success(result)

        remote_tokens = RemotePushDeviceToken.objects.filter(token=payload['token'])
        self.assertEqual(len(remote_tokens), 0)

    def test_invalid_apns_token(self) -> None:
        endpoints = [
            ('/api/v1/remotes/push/register', 'apple-token'),
        ]

        for endpoint, method in endpoints:
            payload = {
                'user_id': 10,
                'token': '<PASSWORD>',
                'token_kind': PushDeviceToken.APNS,
            }
            result = self.uuid_post(self.server_uuid, endpoint, payload)
            self.assert_json_error(result, 'Invalid APNS token')

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_push_bouncer_api(self, mock_request: Any) -> None:
        """This is a variant of the below test_push_api, but using the full
        push notification bouncer flow
        """
        mock_request.side_effect = self.bounce_request
        user = self.example_user('cordelia')
        self.login_user(user)
        server = RemoteZulipServer.objects.get(uuid=self.server_uuid)

        endpoints = [
            ('/json/users/me/apns_device_token', 'apple-tokenaz', RemotePushDeviceToken.APNS),
            ('/json/users/me/android_gcm_reg_id', 'android-token', RemotePushDeviceToken.GCM),
        ]

        # Test error handling
        for endpoint, token, kind in endpoints:
            # Try adding/removing tokens that are too big...
            broken_token = "a" * 5000  # too big
            result = self.client_post(endpoint, {'token': broken_token,
                                                 'token_kind': kind},
                                      subdomain="zulip")
            self.assert_json_error(result, 'Empty or invalid length token')

            result = self.client_delete(endpoint, {'token': broken_token,
                                                   'token_kind': kind},
                                        subdomain="zulip")
            self.assert_json_error(result, 'Empty or invalid length token')

            # Try to remove a non-existent token...
            result = self.client_delete(endpoint, {'token': '<KEY>',
                                                   'token_kind': kind},
                                        subdomain="zulip")
            self.assert_json_error(result, 'Token does not exist')

            with mock.patch('zerver.lib.remote_server.requests.request',
                            side_effect=requests.ConnectionError):
                result = self.client_post(endpoint, {'token': token},
                                          subdomain="zulip")
                self.assert_json_error(
                    result, "ConnectionError while trying to connect to push notification bouncer", 502)

            with mock.patch('zerver.lib.remote_server.requests.request',
                            return_value=Result(status=500)):
                result = self.client_post(endpoint, {'token': token},
                                          subdomain="zulip")
                self.assert_json_error(result, "Received 500 from push notification bouncer", 502)

        # Add tokens
        for endpoint, token, kind in endpoints:
            # Test that we can push twice
            result = self.client_post(endpoint, {'token': token},
                                      subdomain="zulip")
            self.assert_json_success(result)

            result = self.client_post(endpoint, {'token': token},
                                      subdomain="zulip")
            self.assert_json_success(result)

            tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id, token=token,
                                                               server=server))
            self.assertEqual(len(tokens), 1)
            self.assertEqual(tokens[0].token, token)

        # User should have tokens for both devices now.
        tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                           server=server))
        self.assertEqual(len(tokens), 2)

        # Remove tokens
        for endpoint, token, kind in endpoints:
            result = self.client_delete(endpoint, {'token': token,
                                                   'token_kind': kind},
                                        subdomain="zulip")
            self.assert_json_success(result)
            tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id, token=token,
                                                               server=server))
            self.assertEqual(len(tokens), 0)

        # Re-add copies of those tokens
        for endpoint, token, kind in endpoints:
            result = self.client_post(endpoint, {'token': token},
                                      subdomain="zulip")
            self.assert_json_success(result)
        tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                           server=server))
        self.assertEqual(len(tokens), 2)

        # Now we want to remove them using the bouncer after an API key change.
        # First we test error handling in case of issues with the bouncer:
        with mock.patch('zerver.worker.queue_processors.clear_push_device_tokens',
                        side_effect=PushNotificationBouncerRetryLaterError("test")), \
                mock.patch('zerver.worker.queue_processors.retry_event') as mock_retry:
            do_regenerate_api_key(user, user)
            mock_retry.assert_called()

            # We didn't manage to communicate with the bouncer, to the tokens are still there:
            tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                               server=server))
            self.assertEqual(len(tokens), 2)

        # Now we successfully remove them:
        do_regenerate_api_key(user, user)
        tokens = list(RemotePushDeviceToken.objects.filter(user_id=user.id,
                                                           server=server))
        self.assertEqual(len(tokens), 0)

class AnalyticsBouncerTest(BouncerTestCase):
    TIME_ZERO = datetime.datetime(1988, 3, 14, tzinfo=datetime.timezone.utc)

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_analytics_api(self, mock_request: Any) -> None:
        """This is a variant of the below test_push_api, but using the full
        push notification bouncer flow
        """
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        end_time = self.TIME_ZERO

        with mock.patch('zerver.lib.remote_server.requests.request',
                        side_effect=requests.ConnectionError), \
                mock.patch('zerver.lib.remote_server.logging.warning') as mock_warning:
            send_analytics_to_remote_server()
            mock_warning.assert_called_once_with(
                "ConnectionError while trying to connect to push notification bouncer")

        # Send any existing data over, so that we can start the test with a "clean" slate
        audit_log_max_id = RealmAuditLog.objects.all().order_by('id').last().id
        send_analytics_to_remote_server()
        self.assertEqual(mock_request.call_count, 2)
        remote_audit_log_count = RemoteRealmAuditLog.objects.count()
        self.assertEqual(RemoteRealmCount.objects.count(), 0)
        self.assertEqual(RemoteInstallationCount.objects.count(), 0)

        def check_counts(mock_request_call_count: int, remote_realm_count: int,
                         remote_installation_count: int, remote_realm_audit_log: int) -> None:
            self.assertEqual(mock_request.call_count, mock_request_call_count)
            self.assertEqual(RemoteRealmCount.objects.count(), remote_realm_count)
            self.assertEqual(RemoteInstallationCount.objects.count(), remote_installation_count)
            self.assertEqual(RemoteRealmAuditLog.objects.count(),
                             remote_audit_log_count + remote_realm_audit_log)

        # Create some rows we'll send to remote server
        realm_stat = LoggingCountStat('invites_sent::day', RealmCount, CountStat.DAY)
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time, value=5)
        InstallationCount.objects.create(
            property=realm_stat.property, end_time=end_time, value=5,
            # We set a subgroup here to work around:
            # https://github.com/zulip/zulip/issues/12362
            subgroup="test_subgroup")
        # Event type in SYNCED_BILLING_EVENTS -- should be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.USER_CREATED,
            event_time=end_time, extra_data='data')
        # Event type not in SYNCED_BILLING_EVENTS -- should not be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.REALM_LOGO_CHANGED,
            event_time=end_time, extra_data='data')
        self.assertEqual(RealmCount.objects.count(), 1)
        self.assertEqual(InstallationCount.objects.count(), 1)
        self.assertEqual(RealmAuditLog.objects.filter(id__gt=audit_log_max_id).count(), 2)

        send_analytics_to_remote_server()
        check_counts(4, 1, 1, 1)

        # Test having no new rows
        send_analytics_to_remote_server()
        check_counts(5, 1, 1, 1)

        # Test only having new RealmCount rows
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time + datetime.timedelta(days=1), value=6)
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time + datetime.timedelta(days=2), value=9)
        send_analytics_to_remote_server()
        check_counts(7, 3, 1, 1)

        # Test only having new InstallationCount rows
        InstallationCount.objects.create(
            property=realm_stat.property, end_time=end_time + datetime.timedelta(days=1), value=6)
        send_analytics_to_remote_server()
        check_counts(9, 3, 2, 1)

        # Test only having new RealmAuditLog rows
        # Non-synced event
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.REALM_LOGO_CHANGED,
            event_time=end_time, extra_data='data')
        send_analytics_to_remote_server()
        check_counts(10, 3, 2, 1)
        # Synced event
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.USER_REACTIVATED,
            event_time=end_time, extra_data='data')
        send_analytics_to_remote_server()
        check_counts(12, 3, 2, 2)

        (realm_count_data,
         installation_count_data,
         realmauditlog_data) = build_analytics_data(RealmCount.objects.all(),
                                                    InstallationCount.objects.all(),
                                                    RealmAuditLog.objects.all())
        result = self.uuid_post(
            self.server_uuid,
            '/api/v1/remotes/server/analytics',
            {'realm_counts': ujson.dumps(realm_count_data),
             'installation_counts': ujson.dumps(installation_count_data),
             'realmauditlog_rows': ujson.dumps(realmauditlog_data)},
            subdomain="")
        self.assert_json_error(result, "Data is out of order.")

        with mock.patch("zilencer.views.validate_incoming_table_data"):
            # We need to wrap a transaction here to avoid the
            # IntegrityError that will be thrown in here from breaking
            # the unittest transaction.
            with transaction.atomic():
                result = self.uuid_post(
                    self.server_uuid,
                    '/api/v1/remotes/server/analytics',
                    {'realm_counts': ujson.dumps(realm_count_data),
                     'installation_counts': ujson.dumps(installation_count_data),
                     'realmauditlog_rows': ujson.dumps(realmauditlog_data)},
                    subdomain="")
            self.assert_json_error(result, "Invalid data.")

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_analytics_api_invalid(self, mock_request: Any) -> None:
        """This is a variant of the below test_push_api, but using the full
        push notification bouncer flow
        """
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        end_time = self.TIME_ZERO

        realm_stat = LoggingCountStat('invalid count stat', RealmCount, CountStat.DAY)
        RealmCount.objects.create(
            realm=user.realm, property=realm_stat.property, end_time=end_time, value=5)

        self.assertEqual(RealmCount.objects.count(), 1)

        self.assertEqual(RemoteRealmCount.objects.count(), 0)
        with mock.patch('zerver.lib.remote_server.logging.warning') as log_warning:
            send_analytics_to_remote_server()
            log_warning.assert_called_once()
        self.assertEqual(RemoteRealmCount.objects.count(), 0)

    # Servers on Zulip 2.0.6 and earlier only send realm_counts and installation_counts data,
    # and don't send realmauditlog_rows. Make sure that continues to work.
    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_old_two_table_format(self, mock_request: Any) -> None:
        mock_request.side_effect = self.bounce_request
        # Send fixture generated with Zulip 2.0 code
        send_to_push_bouncer('POST', 'server/analytics', {
            'realm_counts': '[{"id":1,"property":"invites_sent::day","subgroup":null,"end_time":574300800.0,"value":5,"realm":2}]',  # lint:ignore
            'installation_counts': '[]',
            'version': '"2.0.6+git"'})
        self.assertEqual(mock_request.call_count, 1)
        self.assertEqual(RemoteRealmCount.objects.count(), 1)
        self.assertEqual(RemoteInstallationCount.objects.count(), 0)
        self.assertEqual(RemoteRealmAuditLog.objects.count(), 0)

    # Make sure we aren't sending data we don't mean to, even if we don't store it.
    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_only_sending_intended_realmauditlog_data(self, mock_request: Any) -> None:
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        # Event type in SYNCED_BILLING_EVENTS -- should be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.USER_REACTIVATED,
            event_time=self.TIME_ZERO, extra_data='data')
        # Event type not in SYNCED_BILLING_EVENTS -- should not be included
        RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, event_type=RealmAuditLog.REALM_LOGO_CHANGED,
            event_time=self.TIME_ZERO, extra_data='data')

        def check_for_unwanted_data(*args: Any) -> Any:
            if check_for_unwanted_data.first_call:  # type: ignore[attr-defined]
                check_for_unwanted_data.first_call = False  # type: ignore[attr-defined]
            else:
                # Test that we're respecting SYNCED_BILLING_EVENTS
                self.assertIn('"event_type":{}'.format(RealmAuditLog.USER_REACTIVATED), str(args))
                self.assertNotIn('"event_type":{}'.format(RealmAuditLog.REALM_LOGO_CHANGED), str(args))
                # Test that we're respecting REALMAUDITLOG_PUSHED_FIELDS
                self.assertIn('backfilled', str(args))
                self.assertNotIn('modified_user', str(args))
            return send_to_push_bouncer(*args)

        # send_analytics_to_remote_server calls send_to_push_bouncer twice.
        # We need to distinguish the first and second calls.
        check_for_unwanted_data.first_call = True  # type: ignore[attr-defined]
        with mock.patch('zerver.lib.remote_server.send_to_push_bouncer',
                        side_effect=check_for_unwanted_data):
            send_analytics_to_remote_server()

    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL='https://push.zulip.org.example.com')
    @mock.patch('zerver.lib.remote_server.requests.request')
    def test_realmauditlog_data_mapping(self, mock_request: Any) -> None:
        mock_request.side_effect = self.bounce_request
        user = self.example_user('hamlet')
        log_entry = RealmAuditLog.objects.create(
            realm=user.realm, modified_user=user, backfilled=True,
            event_type=RealmAuditLog.USER_REACTIVATED, event_time=self.TIME_ZERO, extra_data='data')
        send_analytics_to_remote_server()
        remote_log_entry = RemoteRealmAuditLog.objects.order_by('id').last()
        self.assertEqual(remote_log_entry.server.uuid, self.server_uuid)
        self.assertEqual(remote_log_entry.remote_id, log_entry.id)
        self.assertEqual(remote_log_entry.event_time, self.TIME_ZERO)
        self.assertEqual(remote_log_entry.backfilled, True)
        self.assertEqual(remote_log_entry.extra_data, 'data')
        self.assertEqual(remote_log_entry.event_type, RealmAuditLog.USER_REACTIVATED)

class PushNotificationTest(BouncerTestCase):
    def setUp(self) -> None:
        super().setUp()
        self.user_profile = self.example_user('hamlet')
        self.sending_client = get_client('test')
        self.sender = self.example_user('hamlet')

    def get_message(self, type: int, type_id: int=100) -> Message:
        recipient, _ = Recipient.objects.get_or_create(
            type_id=type_id,
            type=type,
        )

        message = Message(
            sender=self.sender,
            recipient=recipient,
            content='This is test content',
            rendered_content='This is test content',
            date_sent=now(),
            sending_client=self.sending_client,
        )
        message.set_topic_name('Test Topic')
        message.save()

        return message

    @contextmanager
    def mock_apns(self) -> Iterator[mock.MagicMock]:
        mock_apns = mock.Mock()
        with mock.patch('zerver.lib.push_notifications.get_apns_client') as mock_get:
            mock_get.return_value = mock_apns
            yield mock_apns

    def setup_apns_tokens(self) -> None:
        self.tokens = ['aaaa', 'bbbb']
        for token in self.tokens:
            PushDeviceToken.objects.create(
                kind=PushDeviceToken.APNS,
                token=hex_to_b64(token),
                user=self.user_profile,
                ios_app_id=settings.ZULIP_IOS_APP_ID)

        self.remote_tokens = ['cccc']
        for token in self.remote_tokens:
            RemotePushDeviceToken.objects.create(
                kind=RemotePushDeviceToken.APNS,
                token=hex_to_b64(token),
                user_id=self.user_profile.id,
                server=RemoteZulipServer.objects.get(uuid=self.server_uuid),
            )

    def setup_gcm_tokens(self) -> None:
        self.gcm_tokens = ['<PASSWORD>', '<PASSWORD>']
        for token in self.gcm_tokens:
            PushDeviceToken.objects.create(
                kind=PushDeviceToken.GCM,
                token=hex_to_b64(token),
                user=self.user_profile,
{'fraction_non_alphanumeric': 0.09326162820102163, 'fraction_numerical': 0.006288247888463678, 'mean_word_length': 2.3816772133857818, 'pattern_counts': {'":': 20, '<': 19, '<?xml version=': 0, '>': 63, 'https://': 14, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 86, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4541080', 'n_tokens_mistral': 17343, 'n_tokens_neox': 16828, 'n_words': 3189}