zulip/zerver/tests/test_events.py

# -*- coding: utf-8 -*-
# See http://zulip.readthedocs.io/en/latest/events-system.html for
# high-level documentation on how this system works.
from __future__ import absolute_import
from __future__ import print_function
from typing import Any, Callable, Dict, List, Optional, Union, Text

from django.conf import settings
from django.http import HttpRequest, HttpResponse
from django.test import TestCase
from django.utils import timezone

from zerver.models import (
    get_client, get_realm, get_recipient, get_stream, get_user_profile_by_email,
    Message, RealmDomain, Recipient, UserMessage, UserPresence, UserProfile,
    Realm,
)

from zerver.lib.actions import (
    bulk_add_subscriptions,
    bulk_remove_subscriptions,
    do_add_alert_words,
    check_add_realm_emoji,
    check_send_typing_notification,
    do_add_realm_filter,
    do_add_reaction,
    do_remove_reaction,
    do_change_avatar_fields,
    do_change_default_all_public_streams,
    do_change_default_events_register_stream,
    do_change_default_sending_stream,
    do_change_full_name,
    do_change_bot_owner,
    do_change_is_admin,
    do_change_stream_description,
    do_change_subscription_property,
    do_create_user,
    do_deactivate_stream,
    do_deactivate_user,
    do_mark_hotspot_as_read,
    do_reactivate_user,
    do_refer_friend,
    do_regenerate_api_key,
    do_remove_alert_words,
    do_remove_realm_emoji,
    do_remove_realm_filter,
    do_rename_stream,
    do_add_default_stream,
    do_remove_default_stream,
    do_set_muted_topics,
    do_set_realm_property,
    do_set_realm_authentication_methods,
    do_set_realm_message_editing,
    do_update_embedded_data,
    do_update_message,
    do_update_message_flags,
    do_update_muted_topic,
    do_update_pointer,
    do_update_user_presence,
    do_set_user_display_setting,
    do_change_enable_stream_desktop_notifications,
    do_change_enable_stream_sounds,
    do_change_enable_desktop_notifications,
    do_change_enable_sounds,
    do_change_enable_offline_email_notifications,
    do_change_enable_offline_push_notifications,
    do_change_enable_online_push_notifications,
    do_change_pm_content_in_desktop_notifications,
    do_change_enable_digest_emails,
    do_add_realm_domain,
    do_change_realm_domain,
    do_remove_realm_domain,
    do_change_icon_source,
)
from zerver.lib.events import (
    apply_events,
    fetch_initial_state_data,
)
from zerver.lib.message import render_markdown
from zerver.lib.test_helpers import POSTRequestMock, get_subscription
from zerver.lib.test_classes import (
    ZulipTestCase,
)
from zerver.lib.validator import (
    check_bool, check_dict, check_dict_only, check_float, check_int, check_list, check_string,
    equals, check_none_or, Validator
)

from zerver.views.events_register import _default_all_public_streams, _default_narrow

from zerver.tornado.event_queue import allocate_client_descriptor, EventQueue
from zerver.tornado.views import get_events_backend

from collections import OrderedDict
import mock
import time
import ujson
from six.moves import range

class EventsEndpointTest(ZulipTestCase):
    def test_events_register_endpoint(self):
        # type: () -> None

        # This test is intended to get minimal coverage on the
        # events_register code paths
        email = 'hamlet@zulip.com'
        with mock.patch('zerver.views.events_register.do_events_register', return_value={}):
            result = self.client_post('/json/register', **self.api_auth(email))
        self.assert_json_success(result)

        with mock.patch('zerver.lib.events.request_event_queue', return_value=None):
            result = self.client_post('/json/register', **self.api_auth(email))
        self.assert_json_error(result, "Could not allocate event queue")

        with mock.patch('zerver.lib.events.request_event_queue', return_value='15:11'):
            with mock.patch('zerver.lib.events.get_user_events',
                            return_value=[]):
                result = self.client_post('/json/register', dict(event_types=ujson.dumps(['pointer'])),
                                          **self.api_auth(email))
        self.assert_json_success(result)
        result_dict = ujson.loads(result.content)
        self.assertEqual(result_dict['last_event_id'], -1)
        self.assertEqual(result_dict['queue_id'], '15:11')

        with mock.patch('zerver.lib.events.request_event_queue', return_value='15:12'):
            with mock.patch('zerver.lib.events.get_user_events',
                            return_value=[{
                                'id': 6,
                                'type': 'pointer',
                                'pointer': 15,
                            }]):
                result = self.client_post('/json/register', dict(event_types=ujson.dumps(['pointer'])),
                                          **self.api_auth(email))
        self.assert_json_success(result)
        result_dict = ujson.loads(result.content)
        self.assertEqual(result_dict['last_event_id'], 6)
        self.assertEqual(result_dict['pointer'], 15)
        self.assertEqual(result_dict['queue_id'], '15:12')

    def test_tornado_endpoint(self):
        # type: () -> None

        # This test is mostly intended to get minimal coverage on
        # the /notify_tornado endpoint, so we can have 100% URL coverage,
        # but it does exercise a little bit of the codepath.
        post_data = dict(
            data=ujson.dumps(
                dict(
                    event=dict(
                        type='other'
                    ),
                    users=[get_user_profile_by_email('hamlet@zulip.com').id],
                ),
            ),
        )
        req = POSTRequestMock(post_data, user_profile=None)
        req.META['REMOTE_ADDR'] = '127.0.0.1'
        result = self.client_post_request('/notify_tornado', req)
        self.assert_json_error(result, 'Access denied', status_code=403)

        post_data['secret'] = settings.SHARED_SECRET
        req = POSTRequestMock(post_data, user_profile=None)
        req.META['REMOTE_ADDR'] = '127.0.0.1'
        result = self.client_post_request('/notify_tornado', req)
        self.assert_json_success(result)

class GetEventsTest(ZulipTestCase):
    def tornado_call(self, view_func, user_profile, post_data):
        # type: (Callable[[HttpRequest, UserProfile], HttpResponse], UserProfile, Dict[str, Any]) -> HttpResponse
        request = POSTRequestMock(post_data, user_profile)
        return view_func(request, user_profile)

    def test_get_events(self):
        # type: () -> None
        email = "hamlet@zulip.com"
        recipient_email = "othello@zulip.com"
        user_profile = get_user_profile_by_email(email)
        recipient_user_profile = get_user_profile_by_email(recipient_email)
        self.login(email)

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"apply_markdown": ujson.dumps(True),
                                    "event_types": ujson.dumps(["message"]),
                                    "user_client": "website",
                                    "dont_block": ujson.dumps(True),
                                    })
        self.assert_json_success(result)
        queue_id = ujson.loads(result.content)["queue_id"]

        recipient_result = self.tornado_call(get_events_backend, recipient_user_profile,
                                             {"apply_markdown": ujson.dumps(True),
                                              "event_types": ujson.dumps(["message"]),
                                              "user_client": "website",
                                              "dont_block": ujson.dumps(True),
                                              })
        self.assert_json_success(recipient_result)
        recipient_queue_id = ujson.loads(recipient_result.content)["queue_id"]

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"queue_id": queue_id,
                                    "user_client": "website",
                                    "last_event_id": -1,
                                    "dont_block": ujson.dumps(True),
                                    })
        events = ujson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 0)

        local_id = 10.01
        self.send_message(email, recipient_email, Recipient.PERSONAL, "hello", local_id=local_id, sender_queue_id=queue_id)

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"queue_id": queue_id,
                                    "user_client": "website",
                                    "last_event_id": -1,
                                    "dont_block": ujson.dumps(True),
                                    })
        events = ujson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 1)
        self.assertEqual(events[0]["type"], "message")
        self.assertEqual(events[0]["message"]["sender_email"], email)
        self.assertEqual(events[0]["local_message_id"], local_id)
        self.assertEqual(events[0]["message"]["display_recipient"][0]["is_mirror_dummy"], False)
        self.assertEqual(events[0]["message"]["display_recipient"][1]["is_mirror_dummy"], False)

        last_event_id = events[0]["id"]
        local_id += 0.01

        self.send_message(email, recipient_email, Recipient.PERSONAL, "hello", local_id=local_id, sender_queue_id=queue_id)

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"queue_id": queue_id,
                                    "user_client": "website",
                                    "last_event_id": last_event_id,
                                    "dont_block": ujson.dumps(True),
                                    })
        events = ujson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 1)
        self.assertEqual(events[0]["type"], "message")
        self.assertEqual(events[0]["message"]["sender_email"], email)
        self.assertEqual(events[0]["local_message_id"], local_id)

        # Test that the received message in the receiver's event queue
        # exists and does not contain a local id
        recipient_result = self.tornado_call(get_events_backend, recipient_user_profile,
                                             {"queue_id": recipient_queue_id,
                                              "user_client": "website",
                                              "last_event_id": -1,
                                              "dont_block": ujson.dumps(True),
                                              })
        recipient_events = ujson.loads(recipient_result.content)["events"]
        self.assert_json_success(recipient_result)
        self.assertEqual(len(recipient_events), 2)
        self.assertEqual(recipient_events[0]["type"], "message")
        self.assertEqual(recipient_events[0]["message"]["sender_email"], email)
        self.assertTrue("local_message_id" not in recipient_events[0])
        self.assertEqual(recipient_events[1]["type"], "message")
        self.assertEqual(recipient_events[1]["message"]["sender_email"], email)
        self.assertTrue("local_message_id" not in recipient_events[1])

    def test_get_events_narrow(self):
        # type: () -> None
        email = "hamlet@zulip.com"
        user_profile = get_user_profile_by_email(email)
        self.login(email)

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"apply_markdown": ujson.dumps(True),
                                    "event_types": ujson.dumps(["message"]),
                                    "narrow": ujson.dumps([["stream", "denmark"]]),
                                    "user_client": "website",
                                    "dont_block": ujson.dumps(True),
                                    })
        self.assert_json_success(result)
        queue_id = ujson.loads(result.content)["queue_id"]

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"queue_id": queue_id,
                                    "user_client": "website",
                                    "last_event_id": -1,
                                    "dont_block": ujson.dumps(True),
                                    })
        events = ujson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 0)

        self.send_message(email, "othello@zulip.com", Recipient.PERSONAL, "hello")
        self.send_message(email, "Denmark", Recipient.STREAM, "hello")

        result = self.tornado_call(get_events_backend, user_profile,
                                   {"queue_id": queue_id,
                                    "user_client": "website",
                                    "last_event_id": -1,
                                    "dont_block": ujson.dumps(True),
                                    })
        events = ujson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 1)
        self.assertEqual(events[0]["type"], "message")
        self.assertEqual(events[0]["message"]["display_recipient"], "Denmark")

class EventsRegisterTest(ZulipTestCase):
    user_profile = get_user_profile_by_email("hamlet@zulip.com")
    maxDiff = None # type: Optional[int]

    def create_bot(self, email):
        # type: (str) -> UserProfile
        return do_create_user(email, '123',
                              get_realm('zulip'), 'Test Bot', 'test',
                              bot_type=UserProfile.DEFAULT_BOT, bot_owner=self.user_profile)

    def realm_bot_schema(self, field_name, check):
        # type: (str, Validator) -> Validator
        return check_dict_only([
            ('id', check_int),
            ('type', equals('realm_bot')),
            ('op', equals('update')),
            ('bot', check_dict_only([
                ('email', check_string),
                ('user_id', check_int),
                (field_name, check),
            ])),
        ])

    def do_test(self, action, event_types=None, include_subscribers=True, state_change_expected=True,
                num_events=1):
        # type: (Callable[[], Any], Optional[List[str]], bool, bool, int) -> List[Dict[str, Any]]
        client = allocate_client_descriptor(
            dict(user_profile_id = self.user_profile.id,
                 user_profile_email = self.user_profile.email,
                 realm_id = self.user_profile.realm_id,
                 event_types = event_types,
                 client_type_name = "website",
                 apply_markdown = True,
                 all_public_streams = False,
                 queue_timeout = 600,
                 last_connection_time = time.time(),
                 narrow = [])
        )
        # hybrid_state = initial fetch state + re-applying events triggered by our action
        # normal_state = do action then fetch at the end (the "normal" code path)
        hybrid_state = fetch_initial_state_data(self.user_profile, event_types, "", include_subscribers=include_subscribers)
        action()
        events = client.event_queue.contents()
        self.assertTrue(len(events) == num_events)

        before = ujson.dumps(hybrid_state)
        apply_events(hybrid_state, events, self.user_profile, include_subscribers=include_subscribers)
        after = ujson.dumps(hybrid_state)

        if state_change_expected:
            if before == after:
                print(events)  # nocoverage
                raise AssertionError('Test does not exercise enough code -- events do not change state.')
        else:
            if before != after:
                raise AssertionError('Test is invalid--state actually does change here.')

        normal_state = fetch_initial_state_data(self.user_profile, event_types, "", include_subscribers=include_subscribers)
        self.match_states(hybrid_state, normal_state)
        return events

    def assert_on_error(self, error):
        # type: (Optional[str]) -> None
        if error:
            raise AssertionError(error)

    def match_states(self, state1, state2):
        # type: (Dict[str, Any], Dict[str, Any]) -> None
        def normalize(state):
            # type: (Dict[str, Any]) -> None
            state['realm_users'] = {u['email']: u for u in state['realm_users']}
            for u in state['subscriptions']:
                if 'subscribers' in u:
                    u['subscribers'].sort()
            state['subscriptions'] = {u['name']: u for u in state['subscriptions']}
            state['unsubscribed'] = {u['name']: u for u in state['unsubscribed']}
            if 'realm_bots' in state:
                state['realm_bots'] = {u['email']: u for u in state['realm_bots']}
        normalize(state1)
        normalize(state2)
        self.assertEqual(state1, state2)

    def test_send_message_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('message')),
            ('flags', check_list(None)),
            ('message', check_dict([
                ('avatar_url', check_string),
                ('client', check_string),
                ('content', check_string),
                ('content_type', equals('text/html')),
                ('display_recipient', check_string),
                ('gravatar_hash', check_string),
                ('id', check_int),
                ('recipient_id', check_int),
                ('sender_realm_str', check_string),
                ('sender_email', check_string),
                ('sender_full_name', check_string),
                ('sender_id', check_int),
                ('sender_short_name', check_string),
                ('subject', check_string),
                ('subject_links', check_list(None)),
                ('timestamp', check_int),
                ('type', check_string),
            ])),
        ])
        events = self.do_test(
            lambda: self.send_message("hamlet@zulip.com", "Verona", Recipient.STREAM, "hello"),
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        # Verify message editing
        schema_checker = check_dict([
            ('type', equals('update_message')),
            ('flags', check_list(None)),
            ('content', check_string),
            ('edit_timestamp', check_int),
            ('flags', check_list(None)),
            ('message_id', check_int),
            ('message_ids', check_list(check_int)),
            ('orig_content', check_string),
            ('orig_rendered_content', check_string),
            ('orig_subject', check_string),
            ('propagate_mode', check_string),
            ('rendered_content', check_string),
            ('sender', check_string),
            ('stream_id', check_int),
            ('subject', check_string),
            ('subject_links', check_list(None)),
            ('user_id', check_int),
            # There is also a timestamp field in the event, but we ignore it, as
            # it's kind of an unwanted but harmless side effect of calling log_event.
        ])

        message = Message.objects.order_by('-id')[0]
        topic = 'new_topic'
        propagate_mode = 'change_all'
        content = 'new content'
        rendered_content = render_markdown(message, content)
        events = self.do_test(
            lambda: do_update_message(self.user_profile, message, topic,
                                      propagate_mode, content, rendered_content),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        # Verify do_update_embedded_data
        schema_checker = check_dict([
            ('type', equals('update_message')),
            ('flags', check_list(None)),
            ('content', check_string),
            ('flags', check_list(None)),
            ('message_id', check_int),
            ('message_ids', check_list(check_int)),
            ('rendered_content', check_string),
            ('sender', check_string),
        ])

        events = self.do_test(
            lambda: do_update_embedded_data(self.user_profile, message,
                                            u"embed_content", "<p>embed_content</p>"),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_update_message_flags(self):
        # type: () -> None
        # Test message flag update events
        schema_checker = check_dict([
            ('id', check_int),
            ('type', equals('update_message_flags')),
            ('flag', check_string),
            ('messages', check_list(check_int)),
            ('operation', equals("add")),
        ])

        message = self.send_message("cordelia@zulip.com", "hamlet@zulip.com", Recipient.PERSONAL, "hello")
        user_profile = get_user_profile_by_email("hamlet@zulip.com")
        events = self.do_test(
            lambda: do_update_message_flags(user_profile, 'add', 'starred',
                                            [message], False, None, None),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        schema_checker = check_dict([
            ('id', check_int),
            ('type', equals('update_message_flags')),
            ('flag', check_string),
            ('messages', check_list(check_int)),
            ('operation', equals("remove")),
        ])
        events = self.do_test(
            lambda: do_update_message_flags(user_profile, 'remove', 'starred',
                                            [message], False, None, None),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_send_reaction(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('reaction')),
            ('op', equals('add')),
            ('message_id', check_int),
            ('emoji_name', check_string),
            ('user', check_dict([
                ('email', check_string),
                ('full_name', check_string),
                ('user_id', check_int)
            ])),
        ])

        message_id = self.send_message("hamlet@zulip.com", "Verona", Recipient.STREAM, "hello")
        message = Message.objects.get(id=message_id)
        events = self.do_test(
            lambda: do_add_reaction(
                self.user_profile, message, "tada"),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_remove_reaction(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('reaction')),
            ('op', equals('remove')),
            ('message_id', check_int),
            ('emoji_name', check_string),
            ('user', check_dict([
                ('email', check_string),
                ('full_name', check_string),
                ('user_id', check_int)
            ])),
        ])

        message_id = self.send_message("hamlet@zulip.com", "Verona", Recipient.STREAM, "hello")
        message = Message.objects.get(id=message_id)
        events = self.do_test(
            lambda: do_remove_reaction(
                self.user_profile, message, "tada"),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_typing_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('typing')),
            ('op', equals('start')),
            ('sender', check_dict([
                ('email', check_string),
                ('user_id', check_int)])),
            ('recipients', check_list(check_dict([
                ('email', check_string),
                ('user_id', check_int),
            ]))),
        ])

        events = self.do_test(
            lambda: check_send_typing_notification(
                self.user_profile, ["cordelia@zulip.com"], "start"),
            state_change_expected=False,
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_presence_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('pointer')),
            ('email', check_string),
            ('timestamp', check_float),
            ('presence', check_dict([
                # TODO: Add more here once the test below works
            ])),
        ])
        # BUG: Marked as failing for now because this is a failing
        # test, due to the `aggregated` feature not being supported by
        # our events code.

        with self.assertRaises(AssertionError):
            events = self.do_test(lambda: do_update_user_presence(
                self.user_profile, get_client("website"), timezone.now(), UserPresence.ACTIVE))
            # Marked as nocoverage since unreachable
            error = schema_checker('events[0]', events[0])  # nocoverage
            self.assert_on_error(error)  # nocoverage

    def test_pointer_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('pointer')),
            ('pointer', check_int)
        ])
        events = self.do_test(lambda: do_update_pointer(self.user_profile, 1500))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_referral_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('referral')),
            ('referrals', check_dict([
                ('granted', check_int),
                ('used', check_int),
            ])),
        ])
        events = self.do_test(lambda: do_refer_friend(self.user_profile, "friend@example.com"))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_register_events(self):
        # type: () -> None
        realm_user_add_checker = check_dict([
            ('type', equals('realm_user')),
            ('op', equals('add')),
            ('person', check_dict([
                ('email', check_string),
                ('full_name', check_string),
                ('is_admin', check_bool),
                ('is_bot', check_bool),
            ])),
        ])

        events = self.do_test(lambda: self.register("test1@zulip.com", "test1"))
        self.assert_length(events, 1)
        error = realm_user_add_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_alert_words_events(self):
        # type: () -> None
        alert_words_checker = check_dict([
            ('type', equals('alert_words')),
            ('alert_words', check_list(check_string)),
        ])

        events = self.do_test(lambda: do_add_alert_words(self.user_profile, ["alert_word"]))
        error = alert_words_checker('events[0]', events[0])
        self.assert_on_error(error)

        events = self.do_test(lambda: do_remove_alert_words(self.user_profile, ["alert_word"]))
        error = alert_words_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_default_streams_events(self):
        # type: () -> None
        default_streams_checker = check_dict([
            ('type', equals('default_streams')),
            ('default_streams', check_list(check_dict([
                ('description', check_string),
                ('invite_only', check_bool),
                ('name', check_string),
                ('stream_id', check_int),
            ]))),
        ])

        stream = get_stream("Scotland", self.user_profile.realm)
        events = self.do_test(lambda: do_add_default_stream(stream))
        error = default_streams_checker('events[0]', events[0])
        events = self.do_test(lambda: do_remove_default_stream(stream))
        error = default_streams_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_muted_topics_events(self):
        # type: () -> None
        muted_topics_checker = check_dict([
            ('type', equals('muted_topics')),
            ('muted_topics', check_list(check_list(check_string, 2))),
        ])
        events = self.do_test(lambda: do_set_muted_topics(self.user_profile, [[u"Denmark", u"topic"]]))
        error = muted_topics_checker('events[0]', events[0])
        self.assert_on_error(error)

        events = self.do_test(lambda: do_update_muted_topic(
            self.user_profile, "Denmark", "topic", "add"))
        error = muted_topics_checker('events[0]', events[0])
        self.assert_on_error(error)

        events = self.do_test(lambda: do_update_muted_topic(
            self.user_profile, "Denmark", "topic", "remove"))
        error = muted_topics_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_avatar_fields(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm_user')),
            ('op', equals('update')),
            ('person', check_dict([
                ('email', check_string),
                ('user_id', check_int),
                ('avatar_url', check_string),
            ])),
        ])
        events = self.do_test(
            lambda: do_change_avatar_fields(self.user_profile, UserProfile.AVATAR_FROM_USER),
        )
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_full_name(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm_user')),
            ('op', equals('update')),
            ('person', check_dict([
                ('email', check_string),
                ('full_name', check_string),
            ])),
        ])
        events = self.do_test(lambda: do_change_full_name(self.user_profile, 'Sir Hamlet'))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def do_set_realm_property_test(self, name):
        # type: (str) -> None
        bool_tests = [True, False, True]  # type: List[bool]
        test_values = dict(
            add_emoji_by_admins_only=bool_tests,
            create_stream_by_admins_only=bool_tests,
            default_language=[u'es', u'de', u'en'],
            description=[u'Realm description', u'New description'],
            email_changes_disabled=bool_tests,
            invite_required=bool_tests,
            invite_by_admins_only=bool_tests,
            inline_image_preview=bool_tests,
            inline_url_embed_preview=bool_tests,
            message_retention_days=[10, 20],
            name=[u'Zulip', u'New Name'],
            name_changes_disabled=bool_tests,
            restricted_to_domain=bool_tests,
            waiting_period_threshold=[10, 20],
        )  # type: Dict[str, Any]

        property_type = Realm.property_types[name]
        if property_type is bool:
            validator = check_bool
        elif property_type is Text:
            validator = check_string
        elif property_type is int:
            validator = check_int
        elif property_type == (int, type(None)):
            validator = check_int
        else:
            raise AssertionError("Unexpected property type %s" % (property_type,))

        schema_checker = check_dict([
            ('type', equals('realm')),
            ('op', equals('update')),
            ('property', equals(name)),
            ('value', validator),
        ])

        vals = test_values.get(name)
        if vals is None:
            raise AssertionError('No test created for %s' % (name))
        do_set_realm_property(self.user_profile.realm, name, vals[0])
        for val in vals[1:]:
            events = self.do_test(
                lambda: do_set_realm_property(self.user_profile.realm, name, val))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_realm_property(self):
        # type: () -> None

        for prop in Realm.property_types:
            self.do_set_realm_property_test(prop)

    def test_change_realm_authentication_methods(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm')),
            ('op', equals('update_dict')),
            ('property', equals('default')),
            ('data', check_dict([])),
        ])

        def fake_backends():
            # type: () -> Any
            backends = (
                'zproject.backends.DevAuthBackend',
                'zproject.backends.EmailAuthBackend',
                'zproject.backends.GitHubAuthBackend',
                'zproject.backends.GoogleMobileOauth2Backend',
                'zproject.backends.ZulipLDAPAuthBackend',
            )
            return self.settings(AUTHENTICATION_BACKENDS=backends)

        # Test transitions; any new backends should be tested with T/T/T/F/T
        for (auth_method_dict) in \
                ({'Google': True, 'Email': True, 'GitHub': True, 'LDAP': False, 'Dev': False},
                 {'Google': True, 'Email': True, 'GitHub': False, 'LDAP': False, 'Dev': False},
                 {'Google': True, 'Email': False, 'GitHub': False, 'LDAP': False, 'Dev': False},
                 {'Google': True, 'Email': False, 'GitHub': True, 'LDAP': False, 'Dev': False},
                 {'Google': False, 'Email': False, 'GitHub': False, 'LDAP': False, 'Dev': True},
                 {'Google': False, 'Email': False, 'GitHub': True, 'LDAP': False, 'Dev': True},
                 {'Google': False, 'Email': True, 'GitHub': True, 'LDAP': True, 'Dev': False}):
            with fake_backends():
                events = self.do_test(
                    lambda: do_set_realm_authentication_methods(
                        self.user_profile.realm,
                        auth_method_dict))

            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_pin_stream(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('subscription')),
            ('op', equals('update')),
            ('property', equals('pin_to_top')),
            ('stream_id', check_int),
            ('value', check_bool),
        ])
        stream = get_stream("Denmark", self.user_profile.realm)
        sub = get_subscription(stream.name, self.user_profile)
        do_change_subscription_property(self.user_profile, sub, stream, "pin_to_top", False)
        for pinned in (True, False):
            events = self.do_test(lambda: do_change_subscription_property(self.user_profile, sub, stream, "pin_to_top", pinned))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_realm_message_edit_settings(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm')),
            ('op', equals('update_dict')),
            ('property', equals('default')),
            ('data', check_dict([('allow_message_editing', check_bool),
                                 ('message_content_edit_limit_seconds', check_int)])),
        ])
        # Test every transition among the four possibilities {T,F} x {0, non-0}
        for (allow_message_editing, message_content_edit_limit_seconds) in \
            ((True, 0), (False, 0), (True, 0), (False, 1234), (True, 0), (True, 1234), (True, 0),
             (False, 0), (False, 1234), (False, 0), (True, 1234), (False, 0),
             (True, 1234), (True, 600), (False, 600), (False, 1234), (True, 600)):
            events = self.do_test(
                lambda: do_set_realm_message_editing(self.user_profile.realm,
                                                     allow_message_editing,
                                                     message_content_edit_limit_seconds))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_is_admin(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm_user')),
            ('op', equals('update')),
            ('person', check_dict([
                ('email', check_string),
                ('is_admin', check_bool),
            ])),
        ])
        do_change_is_admin(self.user_profile, False)
        for is_admin in [True, False]:
            events = self.do_test(lambda: do_change_is_admin(self.user_profile, is_admin))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def do_set_user_display_settings_test(self, setting_name, values_list):
        # type: (str, List[Union[bool, Text]]) -> None

        property_type = UserProfile.property_types[setting_name]
        if property_type is bool:
            validator = check_bool
        elif property_type is Text:
            validator = check_string
        else:
            raise AssertionError("Unexpected property type %s" % (property_type,))

        num_events = 1
        if setting_name == "timezone":
            num_events = 2
        if property_type == bool:
            do_set_user_display_setting(self.user_profile, setting_name, False)
        for value in values_list:
            events = self.do_test(lambda: do_set_user_display_setting(
                self.user_profile, setting_name, value), num_events=num_events)

            schema_checker = check_dict([
                ('type', equals('update_display_settings')),
                ('setting_name', equals(setting_name)),
                ('user', check_string),
                ('setting', validator),
            ])
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

            timezone_schema_checker = check_dict([
                ('type', equals('realm_user')),
                ('op', equals('update')),
                ('person', check_dict([
                    ('email', check_string),
                    ('user_id', check_int),
                    ('timezone', check_string),
                ])),
            ])
            if setting_name == "timezone":
                error = timezone_schema_checker('events[1]', events[1])

    def test_change_twenty_four_hour_time(self):
        # type: () -> None
        self.do_set_user_display_settings_test("twenty_four_hour_time", [True, False])

    def test_change_left_side_userlist(self):
        # type: () -> None
        self.do_set_user_display_settings_test("left_side_userlist", [True, False])

    def test_change_emoji_alt_code(self):
        # type: () -> None
        self.do_set_user_display_settings_test("emoji_alt_code", [True, False])

    def test_change_default_language(self):
        # type: () -> None
        self.do_set_user_display_settings_test("default_language", [u'de', u'es', u'en'])

    def test_change_timezone(self):
        # type: () -> None
        self.do_set_user_display_settings_test("timezone", [u'US/Mountain', u'US/Samoa', u'Pacific/Galapagos', u''])

    def test_change_enable_stream_desktop_notifications(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_stream_desktop_notifications')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_stream_desktop_notifications(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_stream_desktop_notifications(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_stream_sounds(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_stream_sounds')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_stream_sounds(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_stream_sounds(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_desktop_notifications(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_desktop_notifications')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_desktop_notifications(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_desktop_notifications(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_sounds(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_sounds')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_sounds(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_sounds(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_offline_email_notifications(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_offline_email_notifications')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_offline_email_notifications(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_offline_email_notifications(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_offline_push_notifications(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_offline_push_notifications')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_offline_push_notifications(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_offline_push_notifications(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_online_push_notifications(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_online_push_notifications')),
            ('user', check_string),
            ('setting', check_bool),
        ])

        do_change_enable_online_push_notifications(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_online_push_notifications(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_pm_content_in_desktop_notifications(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('pm_content_in_desktop_notifications')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_pm_content_in_desktop_notifications(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(
                lambda: do_change_pm_content_in_desktop_notifications(self.user_profile,
                                                                      setting_value),
                state_change_expected=False,
            )
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_change_enable_digest_emails(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('update_global_notifications')),
            ('notification_name', equals('enable_digest_emails')),
            ('user', check_string),
            ('setting', check_bool),
        ])
        do_change_enable_digest_emails(self.user_profile, False)
        for setting_value in [True, False]:
            events = self.do_test(lambda: do_change_enable_digest_emails(self.user_profile, setting_value))
            error = schema_checker('events[0]', events[0])
            self.assert_on_error(error)

    def test_realm_emoji_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm_emoji')),
            ('op', equals('update')),
            ('realm_emoji', check_dict([])),
        ])
        events = self.do_test(lambda: check_add_realm_emoji(get_realm("zulip"), "my_emoji",
                                                            "https://realm.com/my_emoji"))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        events = self.do_test(lambda: do_remove_realm_emoji(get_realm("zulip"), "my_emoji"))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_realm_filter_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm_filters')),
            ('realm_filters', check_list(None)), # TODO: validate tuples in the list
        ])
        events = self.do_test(lambda: do_add_realm_filter(get_realm("zulip"), "#(?P<id>[123])",
                                                          "https://realm.com/my_realm_filter/%(id)s"))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        self.do_test(lambda: do_remove_realm_filter(get_realm("zulip"), "#(?P<id>[123])"))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_realm_domain_events(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('realm_domains')),
            ('op', equals('add')),
            ('realm_domain', check_dict([
                ('domain', check_string),
                ('allow_subdomains', check_bool),
            ])),
        ])
        realm = get_realm('zulip')
        events = self.do_test(lambda: do_add_realm_domain(realm, 'zulip.org', False))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        schema_checker = check_dict([
            ('type', equals('realm_domains')),
            ('op', equals('change')),
            ('realm_domain', check_dict([
                ('domain', equals('zulip.org')),
                ('allow_subdomains', equals(True)),
            ])),
        ])
        test_domain = RealmDomain.objects.get(realm=realm, domain='zulip.org')
        events = self.do_test(lambda: do_change_realm_domain(test_domain, True))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        schema_checker = check_dict([
            ('type', equals('realm_domains')),
            ('op', equals('remove')),
            ('domain', equals('zulip.org')),
        ])
        events = self.do_test(lambda: do_remove_realm_domain(test_domain))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_create_bot(self):
        # type: () -> None
        bot_created_checker = check_dict([
            ('type', equals('realm_bot')),
            ('op', equals('add')),
            ('bot', check_dict([
                ('email', check_string),
                ('user_id', check_int),
                ('full_name', check_string),
                ('is_active', check_bool),
                ('api_key', check_string),
                ('default_sending_stream', check_none_or(check_string)),
                ('default_events_register_stream', check_none_or(check_string)),
                ('default_all_public_streams', check_bool),
                ('avatar_url', check_string),
            ])),
        ])
        action = lambda: self.create_bot('test-bot@zulip.com')
        events = self.do_test(action, num_events=2)
        error = bot_created_checker('events[1]', events[1])
        self.assert_on_error(error)

    def test_change_bot_full_name(self):
        # type: () -> None
        bot = self.create_bot('test-bot@zulip.com')
        action = lambda: do_change_full_name(bot, 'New Bot Name')
        events = self.do_test(action, num_events=2)
        error = self.realm_bot_schema('full_name', check_string)('events[1]', events[1])
        self.assert_on_error(error)

    def test_regenerate_bot_api_key(self):
        # type: () -> None
        bot = self.create_bot('test-bot@zulip.com')
        action = lambda: do_regenerate_api_key(bot)
        events = self.do_test(action)
        error = self.realm_bot_schema('api_key', check_string)('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_bot_avatar_source(self):
        # type: () -> None
        bot = self.create_bot('test-bot@zulip.com')
        action = lambda: do_change_avatar_fields(bot, bot.AVATAR_FROM_USER)
        events = self.do_test(action, num_events=2)
        error = self.realm_bot_schema('avatar_url', check_string)('events[0]', events[0])
        self.assertEqual(events[1]['type'], 'realm_user')
        self.assert_on_error(error)

    def test_change_realm_icon_source(self):
        # type: () -> None
        realm = get_realm('zulip')
        action = lambda: do_change_icon_source(realm, realm.ICON_FROM_GRAVATAR)
        events = self.do_test(action, state_change_expected=False)
        schema_checker = check_dict([
            ('type', equals('realm')),
            ('op', equals('update_dict')),
            ('property', equals('icon')),
            ('data',
             check_dict([('icon_url', check_string),
                         ('icon_source', check_string)])),
        ])
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_bot_default_all_public_streams(self):
        # type: () -> None
        bot = self.create_bot('test-bot@zulip.com')
        action = lambda: do_change_default_all_public_streams(bot, True)
        events = self.do_test(action)
        error = self.realm_bot_schema('default_all_public_streams', check_bool)('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_bot_default_sending_stream(self):
        # type: () -> None
        bot = self.create_bot('test-bot@zulip.com')
        stream = get_stream("Rome", bot.realm)

        action = lambda: do_change_default_sending_stream(bot, stream)
        events = self.do_test(action)
        error = self.realm_bot_schema('default_sending_stream', check_string)('events[0]', events[0])
        self.assert_on_error(error)

        action = lambda: do_change_default_sending_stream(bot, None)
        events = self.do_test(action)
        error = self.realm_bot_schema('default_sending_stream', equals(None))('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_bot_default_events_register_stream(self):
        # type: () -> None
        bot = self.create_bot('test-bot@zulip.com')
        stream = get_stream("Rome", bot.realm)

        action = lambda: do_change_default_events_register_stream(bot, stream)
        events = self.do_test(action)
        error = self.realm_bot_schema('default_events_register_stream', check_string)('events[0]', events[0])
        self.assert_on_error(error)

        action = lambda: do_change_default_events_register_stream(bot, None)
        events = self.do_test(action)
        error = self.realm_bot_schema('default_events_register_stream', equals(None))('events[0]', events[0])
        self.assert_on_error(error)

    def test_change_bot_owner(self):
        # type: () -> None
        change_bot_owner_checker = check_dict([
            ('type', equals('realm_bot')),
            ('op', equals('update')),
            ('bot', check_dict([
                ('email', check_string),
                ('user_id', check_int),
                ('owner_id', check_int),
            ])),
        ])
        self.user_profile = get_user_profile_by_email('iago@zulip.com')
        owner = get_user_profile_by_email('hamlet@zulip.com')
        bot = self.create_bot('test-bot@zulip.com')
        action = lambda: do_change_bot_owner(bot, owner, self.user_profile)
        events = self.do_test(action)
        error = change_bot_owner_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_do_deactivate_user(self):
        # type: () -> None
        bot_deactivate_checker = check_dict([
            ('type', equals('realm_bot')),
            ('op', equals('remove')),
            ('bot', check_dict([
                ('email', check_string),
                ('full_name', check_string),
            ])),
        ])
        bot = self.create_bot('foo-bot@zulip.com')
        action = lambda: do_deactivate_user(bot)
        events = self.do_test(action, num_events=2)
        error = bot_deactivate_checker('events[1]', events[1])
        self.assert_on_error(error)

    def test_do_reactivate_user(self):
        # type: () -> None
        bot_reactivate_checker = check_dict([
            ('type', equals('realm_bot')),
            ('op', equals('add')),
            ('bot', check_dict([
                ('email', check_string),
                ('user_id', check_int),
                ('full_name', check_string),
                ('is_active', check_bool),
                ('api_key', check_string),
                ('default_sending_stream', check_none_or(check_string)),
                ('default_events_register_stream', check_none_or(check_string)),
                ('default_all_public_streams', check_bool),
                ('avatar_url', check_string),
                ('owner', check_none_or(check_string)),
            ])),
        ])
        bot = self.create_bot('foo-bot@zulip.com')
        do_deactivate_user(bot)
        action = lambda: do_reactivate_user(bot)
        events = self.do_test(action, num_events=2)
        error = bot_reactivate_checker('events[1]', events[1])
        self.assert_on_error(error)

    def test_do_mark_hotspot_as_read(self):
        # type: () -> None
        schema_checker = check_dict([
            ('type', equals('hotspots')),
            ('hotspots', check_list(check_string)),
        ])
        events = self.do_test(lambda: do_mark_hotspot_as_read(self.user_profile, 'welcome'))
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_rename_stream(self):
        # type: () -> None
        stream = self.make_stream('old_name')
        new_name = u'stream with a brand new name'
        self.subscribe_to_stream(self.user_profile.email, stream.name)

        action = lambda: do_rename_stream(stream, new_name)
        events = self.do_test(action, num_events=2)

        schema_checker = check_dict([
            ('type', equals('stream')),
            ('op', equals('update')),
            ('property', equals('email_address')),
            ('value', check_string),
            ('stream_id', check_int),
            ('name', equals('old_name')),
        ])
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        schema_checker = check_dict([
            ('type', equals('stream')),
            ('op', equals('update')),
            ('property', equals('name')),
            ('value', equals(new_name)),
            ('name', equals('old_name')),
        ])
        error = schema_checker('events[1]', events[1])
        self.assert_on_error(error)

    def test_deactivate_stream_neversubscribed(self):
        # type: () -> None
        stream = self.make_stream('old_name')

        action = lambda: do_deactivate_stream(stream)
        events = self.do_test(action)

        schema_checker = check_dict([
            ('type', equals('stream')),
            ('op', equals('delete')),
            ('streams', check_list(check_dict([]))),
        ])
        error = schema_checker('events[0]', events[0])
        self.assert_on_error(error)

    def test_subscribe_other_user_never_subscribed(self):
        # type: () -> None
        action = lambda: self.subscribe_to_stream("othello@zulip.com", u"test_stream")
        events = self.do_test(action, num_events=2)
        peer_add_schema_checker = check_dict([
            ('type', equals('subscription')),
            ('op', equals('peer_add')),
            ('user_id', check_int),
            ('subscriptions', check_list(check_string)),
        ])
        error = peer_add_schema_checker('events[1]', events[1])
        self.assert_on_error(error)

    def test_subscribe_events(self):
        # type: () -> None
        self.do_test_subscribe_events(include_subscribers=True)

    def test_subscribe_events_no_include_subscribers(self):
        # type: () -> None
        self.do_test_subscribe_events(include_subscribers=False)

    def do_test_subscribe_events(self, include_subscribers):
        # type: (bool) -> None
        subscription_fields = [
            ('color', check_string),
            ('description', check_string),
            ('email_address', check_string),
            ('invite_only', check_bool),
            ('in_home_view', check_bool),
            ('name', check_string),
            ('desktop_notifications', check_bool),
            ('audible_notifications', check_bool),
            ('stream_id', check_int),
        ]
        if include_subscribers:
            subscription_fields.append(('subscribers', check_list(check_int)))  # type: ignore
        subscription_schema_checker = check_list(
            check_dict(subscription_fields),
        )
        stream_create_schema_checker = check_dict([
            ('type', equals('stream')),
            ('op', equals('create')),
            ('streams', check_list(check_dict([
                ('name', check_string),
                ('stream_id', check_int),
                ('invite_only', check_bool),
                ('description', check_string),
            ]))),
        ])
        add_schema_checker = check_dict([
            ('type', equals('subscription')),
            ('op', equals('add')),
            ('subscriptions', subscription_schema_checker),
        ])
        remove_schema_checker = check_dict([
            ('type', equals('subscription')),
            ('op', equals('remove')),
            ('subscriptions', check_list(
                check_dict([
                    ('name', equals('test_stream')),
                    ('stream_id', check_int),
                ]),
            )),
        ])
        peer_add_schema_checker = check_dict([
            ('type', equals('subscription')),
            ('op', equals('peer_add')),
            ('user_id', check_int),
            ('subscriptions', check_list(check_string)),
        ])
        peer_remove_schema_checker = check_dict([
            ('type', equals('subscription')),
            ('op', equals('peer_remove')),
            ('user_id', check_int),
            ('subscriptions', check_list(check_string)),
        ])
        stream_update_schema_checker = check_dict([
            ('type', equals('stream')),
            ('op', equals('update')),
            ('property', equals('description')),
            ('value', check_string),
            ('stream_id', check_int),
            ('name', check_string),
        ])

        # Subscribe to a totally new stream, so it's just Hamlet on it
        action = lambda: self.subscribe_to_stream("hamlet@zulip.com", "test_stream") # type: Callable
        events = self.do_test(action, event_types=["subscription", "realm_user"],
                              include_subscribers=include_subscribers)
        error = add_schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        # Add another user to that totally new stream
        action = lambda: self.subscribe_to_stream("othello@zulip.com", "test_stream")
        events = self.do_test(action,
                              include_subscribers=include_subscribers,
                              state_change_expected=include_subscribers,
                              )
        error = peer_add_schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        stream = get_stream("test_stream", self.user_profile.realm)

        # Now remove the first user, to test the normal unsubscribe flow
        action = lambda: bulk_remove_subscriptions(
            [get_user_profile_by_email("othello@zulip.com")],
            [stream])
        events = self.do_test(action,
                              include_subscribers=include_subscribers,
                              state_change_expected=include_subscribers,
                              )
        error = peer_remove_schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        # Now remove the second user, to test the 'vacate' event flow
        action = lambda: bulk_remove_subscriptions(
            [get_user_profile_by_email("hamlet@zulip.com")],
            [stream])
        events = self.do_test(action,
                              include_subscribers=include_subscribers,
                              num_events=2)
        error = remove_schema_checker('events[1]', events[1])
        self.assert_on_error(error)

        # Now resubscribe a user, to make sure that works on a vacated stream
        action = lambda: self.subscribe_to_stream("hamlet@zulip.com", "test_stream")
        events = self.do_test(action,
                              include_subscribers=include_subscribers,
                              num_events=2)
        error = add_schema_checker('events[1]', events[1])
        self.assert_on_error(error)

        action = lambda: do_change_stream_description(stream, u'new description')
        events = self.do_test(action,
                              include_subscribers=include_subscribers)
        error = stream_update_schema_checker('events[0]', events[0])
        self.assert_on_error(error)

        # Subscribe to a totally new invite-only stream, so it's just Hamlet on it
        stream = self.make_stream("private", get_realm("zulip"), invite_only=True)
        user_profile = get_user_profile_by_email("hamlet@zulip.com")
        action = lambda: bulk_add_subscriptions([stream], [user_profile])
        events = self.do_test(action, include_subscribers=include_subscribers,
                              num_events=2)
        error = stream_create_schema_checker('events[0]', events[0])
        error = add_schema_checker('events[1]', events[1])
        self.assert_on_error(error)

class FetchInitialStateDataTest(ZulipTestCase):
    # Non-admin users don't have access to all bots
    def test_realm_bots_non_admin(self):
        # type: () -> None
        email = 'cordelia@zulip.com'
        user_profile = get_user_profile_by_email(email)
        self.assertFalse(user_profile.is_realm_admin)
        result = fetch_initial_state_data(user_profile, None, "")
        self.assert_length(result['realm_bots'], 0)

        # additionally the API key for a random bot is not present in the data
        api_key = get_user_profile_by_email('notification-bot@zulip.com').api_key
        self.assertNotIn(api_key, str(result))

    # Admin users have access to all bots in the realm_bots field
    def test_realm_bots_admin(self):
        # type: () -> None
        email = 'hamlet@zulip.com'
        user_profile = get_user_profile_by_email(email)
        do_change_is_admin(user_profile, True)
        self.assertTrue(user_profile.is_realm_admin)
        result = fetch_initial_state_data(user_profile, None, "")
        self.assertTrue(len(result['realm_bots']) > 5)

    def test_max_message_id_with_no_history(self):
        # type: () -> None
        email = 'aaron@zulip.com'
        user_profile = get_user_profile_by_email(email)
        # Delete all historical messages for this user
        UserMessage.objects.filter(user_profile=user_profile).delete()
        result = fetch_initial_state_data(user_profile, None, "")
        self.assertEqual(result['max_message_id'], -1)

class EventQueueTest(TestCase):
    def test_one_event(self):
        # type: () -> None
        queue = EventQueue("1")
        queue.push({"type": "pointer",
                    "pointer": 1,
                    "timestamp": "1"})
        self.assertFalse(queue.empty())
        self.assertEqual(queue.contents(),
                         [{'id': 0,
                           'type': 'pointer',
                           "pointer": 1,
                           "timestamp": "1"}])

    def test_event_collapsing(self):
        # type: () -> None
        queue = EventQueue("1")
        for pointer_val in range(1, 10):
            queue.push({"type": "pointer",
                        "pointer": pointer_val,
                        "timestamp": str(pointer_val)})
        self.assertEqual(queue.contents(),
                         [{'id': 8,
                           'type': 'pointer',
                           "pointer": 9,
                           "timestamp": "9"}])

        queue = EventQueue("2")
        for pointer_val in range(1, 10):
            queue.push({"type": "pointer",
                        "pointer": pointer_val,
                        "timestamp": str(pointer_val)})
        queue.push({"type": "unknown"})
        queue.push({"type": "restart", "server_generation": "1"})
        for pointer_val in range(11, 20):
            queue.push({"type": "pointer",
                        "pointer": pointer_val,
                        "timestamp": str(pointer_val)})
        queue.push({"type": "restart", "server_generation": "2"})
        self.assertEqual(queue.contents(),
                         [{"type": "unknown",
                           "id": 9},
                          {'id': 19,
                           'type': 'pointer',
                           "pointer": 19,
                           "timestamp": "19"},
                          {"id": 20,
                           "type": "restart",
                           "server_generation": "2"}])
        for pointer_val in range(21, 23):
            queue.push({"type": "pointer",
                        "pointer": pointer_val,
                        "timestamp": str(pointer_val)})
        self.assertEqual(queue.contents(),
                         [{"type": "unknown",
                           "id": 9},
                          {'id': 19,
                           'type': 'pointer',
                           "pointer": 19,
                           "timestamp": "19"},
                          {"id": 20,
                           "type": "restart",
                           "server_generation": "2"},
                          {'id': 22,
                           'type': 'pointer',
                           "pointer": 22,
                           "timestamp": "22"},
                          ])

    def test_flag_add_collapsing(self):
        # type: () -> None
        queue = EventQueue("1")
        queue.push({"type": "update_message_flags",
                    "flag": "read",
                    "operation": "add",
                    "all": False,
                    "messages": [1, 2, 3, 4],
                    "timestamp": "1"})
        queue.push({"type": "update_message_flags",
                    "flag": "read",
                    "all": False,
                    "operation": "add",
                    "messages": [5, 6],
                    "timestamp": "1"})
        self.assertEqual(queue.contents(),
                         [{'id': 1,
                           'type': 'update_message_flags',
                           "all": False,
                           "flag": "read",
                           "operation": "add",
                           "messages": [1, 2, 3, 4, 5, 6],
                           "timestamp": "1"}])

    def test_flag_remove_collapsing(self):
        # type: () -> None
        queue = EventQueue("1")
        queue.push({"type": "update_message_flags",
                    "flag": "collapsed",
                    "operation": "remove",
                    "all": False,
                    "messages": [1, 2, 3, 4],
                    "timestamp": "1"})
        queue.push({"type": "update_message_flags",
                    "flag": "collapsed",
                    "all": False,
                    "operation": "remove",
                    "messages": [5, 6],
                    "timestamp": "1"})
        self.assertEqual(queue.contents(),
                         [{'id': 1,
                           'type': 'update_message_flags',
                           "all": False,
                           "flag": "collapsed",
                           "operation": "remove",
                           "messages": [1, 2, 3, 4, 5, 6],
                           "timestamp": "1"}])

    def test_collapse_event(self):
        # type: () -> None
        queue = EventQueue("1")
        queue.push({"type": "pointer",
                    "pointer": 1,
                    "timestamp": "1"})
        queue.push({"type": "unknown",
                    "timestamp": "1"})
        self.assertEqual(queue.contents(),
                         [{'id': 0,
                           'type': 'pointer',
                           "pointer": 1,
                           "timestamp": "1"},
                          {'id': 1,
                           'type': 'unknown',
                           "timestamp": "1"}])

class TestEventsRegisterAllPublicStreamsDefaults(TestCase):
    def setUp(self):
        # type: () -> None
        self.email = 'hamlet@zulip.com'
        self.user_profile = get_user_profile_by_email(self.email)

    def test_use_passed_all_public_true_default_false(self):
        # type: () -> None
        self.user_profile.default_all_public_streams = False
        self.user_profile.save()
        result = _default_all_public_streams(self.user_profile, True)
        self.assertTrue(result)

    def test_use_passed_all_public_true_default(self):
        # type: () -> None
        self.user_profile.default_all_public_streams = True
        self.user_profile.save()
        result = _default_all_public_streams(self.user_profile, True)
        self.assertTrue(result)

    def test_use_passed_all_public_false_default_false(self):
        # type: () -> None
        self.user_profile.default_all_public_streams = False
        self.user_profile.save()
        result = _default_all_public_streams(self.user_profile, False)
        self.assertFalse(result)

    def test_use_passed_all_public_false_default_true(self):
        # type: () -> None
        self.user_profile.default_all_public_streams = True
        self.user_profile.save()
        result = _default_all_public_streams(self.user_profile, False)
        self.assertFalse(result)

    def test_use_true_default_for_none(self):
        # type: () -> None
        self.user_profile.default_all_public_streams = True
        self.user_profile.save()
        result = _default_all_public_streams(self.user_profile, None)
        self.assertTrue(result)

    def test_use_false_default_for_none(self):
        # type: () -> None
        self.user_profile.default_all_public_streams = False
        self.user_profile.save()
        result = _default_all_public_streams(self.user_profile, None)
        self.assertFalse(result)

class TestEventsRegisterNarrowDefaults(TestCase):
    def setUp(self):
        # type: () -> None
        self.email = 'hamlet@zulip.com'
        self.user_profile = get_user_profile_by_email(self.email)
        self.stream = get_stream('Verona', self.user_profile.realm)

    def test_use_passed_narrow_no_default(self):
        # type: () -> None
        self.user_profile.default_events_register_stream_id = None
        self.user_profile.save()
        result = _default_narrow(self.user_profile, [[u'stream', u'my_stream']])
        self.assertEqual(result, [[u'stream', u'my_stream']])

    def test_use_passed_narrow_with_default(self):
        # type: () -> None
        self.user_profile.default_events_register_stream_id = self.stream.id
        self.user_profile.save()
        result = _default_narrow(self.user_profile, [[u'stream', u'my_stream']])
        self.assertEqual(result, [[u'stream', u'my_stream']])

    def test_use_default_if_narrow_is_empty(self):
        # type: () -> None
        self.user_profile.default_events_register_stream_id = self.stream.id
        self.user_profile.save()
        result = _default_narrow(self.user_profile, [])
        self.assertEqual(result, [[u'stream', u'Verona']])

    def test_use_narrow_if_default_is_none(self):
        # type: () -> None
        self.user_profile.default_events_register_stream_id = None
        self.user_profile.save()
        result = _default_narrow(self.user_profile, [])
        self.assertEqual(result, [])