hello
") self.assertIn('gravatar.com', message["avatar_url"]) message = get_message(apply_markdown=False, client_gravatar=True) self.assertEqual(message["display_recipient"], "Denmark") self.assertEqual(message["content"], "**hello**") self.assertEqual(message["avatar_url"], None) message = get_message(apply_markdown=True, client_gravatar=True) self.assertEqual(message["display_recipient"], "Denmark") self.assertEqual(message["content"], "hello
") self.assertEqual(message["avatar_url"], None) class EventsRegisterTest(ZulipTestCase): def setUp(self) -> None: super().setUp() self.user_profile = self.example_user('hamlet') def create_bot(self, email: str, **extras: Any) -> Optional[UserProfile]: return self.create_test_bot(email, self.user_profile, **extras) def realm_bot_schema(self, field_name: str, check: Validator) -> Validator: return self.check_events_dict([ ('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: Callable[[], Any], event_types: Optional[List[str]]=None, include_subscribers: bool=True, state_change_expected: bool=True, notification_settings_null: bool=False, client_gravatar: bool=False, num_events: int=1) -> List[Dict[str, Any]]: ''' Make sure we have a clean slate of client descriptors for these tests. If we don't do this, then certain failures will only manifest when you run multiple tests within a single test function. See also https://zulip.readthedocs.io/en/latest/subsystems/events-system.html#testing for details on the design of this test system. ''' clear_client_event_queues_for_testing() 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, client_gravatar = client_gravatar, 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, "", client_gravatar=True, include_subscribers=include_subscribers ) action() events = client.event_queue.contents() self.assertEqual(len(events), num_events) initial_state = copy.deepcopy(hybrid_state) post_process_state(self.user_profile, initial_state, notification_settings_null) before = ujson.dumps(initial_state) apply_events(hybrid_state, events, self.user_profile, client_gravatar=True, include_subscribers=include_subscribers) post_process_state(self.user_profile, hybrid_state, notification_settings_null) after = ujson.dumps(hybrid_state) if state_change_expected: if before == after: # nocoverage print(ujson.dumps(initial_state, indent=2)) print(events) raise AssertionError('Test does not exercise enough code -- events do not change state.') else: try: self.match_states(initial_state, copy.deepcopy(hybrid_state), events) except AssertionError: # nocoverage raise AssertionError('Test is invalid--state actually does change here.') normal_state = fetch_initial_state_data( self.user_profile, event_types, "", client_gravatar=True, include_subscribers=include_subscribers, ) post_process_state(self.user_profile, normal_state, notification_settings_null) self.match_states(hybrid_state, normal_state, events) return events def assert_on_error(self, error: Optional[str]) -> None: if error: raise AssertionError(error) def match_states(self, state1: Dict[str, Any], state2: Dict[str, Any], events: List[Dict[str, Any]]) -> None: def normalize(state: Dict[str, Any]) -> None: for u in state['never_subscribed']: if 'subscribers' in u: u['subscribers'].sort() 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) # If this assertions fails, we have unusual problems. self.assertEqual(state1.keys(), state2.keys()) # The far more likely scenario is that some section of # our enormous payload does not get updated properly. We # want the diff here to be developer-friendly, hence # the somewhat tedious code to provide useful output. if state1 != state2: # nocoverage print('\n---States DO NOT MATCH---') print('\nEVENTS:\n') # Printing out the events is a big help to # developers. import json for event in events: print(json.dumps(event, indent=4)) print('\nMISMATCHES:\n') for k in state1: if state1[k] != state2[k]: print('\nkey = ' + k) try: self.assertEqual({k: state1[k]}, {k: state2[k]}) except AssertionError as e: print(e) print(''' NOTE: This is an advanced test that verifies how we apply events after fetching data. If you do not know how to debug it, you can ask for help on chat. ''') sys.stdout.flush() raise AssertionError('Mismatching states') def check_events_dict(self, required_keys: List[Tuple[str, Validator]]) -> Validator: required_keys.append(('id', check_int)) # Raise AssertionError if `required_keys` contains duplicate items. keys = [key[0] for key in required_keys] self.assertEqual(len(keys), len(set(keys)), 'Duplicate items found in required_keys.') return check_dict_only(required_keys) def test_mentioned_send_message_events(self) -> None: user = self.example_user('hamlet') for i in range(3): content = 'mentioning... @**' + user.full_name + '** hello ' + str(i) self.do_test( lambda: self.send_stream_message(self.example_email('cordelia'), "Verona", content) ) def test_pm_send_message_events(self) -> None: self.do_test( lambda: self.send_personal_message(self.example_email('cordelia'), self.example_email('hamlet'), 'hola') ) def test_huddle_send_message_events(self) -> None: huddle = [ self.example_email('hamlet'), self.example_email('othello'), ] self.do_test( lambda: self.send_huddle_message(self.example_email('cordelia'), huddle, 'hola') ) def test_stream_send_message_events(self) -> None: def check_none(var_name: str, val: Any) -> Optional[str]: assert(val is None) return None def get_checker(check_gravatar: Validator) -> Validator: schema_checker = self.check_events_dict([ ('type', equals('message')), ('flags', check_list(None)), ('message', self.check_events_dict([ ('avatar_url', check_gravatar), ('client', check_string), ('content', check_string), ('content_type', equals('text/html')), ('display_recipient', check_string), ('is_me_message', check_bool), ('reactions', check_list(None)), ('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), ('stream_id', check_int), (TOPIC_NAME, check_string), (TOPIC_LINKS, check_list(None)), ('submessages', check_list(None)), ('timestamp', check_int), ('type', check_string), ])), ]) return schema_checker events = self.do_test( lambda: self.send_stream_message(self.example_email("hamlet"), "Verona", "hello"), client_gravatar=False, ) schema_checker = get_checker(check_gravatar=check_string) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test( lambda: self.send_stream_message(self.example_email("hamlet"), "Verona", "hello"), client_gravatar=True, ) schema_checker = get_checker(check_gravatar=check_none) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) # Verify message editing schema_checker = self.check_events_dict([ ('type', equals('update_message')), ('flags', check_list(None)), ('content', check_string), ('edit_timestamp', check_int), ('message_id', check_int), ('message_ids', check_list(check_int)), ('prior_mention_user_ids', check_list(check_int)), ('mention_user_ids', check_list(check_int)), ('presence_idle_user_ids', check_list(check_int)), ('stream_push_user_ids', check_list(check_int)), ('stream_email_user_ids', check_list(check_int)), ('push_notify_user_ids', check_list(check_int)), ('orig_content', check_string), ('orig_rendered_content', check_string), (ORIG_TOPIC, check_string), ('prev_rendered_content_version', check_int), ('propagate_mode', check_string), ('rendered_content', check_string), ('sender', check_string), ('stream_id', check_int), ('stream_name', check_string), (TOPIC_NAME, check_string), (TOPIC_LINKS, check_list(None)), ('user_id', check_int), ('is_me_message', check_bool), ]) message = Message.objects.order_by('-id')[0] topic = 'new_topic' propagate_mode = 'change_all' content = 'new content' rendered_content = render_markdown(message, content) prior_mention_user_ids = set() # type: Set[int] mentioned_user_ids = set() # type: Set[int] events = self.do_test( lambda: do_update_message(self.user_profile, message, topic, propagate_mode, content, rendered_content, prior_mention_user_ids, mentioned_user_ids), state_change_expected=True, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) # Verify do_update_embedded_data schema_checker = self.check_events_dict([ ('type', equals('update_message')), ('flags', check_list(None)), ('content', check_string), ('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", "embed_content
"), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_update_message_flags(self) -> None: # Test message flag update events schema_checker = self.check_events_dict([ ('all', check_bool), ('type', equals('update_message_flags')), ('flag', check_string), ('messages', check_list(check_int)), ('operation', equals("add")), ]) message = self.send_personal_message( self.example_email("cordelia"), self.example_email("hamlet"), "hello", ) user_profile = self.example_user('hamlet') events = self.do_test( lambda: do_update_message_flags(user_profile, get_client("website"), 'add', 'starred', [message]), state_change_expected=True, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) schema_checker = self.check_events_dict([ ('all', check_bool), ('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, get_client("website"), 'remove', 'starred', [message]), state_change_expected=True, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_update_read_flag_removes_unread_msg_ids(self) -> None: user_profile = self.example_user('hamlet') mention = '@**' + user_profile.full_name + '**' for content in ['hello', mention]: message = self.send_stream_message( self.example_email('cordelia'), "Verona", content ) self.do_test( lambda: do_update_message_flags(user_profile, get_client("website"), 'add', 'read', [message]), state_change_expected=True, ) def test_send_message_to_existing_recipient(self) -> None: self.send_stream_message( self.example_email('cordelia'), "Verona", "hello 1" ) self.do_test( lambda: self.send_stream_message("cordelia@zulip.com", "Verona", "hello 2"), state_change_expected=True, ) def test_add_reaction_legacy(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('reaction')), ('op', equals('add')), ('message_id', check_int), ('emoji_name', check_string), ('emoji_code', check_string), ('reaction_type', check_string), ('user', check_dict_only([ ('email', check_string), ('full_name', check_string), ('user_id', check_int) ])), ]) message_id = self.send_stream_message(self.example_email("hamlet"), "Verona", "hello") message = Message.objects.get(id=message_id) events = self.do_test( lambda: do_add_reaction_legacy( 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_legacy(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('reaction')), ('op', equals('remove')), ('message_id', check_int), ('emoji_name', check_string), ('emoji_code', check_string), ('reaction_type', check_string), ('user', check_dict_only([ ('email', check_string), ('full_name', check_string), ('user_id', check_int) ])), ]) message_id = self.send_stream_message(self.example_email("hamlet"), "Verona", "hello") message = Message.objects.get(id=message_id) do_add_reaction_legacy(self.user_profile, message, "tada") events = self.do_test( lambda: do_remove_reaction_legacy( self.user_profile, message, "tada"), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_add_reaction(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('reaction')), ('op', equals('add')), ('message_id', check_int), ('emoji_name', check_string), ('emoji_code', check_string), ('reaction_type', check_string), ('user', check_dict_only([ ('email', check_string), ('full_name', check_string), ('user_id', check_int) ])), ]) message_id = self.send_stream_message(self.example_email("hamlet"), "Verona", "hello") message = Message.objects.get(id=message_id) events = self.do_test( lambda: do_add_reaction( self.user_profile, message, "tada", "1f389", "unicode_emoji"), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_add_submessage(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('submessage')), ('message_id', check_int), ('submessage_id', check_int), ('sender_id', check_int), ('msg_type', check_string), ('content', check_string), ]) cordelia = self.example_user('cordelia') stream_name = 'Verona' message_id = self.send_stream_message( sender_email=cordelia.email, stream_name=stream_name, ) events = self.do_test( lambda: do_add_submessage( realm=cordelia.realm, sender_id=cordelia.id, message_id=message_id, msg_type='whatever', content='"stuff"', ), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_remove_reaction(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('reaction')), ('op', equals('remove')), ('message_id', check_int), ('emoji_name', check_string), ('emoji_code', check_string), ('reaction_type', check_string), ('user', check_dict_only([ ('email', check_string), ('full_name', check_string), ('user_id', check_int) ])), ]) message_id = self.send_stream_message(self.example_email("hamlet"), "Verona", "hello") message = Message.objects.get(id=message_id) do_add_reaction(self.user_profile, message, "tada", "1f389", "unicode_emoji") events = self.do_test( lambda: do_remove_reaction( self.user_profile, message, "1f389", "unicode_emoji"), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_invite_user_event(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('invites_changed')), ]) self.user_profile = self.example_user('iago') streams = [] for stream_name in ["Denmark", "Scotland"]: streams.append(get_stream(stream_name, self.user_profile.realm)) events = self.do_test( lambda: do_invite_users(self.user_profile, ["foo@zulip.com"], streams, False), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_create_multiuse_invite_event(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('invites_changed')), ]) self.user_profile = self.example_user('iago') streams = [] for stream_name in ["Denmark", "Verona"]: streams.append(get_stream(stream_name, self.user_profile.realm)) events = self.do_test( lambda: do_create_multiuse_invite_link(self.user_profile, PreregistrationUser.INVITE_AS['MEMBER'], streams), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_revoke_user_invite_event(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('invites_changed')), ]) self.user_profile = self.example_user('iago') streams = [] for stream_name in ["Denmark", "Verona"]: streams.append(get_stream(stream_name, self.user_profile.realm)) do_invite_users(self.user_profile, ["foo@zulip.com"], streams, False) prereg_users = PreregistrationUser.objects.filter(referred_by__realm=self.user_profile.realm) events = self.do_test( lambda: do_revoke_user_invite(prereg_users[0]), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_revoke_multiuse_invite_event(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('invites_changed')), ]) self.user_profile = self.example_user('iago') streams = [] for stream_name in ["Denmark", "Verona"]: streams.append(get_stream(stream_name, self.user_profile.realm)) do_create_multiuse_invite_link(self.user_profile, PreregistrationUser.INVITE_AS['MEMBER'], streams) multiuse_object = MultiuseInvite.objects.get() events = self.do_test( lambda: do_revoke_multi_use_invite(multiuse_object), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_invitation_accept_invite_event(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('invites_changed')), ]) self.user_profile = self.example_user('iago') streams = [] for stream_name in ["Denmark", "Scotland"]: streams.append(get_stream(stream_name, self.user_profile.realm)) do_invite_users(self.user_profile, ["foo@zulip.com"], streams, False) prereg_users = PreregistrationUser.objects.get(email="foo@zulip.com") events = self.do_test( lambda: do_create_user('foo@zulip.com', 'password', self.user_profile.realm, 'full name', 'short name', prereg_user=prereg_users), state_change_expected=True, num_events=5, ) error = schema_checker('events[4]', events[4]) self.assert_on_error(error) def test_typing_events(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('typing')), ('op', equals('start')), ('sender', check_dict_only([ ('email', check_string), ('user_id', check_int)])), ('recipients', check_list(check_dict_only([ ('email', check_string), ('user_id', check_int), ]))), ]) events = self.do_test( lambda: check_send_typing_notification( self.user_profile, [self.example_email("cordelia")], "start"), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_get_typing_user_profiles(self) -> None: """ Make sure we properly assert failures for recipient types that should not get typing... notifications. """ sender_profile = self.example_user('cordelia') stream = get_stream('Rome', sender_profile.realm) # Test stream with self.assertRaisesRegex(ValueError, 'not supported for streams'): recipient = Recipient.objects.get(type_id=stream.id, type=Recipient.STREAM) get_typing_user_profiles(recipient, sender_profile.id) # Test some other recipient type with self.assertRaisesRegex(ValueError, 'Bad recipient type'): recipient = Recipient(type=999) # invalid type get_typing_user_profiles(recipient, sender_profile.id) def test_custom_profile_fields_events(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('custom_profile_fields')), ('op', equals('add')), ('fields', check_list(check_dict_only([ ('id', check_int), ('type', check_int), ('name', check_string), ('hint', check_string), ('field_data', check_string), ('order', check_int), ]))), ]) events = self.do_test( lambda: notify_realm_custom_profile_fields( self.user_profile.realm, 'add'), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) realm = self.user_profile.realm field = realm.customprofilefield_set.get(realm=realm, name='Biography') name = field.name hint = 'Biography of the user' try_update_realm_custom_profile_field(realm, field, name, hint=hint) events = self.do_test( lambda: notify_realm_custom_profile_fields( self.user_profile.realm, 'add'), state_change_expected=False, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_custom_profile_field_data_events(self) -> None: schema_checker_basic = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('user_id', check_int), ('custom_profile_field', check_dict([ ('id', check_int), ('value', check_none_or(check_string)), ])), ])), ]) schema_checker_with_rendered_value = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('user_id', check_int), ('custom_profile_field', check_dict([ ('id', check_int), ('value', check_none_or(check_string)), ('rendered_value', check_none_or(check_string)), ])), ])), ]) field_id = self.user_profile.realm.customprofilefield_set.get( realm=self.user_profile.realm, name='Biography').id field = { "id": field_id, "value": "New value", } events = self.do_test(lambda: do_update_user_custom_profile_data(self.user_profile, [field])) error = schema_checker_with_rendered_value('events[0]', events[0]) self.assert_on_error(error) # Test we pass correct stringify value in custom-user-field data event field_id = self.user_profile.realm.customprofilefield_set.get( realm=self.user_profile.realm, name='Mentor').id field = { "id": field_id, "value": [self.example_user("ZOE").id], } events = self.do_test(lambda: do_update_user_custom_profile_data(self.user_profile, [field])) error = schema_checker_basic('events[0]', events[0]) self.assert_on_error(error) def test_presence_events(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('presence')), ('email', check_string), ('server_timestamp', check_float), ('presence', check_dict_only([ ('website', check_dict_only([ ('status', equals('active')), ('timestamp', check_int), ('client', check_string), ('pushable', check_bool), ])), ])), ]) events = self.do_test(lambda: do_update_user_presence( self.user_profile, get_client("website"), timezone_now(), UserPresence.ACTIVE)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_presence_events_multiple_clients(self) -> None: schema_checker_android = self.check_events_dict([ ('type', equals('presence')), ('email', check_string), ('server_timestamp', check_float), ('presence', check_dict_only([ ('ZulipAndroid/1.0', check_dict_only([ ('status', equals('idle')), ('timestamp', check_int), ('client', check_string), ('pushable', check_bool), ])), ])), ]) self.api_post(self.user_profile.email, "/api/v1/users/me/presence", {'status': 'idle'}, HTTP_USER_AGENT="ZulipAndroid/1.0") self.do_test(lambda: do_update_user_presence( self.user_profile, get_client("website"), timezone_now(), UserPresence.ACTIVE)) events = self.do_test(lambda: do_update_user_presence( self.user_profile, get_client("ZulipAndroid/1.0"), timezone_now(), UserPresence.IDLE)) error = schema_checker_android('events[0]', events[0]) self.assert_on_error(error) def test_pointer_events(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('pointer')), ('pointer', check_int) ]) events = self.do_test(lambda: do_update_pointer(self.user_profile, get_client("website"), 1500)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_register_events(self) -> None: realm_user_add_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('add')), ('person', check_dict_only([ ('user_id', check_int), ('email', check_string), ('avatar_url', check_none_or(check_string)), ('full_name', check_string), ('is_admin', check_bool), ('is_bot', check_bool), ('is_guest', check_bool), ('profile_data', check_dict_only([])), ('timezone', check_string), ('date_joined', check_string), ])), ]) 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) new_user_profile = get_user_by_delivery_email("test1@zulip.com", self.user_profile.realm) self.assertEqual(new_user_profile.email, "test1@zulip.com") def test_register_events_email_address_visibility(self) -> None: realm_user_add_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('add')), ('person', check_dict_only([ ('user_id', check_int), ('email', check_string), ('avatar_url', check_none_or(check_string)), ('full_name', check_string), ('is_admin', check_bool), ('is_bot', check_bool), ('is_guest', check_bool), ('profile_data', check_dict_only([])), ('timezone', check_string), ('date_joined', check_string), ])), ]) do_set_realm_property(self.user_profile.realm, "email_address_visibility", Realm.EMAIL_ADDRESS_VISIBILITY_ADMINS) 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) new_user_profile = get_user_by_delivery_email("test1@zulip.com", self.user_profile.realm) self.assertEqual(new_user_profile.email, "user%s@zulip.testserver" % (new_user_profile.id,)) def test_alert_words_events(self) -> None: alert_words_checker = self.check_events_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_away_events(self) -> None: checker = self.check_events_dict([ ('type', equals('user_status')), ('user_id', check_int), ('away', check_bool), ('status_text', check_string), ]) client = get_client("website") events = self.do_test(lambda: do_update_user_status(user_profile=self.user_profile, away=True, status_text='out to lunch', client_id=client.id)) error = checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_update_user_status(user_profile=self.user_profile, away=False, status_text='', client_id=client.id)) error = checker('events[0]', events[0]) self.assert_on_error(error) def test_user_group_events(self) -> None: user_group_add_checker = self.check_events_dict([ ('type', equals('user_group')), ('op', equals('add')), ('group', check_dict_only([ ('id', check_int), ('name', check_string), ('members', check_list(check_int)), ('description', check_string), ])), ]) othello = self.example_user('othello') events = self.do_test(lambda: check_add_user_group(self.user_profile.realm, 'backend', [othello], 'Backend team')) error = user_group_add_checker('events[0]', events[0]) self.assert_on_error(error) # Test name update user_group_update_checker = self.check_events_dict([ ('type', equals('user_group')), ('op', equals('update')), ('group_id', check_int), ('data', check_dict_only([ ('name', check_string), ])), ]) backend = UserGroup.objects.get(name='backend') events = self.do_test(lambda: do_update_user_group_name(backend, 'backendteam')) error = user_group_update_checker('events[0]', events[0]) self.assert_on_error(error) # Test description update user_group_update_checker = self.check_events_dict([ ('type', equals('user_group')), ('op', equals('update')), ('group_id', check_int), ('data', check_dict_only([ ('description', check_string), ])), ]) description = "Backend team to deal with backend code." events = self.do_test(lambda: do_update_user_group_description(backend, description)) error = user_group_update_checker('events[0]', events[0]) self.assert_on_error(error) # Test add members user_group_add_member_checker = self.check_events_dict([ ('type', equals('user_group')), ('op', equals('add_members')), ('group_id', check_int), ('user_ids', check_list(check_int)), ]) hamlet = self.example_user('hamlet') events = self.do_test(lambda: bulk_add_members_to_user_group(backend, [hamlet])) error = user_group_add_member_checker('events[0]', events[0]) self.assert_on_error(error) # Test remove members user_group_remove_member_checker = self.check_events_dict([ ('type', equals('user_group')), ('op', equals('remove_members')), ('group_id', check_int), ('user_ids', check_list(check_int)), ]) hamlet = self.example_user('hamlet') events = self.do_test(lambda: remove_members_from_user_group(backend, [hamlet])) error = user_group_remove_member_checker('events[0]', events[0]) self.assert_on_error(error) # Test delete event user_group_remove_checker = self.check_events_dict([ ('type', equals('user_group')), ('op', equals('remove')), ('group_id', check_int), ]) events = self.do_test(lambda: check_delete_user_group(backend.id, othello)) error = user_group_remove_checker('events[0]', events[0]) self.assert_on_error(error) def test_default_stream_groups_events(self) -> None: default_stream_groups_checker = self.check_events_dict([ ('type', equals('default_stream_groups')), ('default_stream_groups', check_list(check_dict_only([ ('name', check_string), ('id', check_int), ('description', check_string), ('streams', check_list(check_dict_only([ ('description', check_string), ('rendered_description', check_string), ('invite_only', check_bool), ('is_web_public', check_bool), ('is_announcement_only', check_bool), ('name', check_string), ('stream_id', check_int), ('first_message_id', check_none_or(check_int)), ('history_public_to_subscribers', check_bool)]))), ]))), ]) streams = [] for stream_name in ["Scotland", "Verona", "Denmark"]: streams.append(get_stream(stream_name, self.user_profile.realm)) events = self.do_test(lambda: do_create_default_stream_group( self.user_profile.realm, "group1", "This is group1", streams)) error = default_stream_groups_checker('events[0]', events[0]) self.assert_on_error(error) group = lookup_default_stream_groups(["group1"], self.user_profile.realm)[0] venice_stream = get_stream("Venice", self.user_profile.realm) events = self.do_test(lambda: do_add_streams_to_default_stream_group(self.user_profile.realm, group, [venice_stream])) error = default_stream_groups_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_remove_streams_from_default_stream_group(self.user_profile.realm, group, [venice_stream])) error = default_stream_groups_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_change_default_stream_group_description(self.user_profile.realm, group, "New description")) error = default_stream_groups_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_change_default_stream_group_name(self.user_profile.realm, group, "New Group Name")) error = default_stream_groups_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_remove_default_stream_group(self.user_profile.realm, group)) error = default_stream_groups_checker('events[0]', events[0]) self.assert_on_error(error) def test_default_stream_group_events_guest(self) -> None: streams = [] for stream_name in ["Scotland", "Verona", "Denmark"]: streams.append(get_stream(stream_name, self.user_profile.realm)) do_create_default_stream_group(self.user_profile.realm, "group1", "This is group1", streams) group = lookup_default_stream_groups(["group1"], self.user_profile.realm)[0] do_change_is_guest(self.user_profile, True) venice_stream = get_stream("Venice", self.user_profile.realm) self.do_test(lambda: do_add_streams_to_default_stream_group(self.user_profile.realm, group, [venice_stream]), state_change_expected = False, num_events=0) def test_default_streams_events(self) -> None: default_streams_checker = self.check_events_dict([ ('type', equals('default_streams')), ('default_streams', check_list(check_dict_only([ ('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_default_streams_events_guest(self) -> None: do_change_is_guest(self.user_profile, True) stream = get_stream("Scotland", self.user_profile.realm) self.do_test(lambda: do_add_default_stream(stream), state_change_expected = False, num_events=0) self.do_test(lambda: do_remove_default_stream(stream), state_change_expected = False, num_events=0) def test_muted_topics_events(self) -> None: muted_topics_checker = self.check_events_dict([ ('type', equals('muted_topics')), ('muted_topics', check_list(check_list(check_string, 2))), ]) stream = get_stream('Denmark', self.user_profile.realm) recipient = get_stream_recipient(stream.id) events = self.do_test(lambda: do_mute_topic( self.user_profile, stream, recipient, "topic")) error = muted_topics_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_unmute_topic( self.user_profile, stream, "topic")) error = muted_topics_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_avatar_fields(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('email', check_string), ('user_id', check_int), ('avatar_url', check_string), ('avatar_url_medium', check_string), ('avatar_source', 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) schema_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('email', check_string), ('user_id', check_int), ('avatar_url', check_none_or(check_string)), ('avatar_url_medium', check_none_or(check_string)), ('avatar_source', check_string), ])), ]) events = self.do_test( lambda: do_change_avatar_fields(self.user_profile, UserProfile.AVATAR_FROM_GRAVATAR), ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_full_name(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('email', check_string), ('full_name', check_string), ('user_id', check_int), ])), ]) events = self.do_test(lambda: do_change_full_name(self.user_profile, 'Sir Hamlet', self.user_profile)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_user_delivery_email_email_address_visibilty_admins(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('delivery_email', check_string), ('user_id', check_int), ])), ]) do_set_realm_property(self.user_profile.realm, "email_address_visibility", Realm.EMAIL_ADDRESS_VISIBILITY_ADMINS) # Important: We need to refresh from the database here so that # we don't have a stale UserProfile object with an old value # for email being passed into this next function. self.user_profile.refresh_from_db() action = lambda: do_change_user_delivery_email(self.user_profile, 'newhamlet@zulip.com') events = self.do_test(action, num_events=1) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def do_set_realm_property_test(self, name: str) -> None: bool_tests = [True, False, True] # type: List[bool] test_values = dict( default_language=[u'es', u'de', u'en'], description=[u'Realm description', u'New description'], digest_weekday=[0, 1, 2], message_retention_days=[10, 20], name=[u'Zulip', u'New Name'], waiting_period_threshold=[10, 20], create_stream_policy=[3, 2, 1], invite_to_stream_policy=[3, 2, 1], email_address_visibility=[Realm.EMAIL_ADDRESS_VISIBILITY_ADMINS], bot_creation_policy=[Realm.BOT_CREATION_EVERYONE], video_chat_provider=[ Realm.VIDEO_CHAT_PROVIDERS['jitsi_meet']['id'], Realm.VIDEO_CHAT_PROVIDERS['google_hangouts']['id'] ], google_hangouts_domain=[u"zulip.com", u"zulip.org"], zoom_api_secret=[u"abc", u"xyz"], zoom_api_key=[u"abc", u"xyz"], zoom_user_id=[u"example@example.com", u"example@example.org"] ) # type: Dict[str, Any] vals = test_values.get(name) property_type = Realm.property_types[name] if property_type is bool: validator = check_bool vals = bool_tests elif property_type is str: 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 = self.check_events_dict([ ('type', equals('realm')), ('op', equals('update')), ('property', equals(name)), ('value', validator), ]) 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:]: state_change_expected = True if name == "zoom_api_secret": state_change_expected = False events = self.do_test( lambda: do_set_realm_property(self.user_profile.realm, name, val), state_change_expected=state_change_expected) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) @slow("Actually runs several full-stack fetching tests") def test_change_realm_property(self) -> None: for prop in Realm.property_types: with self.settings(SEND_DIGEST_EMAILS=True): self.do_set_realm_property_test(prop) @slow("Runs a large matrix of tests") def test_change_realm_authentication_methods(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm')), ('op', equals('update_dict')), ('property', equals('default')), ('data', check_dict_only([ ('authentication_methods', check_dict([])) ])), ]) def fake_backends() -> Any: backends = ( 'zproject.backends.DevAuthBackend', 'zproject.backends.EmailAuthBackend', 'zproject.backends.GitHubAuthBackend', 'zproject.backends.GoogleAuthBackend', '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) -> None: schema_checker = self.check_events_dict([ ('type', equals('subscription')), ('op', equals('update')), ('property', equals('pin_to_top')), ('stream_id', check_int), ('value', check_bool), ('name', check_string), ('email', check_string), ]) 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_stream_notification_settings(self) -> None: for setting_name in ['email_notifications']: schema_checker = self.check_events_dict([ ('type', equals('subscription')), ('op', equals('update')), ('property', equals(setting_name)), ('stream_id', check_int), ('value', check_bool), ('name', check_string), ('email', check_string), ]) stream = get_stream("Denmark", self.user_profile.realm) sub = get_subscription(stream.name, self.user_profile) # First test with notification_settings_null enabled for value in (True, False): events = self.do_test(lambda: do_change_subscription_property(self.user_profile, sub, stream, setting_name, value), notification_settings_null=True) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) for value in (True, False): events = self.do_test(lambda: do_change_subscription_property(self.user_profile, sub, stream, setting_name, value)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) @slow("Runs a matrix of 6 queries to the /home view") def test_change_realm_message_edit_settings(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm')), ('op', equals('update_dict')), ('property', equals('default')), ('data', check_dict_only([ ('allow_message_editing', check_bool), ('message_content_edit_limit_seconds', check_int), ('allow_community_topic_editing', check_bool), ])), ]) # 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), (False, 1234), (True, 600), (False, 0), (True, 1234)): events = self.do_test( lambda: do_set_realm_message_editing(self.user_profile.realm, allow_message_editing, message_content_edit_limit_seconds, False)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_realm_notifications_stream(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm')), ('op', equals('update')), ('property', equals('notifications_stream_id')), ('value', check_int), ]) stream = get_stream("Rome", self.user_profile.realm) for notifications_stream, notifications_stream_id in ((stream, stream.id), (None, -1)): events = self.do_test( lambda: do_set_realm_notifications_stream(self.user_profile.realm, notifications_stream, notifications_stream_id)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_realm_signup_notifications_stream(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm')), ('op', equals('update')), ('property', equals('signup_notifications_stream_id')), ('value', check_int), ]) stream = get_stream("Rome", self.user_profile.realm) for signup_notifications_stream, signup_notifications_stream_id in ((stream, stream.id), (None, -1)): events = self.do_test( lambda: do_set_realm_signup_notifications_stream(self.user_profile.realm, signup_notifications_stream, signup_notifications_stream_id)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_is_admin(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('email', check_string), ('is_admin', check_bool), ('user_id', check_int), ])), ]) 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: str) -> None: """Test updating each setting in UserProfile.property_types dict.""" test_changes = dict( emojiset = [u'apple', u'twitter'], default_language = [u'es', u'de', u'en'], timezone = [u'US/Mountain', u'US/Samoa', u'Pacific/Galapogos', u''], demote_inactive_streams = [2, 3, 1], ) # type: Dict[str, Any] property_type = UserProfile.property_types[setting_name] if property_type is bool: validator = check_bool elif property_type is str: validator = check_string elif property_type is int: validator = check_int else: raise AssertionError("Unexpected property type %s" % (property_type,)) num_events = 1 if setting_name == "timezone": num_events = 2 values = test_changes.get(setting_name) if property_type is bool: if getattr(self.user_profile, setting_name) is False: values = [True, False, True] else: values = [False, True, False] if values is None: raise AssertionError('No test created for %s' % (setting_name,)) for value in values: events = self.do_test(lambda: do_set_user_display_setting( self.user_profile, setting_name, value), num_events=num_events) schema_checker = self.check_events_dict([ ('type', equals('update_display_settings')), ('setting_name', equals(setting_name)), ('user', check_string), ('setting', validator), ]) language_schema_checker = self.check_events_dict([ ('type', equals('update_display_settings')), ('language_name', check_string), ('setting_name', equals(setting_name)), ('user', check_string), ('setting', validator), ]) if setting_name == "default_language": error = language_schema_checker('events[0]', events[0]) else: error = schema_checker('events[0]', events[0]) self.assert_on_error(error) timezone_schema_checker = self.check_events_dict([ ('type', equals('realm_user')), ('op', equals('update')), ('person', check_dict_only([ ('email', check_string), ('user_id', check_int), ('timezone', check_string), ])), ]) if setting_name == "timezone": error = timezone_schema_checker('events[1]', events[1]) @slow("Actually runs several full-stack fetching tests") def test_set_user_display_settings(self) -> None: for prop in UserProfile.property_types: self.do_set_user_display_settings_test(prop) @slow("Actually runs several full-stack fetching tests") def test_change_notification_settings(self) -> None: for notification_setting, v in self.user_profile.notification_setting_types.items(): if notification_setting in ["notification_sound", "desktop_icon_count_display"]: # These settings are tested in their own tests. continue schema_checker = self.check_events_dict([ ('type', equals('update_global_notifications')), ('notification_name', equals(notification_setting)), ('user', check_string), ('setting', check_bool), ]) do_change_notification_settings(self.user_profile, notification_setting, False) for setting_value in [True, False]: events = self.do_test(lambda: do_change_notification_settings( self.user_profile, notification_setting, setting_value, log=False)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) # Also test with notification_settings_null=True events = self.do_test( lambda: do_change_notification_settings( self.user_profile, notification_setting, setting_value, log=False), notification_settings_null=True, state_change_expected=False) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_notification_sound(self) -> None: notification_setting = "notification_sound" schema_checker = self.check_events_dict([ ('type', equals('update_global_notifications')), ('notification_name', equals(notification_setting)), ('user', check_string), ('setting', equals("ding")), ]) events = self.do_test(lambda: do_change_notification_settings( self.user_profile, notification_setting, 'ding', log=False)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_change_desktop_icon_count_display(self) -> None: notification_setting = "desktop_icon_count_display" schema_checker = self.check_events_dict([ ('type', equals('update_global_notifications')), ('notification_name', equals(notification_setting)), ('user', check_string), ('setting', equals(2)), ]) events = self.do_test(lambda: do_change_notification_settings( self.user_profile, notification_setting, 2, log=False)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) schema_checker = self.check_events_dict([ ('type', equals('update_global_notifications')), ('notification_name', equals(notification_setting)), ('user', check_string), ('setting', equals(1)), ]) events = self.do_test(lambda: do_change_notification_settings( self.user_profile, notification_setting, 1, log=False)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_realm_update_plan_type(self) -> None: realm = self.user_profile.realm state_data = fetch_initial_state_data(self.user_profile, None, "", False) self.assertEqual(state_data['realm_plan_type'], Realm.SELF_HOSTED) self.assertEqual(state_data['plan_includes_wide_organization_logo'], True) schema_checker = self.check_events_dict([ ('type', equals('realm')), ('op', equals('update')), ('property', equals('plan_type')), ('value', equals(Realm.LIMITED)), ('extra_data', check_dict_only([ ('upload_quota', check_int) ])), ]) events = self.do_test(lambda: do_change_plan_type(realm, Realm.LIMITED)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) state_data = fetch_initial_state_data(self.user_profile, None, "", False) self.assertEqual(state_data['realm_plan_type'], Realm.LIMITED) self.assertEqual(state_data['plan_includes_wide_organization_logo'], False) def test_realm_emoji_events(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('realm_emoji')), ('op', equals('update')), ('realm_emoji', check_dict([])), ]) author = self.example_user('iago') with get_test_image_file('img.png') as img_file: events = self.do_test(lambda: check_add_realm_emoji(self.user_profile.realm, "my_emoji", author, img_file)) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) events = self.do_test(lambda: do_remove_realm_emoji(self.user_profile.realm, "my_emoji")) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_realm_filter_events(self) -> None: schema_checker = self.check_events_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(self.user_profile.realm, "#(?PKing Hamlet renamed stream old_name to stream with a brand new name.
' action = lambda: do_rename_stream(stream, new_name, self.user_profile) events = self.do_test(action, num_events=3) schema_checker = self.check_events_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 = self.check_events_dict([ ('type', equals('stream')), ('op', equals('update')), ('property', equals('name')), ('value', equals(new_name)), ('name', equals('old_name')), ('stream_id', check_int), ]) error = schema_checker('events[1]', events[1]) self.assert_on_error(error) schema_checker = check_dict([ ('flags', check_list(check_string)), ('type', equals('message')), ('message', check_dict([ ('timestamp', check_int), ('content', equals(notification)), ('content_type', equals('text/html')), ('sender_email', equals('notification-bot@zulip.com')), ('sender_id', check_int), ('sender_short_name', equals('notification-bot')), ('display_recipient', equals(new_name)), ('id', check_int), ('stream_id', check_int), ('sender_realm_str', check_string), ('sender_full_name', equals('Notification Bot')), ('is_me_message', equals(False)), ('type', equals('stream')), ('submessages', check_list(check_string)), (TOPIC_LINKS, check_list(check_url)), ('avatar_url', check_url), ('reactions', check_list(None)), ('client', equals('Internal')), (TOPIC_NAME, equals('stream events')), ('recipient_id', check_int) ])), ('id', check_int) ]) error = schema_checker('events[2]', events[2]) self.assert_on_error(error) def test_deactivate_stream_neversubscribed(self) -> None: stream = self.make_stream('old_name') action = lambda: do_deactivate_stream(stream) events = self.do_test(action) schema_checker = self.check_events_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) -> None: action = lambda: self.subscribe(self.example_user("othello"), u"test_stream") events = self.do_test(action, num_events=2) peer_add_schema_checker = self.check_events_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) @slow("Actually several tests combined together") def test_subscribe_events(self) -> None: self.do_test_subscribe_events(include_subscribers=True) @slow("Actually several tests combined together") def test_subscribe_events_no_include_subscribers(self) -> None: self.do_test_subscribe_events(include_subscribers=False) def do_test_subscribe_events(self, include_subscribers: bool) -> None: subscription_fields = [ ('color', check_string), ('description', check_string), ('rendered_description', check_string), ('email_address', check_string), ('invite_only', check_bool), ('is_web_public', check_bool), ('is_announcement_only', check_bool), ('is_muted', check_bool), ('in_home_view', check_bool), ('name', check_string), ('audible_notifications', check_none_or(check_bool)), ('email_notifications', check_none_or(check_bool)), ('desktop_notifications', check_none_or(check_bool)), ('push_notifications', check_none_or(check_bool)), ('stream_id', check_int), ('first_message_id', check_none_or(check_int)), ('history_public_to_subscribers', check_bool), ('pin_to_top', check_bool), ('stream_weekly_traffic', check_none_or(check_int)), ('is_old_stream', check_bool), ] if include_subscribers: subscription_fields.append(('subscribers', check_list(check_int))) # type: ignore subscription_schema_checker = check_list( check_dict_only(subscription_fields), ) stream_create_schema_checker = self.check_events_dict([ ('type', equals('stream')), ('op', equals('create')), ('streams', check_list(check_dict_only([ ('name', check_string), ('stream_id', check_int), ('invite_only', check_bool), ('description', check_string), ('rendered_description', check_string), ]))), ]) add_schema_checker = self.check_events_dict([ ('type', equals('subscription')), ('op', equals('add')), ('subscriptions', subscription_schema_checker), ]) remove_schema_checker = self.check_events_dict([ ('type', equals('subscription')), ('op', equals('remove')), ('subscriptions', check_list( check_dict_only([ ('name', equals('test_stream')), ('stream_id', check_int), ]), )), ]) peer_add_schema_checker = self.check_events_dict([ ('type', equals('subscription')), ('op', equals('peer_add')), ('user_id', check_int), ('subscriptions', check_list(check_string)), ]) peer_remove_schema_checker = self.check_events_dict([ ('type', equals('subscription')), ('op', equals('peer_remove')), ('user_id', check_int), ('subscriptions', check_list(check_string)), ]) stream_update_schema_checker = self.check_events_dict([ ('type', equals('stream')), ('op', equals('update')), ('property', equals('description')), ('value', check_string), ('rendered_description', check_string), ('stream_id', check_int), ('name', check_string), ]) stream_update_invite_only_schema_checker = self.check_events_dict([ ('type', equals('stream')), ('op', equals('update')), ('property', equals('invite_only')), ('stream_id', check_int), ('name', check_string), ('value', check_bool), ('history_public_to_subscribers', check_bool), ]) stream_update_is_announcement_only_schema_checker = self.check_events_dict([ ('type', equals('stream')), ('op', equals('update')), ('property', equals('is_announcement_only')), ('stream_id', check_int), ('name', check_string), ('value', check_bool), ]) # Subscribe to a totally new stream, so it's just Hamlet on it action = lambda: self.subscribe(self.example_user("hamlet"), "test_stream") # type: Callable[[], Any] 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(self.example_user("othello"), "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( [self.example_user('othello')], [stream], get_client("website")) 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( [self.example_user('hamlet')], [stream], get_client("website")) events = self.do_test(action, include_subscribers=include_subscribers, num_events=3) error = remove_schema_checker('events[0]', events[0]) self.assert_on_error(error) # Now resubscribe a user, to make sure that works on a vacated stream action = lambda: self.subscribe(self.example_user("hamlet"), "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) # Update stream privacy action = lambda: do_change_stream_invite_only(stream, True, history_public_to_subscribers=True) events = self.do_test(action, include_subscribers=include_subscribers) error = stream_update_invite_only_schema_checker('events[0]', events[0]) self.assert_on_error(error) # Update stream is_announcement_only property action = lambda: do_change_stream_announcement_only(stream, True) events = self.do_test(action, include_subscribers=include_subscribers) error = stream_update_is_announcement_only_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", self.user_profile.realm, invite_only=True) user_profile = self.example_user('hamlet') 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) def test_do_delete_message_stream(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('delete_message')), ('message_id', check_int), ('sender', check_string), ('sender_id', check_int), ('message_type', equals("stream")), ('stream_id', check_int), ('topic', check_string), ]) msg_id = self.send_stream_message("hamlet@zulip.com", "Verona") message = Message.objects.get(id=msg_id) events = self.do_test( lambda: do_delete_messages(self.user_profile, [message]), state_change_expected=True, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_do_delete_message_personal(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('delete_message')), ('message_id', check_int), ('sender', check_string), ('sender_id', check_int), ('message_type', equals("private")), ('recipient_id', check_int), ]) msg_id = self.send_personal_message( self.example_email("cordelia"), self.user_profile.email, "hello", ) message = Message.objects.get(id=msg_id) events = self.do_test( lambda: do_delete_messages(self.user_profile, [message]), state_change_expected=True, ) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_do_delete_message_no_max_id(self) -> None: user_profile = self.example_user('aaron') # Delete all historical messages for this user user_profile = self.example_user('hamlet') UserMessage.objects.filter(user_profile=user_profile).delete() msg_id = self.send_stream_message("hamlet@zulip.com", "Verona") message = Message.objects.get(id=msg_id) self.do_test( lambda: do_delete_messages(self.user_profile, [message]), state_change_expected=True, ) result = fetch_initial_state_data(user_profile, None, "", client_gravatar=False) self.assertEqual(result['max_message_id'], -1) def test_add_attachment(self) -> None: schema_checker = self.check_events_dict([ ('type', equals('attachment')), ('op', equals('add')), ('attachment', check_dict_only([ ('id', check_int), ('name', check_string), ('size', check_int), ('path_id', check_string), ('create_time', check_float), ('messages', check_list(check_dict_only([ ('id', check_int), ('name', check_float), ]))), ])), ('upload_space_used', equals(6)), ]) self.login(self.example_email("hamlet")) fp = StringIO("zulip!") fp.name = "zulip.txt" data = {'uri': None} def do_upload() -> None: result = self.client_post("/json/user_uploads", {'file': fp}) self.assert_json_success(result) self.assertIn("uri", result.json()) uri = result.json()["uri"] base = '/user_uploads/' self.assertEqual(base, uri[:len(base)]) data['uri'] = uri events = self.do_test( lambda: do_upload(), num_events=1, state_change_expected=False) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) # Verify that the DB has the attachment marked as unclaimed entry = Attachment.objects.get(file_name='zulip.txt') self.assertEqual(entry.is_claimed(), False) # Now we send an actual message using this attachment. schema_checker = self.check_events_dict([ ('type', equals('attachment')), ('op', equals('update')), ('attachment', check_dict_only([ ('id', check_int), ('name', check_string), ('size', check_int), ('path_id', check_string), ('create_time', check_float), ('messages', check_list(check_dict_only([ ('id', check_int), ('name', check_float), ]))), ])), ('upload_space_used', equals(6)), ]) self.subscribe(self.example_user("hamlet"), "Denmark") body = "First message ...[zulip.txt](http://localhost:9991" + data['uri'] + ")" events = self.do_test( lambda: self.send_stream_message(self.example_email("hamlet"), "Denmark", body, "test"), num_events=2) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) # Now remove the attachment schema_checker = self.check_events_dict([ ('type', equals('attachment')), ('op', equals('remove')), ('attachment', check_dict_only([ ('id', check_int), ])), ('upload_space_used', equals(0)), ]) events = self.do_test( lambda: self.client_delete("/json/attachments/%s" % (entry.id,)), num_events=1, state_change_expected=False) error = schema_checker('events[0]', events[0]) self.assert_on_error(error) def test_notify_realm_export(self) -> None: # Create the `realm_exported` RealmAuditLog object with mock.patch('zerver.lib.export.do_export_realm', return_value=create_dummy_file('test-export.tar.gz')): self.client_post('/json/export/realm') schema_checker = self.check_events_dict([ ('type', equals('realm_export')), ('exports', check_list(check_dict_only([ ('id', check_int), ('event_time', check_string), ('acting_user_id', check_int), ('extra_data', check_dict_only([ ('export_path', check_string), ('deleted_timestamp', equals(None)) ])), ]))) ]) # Traditionally, we'd be testing the endpoint, but that # requires somewhat annoying mocking setup for what to do with # the export tarball. events = self.do_test( lambda: notify_realm_export(self.user_profile), state_change_expected=False, num_events=1) error = schema_checker('events[0]', events[0]) 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) -> None: user_profile = self.example_user('cordelia') self.assertFalse(user_profile.is_realm_admin) result = fetch_initial_state_data(user_profile, None, "", client_gravatar=False) self.assert_length(result['realm_bots'], 0) # additionally the API key for a random bot is not present in the data api_key = get_api_key(self.notification_bot()) self.assertNotIn(api_key, str(result)) # Admin users have access to all bots in the realm_bots field def test_realm_bots_admin(self) -> None: user_profile = self.example_user('hamlet') do_change_is_admin(user_profile, True) self.assertTrue(user_profile.is_realm_admin) result = fetch_initial_state_data(user_profile, None, "", client_gravatar=False) self.assertTrue(len(result['realm_bots']) > 2) def test_max_message_id_with_no_history(self) -> None: user_profile = self.example_user('aaron') # Delete all historical messages for this user UserMessage.objects.filter(user_profile=user_profile).delete() result = fetch_initial_state_data(user_profile, None, "", client_gravatar=False) self.assertEqual(result['max_message_id'], -1) class GetUnreadMsgsTest(ZulipTestCase): def mute_stream(self, user_profile: UserProfile, stream: Stream) -> None: recipient = Recipient.objects.get(type_id=stream.id, type=Recipient.STREAM) subscription = Subscription.objects.get( user_profile=user_profile, recipient=recipient ) subscription.is_muted = True subscription.save() def mute_topic(self, user_profile: UserProfile, stream_name: str, topic_name: str) -> None: realm = user_profile.realm stream = get_stream(stream_name, realm) recipient = get_stream_recipient(stream.id) add_topic_mute( user_profile=user_profile, stream_id=stream.id, recipient_id=recipient.id, topic_name=topic_name, ) def test_raw_unread_stream(self) -> None: cordelia = self.example_user('cordelia') hamlet = self.example_user('hamlet') realm = hamlet.realm for stream_name in ['social', 'devel', 'test here']: self.subscribe(hamlet, stream_name) self.subscribe(cordelia, stream_name) all_message_ids = set() # type: Set[int] message_ids = dict() tups = [ ('social', 'lunch'), ('test here', 'bla'), ('devel', 'python'), ('devel', 'ruby'), ] for stream_name, topic_name in tups: message_ids[topic_name] = [ self.send_stream_message( sender_email=cordelia.email, stream_name=stream_name, topic_name=topic_name, ) for i in range(3) ] all_message_ids |= set(message_ids[topic_name]) self.assertEqual(len(all_message_ids), 12) # sanity check on test setup self.mute_stream( user_profile=hamlet, stream=get_stream('test here', realm), ) self.mute_topic( user_profile=hamlet, stream_name='devel', topic_name='ruby', ) raw_unread_data = get_raw_unread_data( user_profile=hamlet, ) stream_dict = raw_unread_data['stream_dict'] self.assertEqual( set(stream_dict.keys()), all_message_ids, ) self.assertEqual( raw_unread_data['unmuted_stream_msgs'], set(message_ids['python']) | set(message_ids['lunch']), ) self.assertEqual( stream_dict[message_ids['lunch'][0]], dict( sender_id=cordelia.id, stream_id=get_stream('social', realm).id, topic='lunch', ) ) def test_raw_unread_huddle(self) -> None: cordelia = self.example_user('cordelia') othello = self.example_user('othello') hamlet = self.example_user('hamlet') prospero = self.example_user('prospero') huddle1_message_ids = [ self.send_huddle_message( cordelia.email, [hamlet.email, othello.email] ) for i in range(3) ] huddle2_message_ids = [ self.send_huddle_message( cordelia.email, [hamlet.email, prospero.email] ) for i in range(3) ] raw_unread_data = get_raw_unread_data( user_profile=hamlet, ) huddle_dict = raw_unread_data['huddle_dict'] self.assertEqual( set(huddle_dict.keys()), set(huddle1_message_ids) | set(huddle2_message_ids) ) huddle_string = ','.join( str(uid) for uid in sorted([cordelia.id, hamlet.id, othello.id]) ) self.assertEqual( huddle_dict[huddle1_message_ids[0]], dict(user_ids_string=huddle_string), ) def test_raw_unread_personal(self) -> None: cordelia = self.example_user('cordelia') othello = self.example_user('othello') hamlet = self.example_user('hamlet') cordelia_pm_message_ids = [ self.send_personal_message(cordelia.email, hamlet.email) for i in range(3) ] othello_pm_message_ids = [ self.send_personal_message(othello.email, hamlet.email) for i in range(3) ] raw_unread_data = get_raw_unread_data( user_profile=hamlet, ) pm_dict = raw_unread_data['pm_dict'] self.assertEqual( set(pm_dict.keys()), set(cordelia_pm_message_ids) | set(othello_pm_message_ids) ) self.assertEqual( pm_dict[cordelia_pm_message_ids[0]], dict(sender_id=cordelia.id), ) def test_unread_msgs(self) -> None: cordelia = self.example_user('cordelia') sender_id = cordelia.id sender_email = cordelia.email user_profile = self.example_user('hamlet') othello = self.example_user('othello') # our tests rely on order assert(sender_email < user_profile.email) assert(user_profile.email < othello.email) pm1_message_id = self.send_personal_message(sender_email, user_profile.email, "hello1") pm2_message_id = self.send_personal_message(sender_email, user_profile.email, "hello2") muted_stream = self.subscribe(user_profile, 'Muted Stream') self.mute_stream(user_profile, muted_stream) self.mute_topic(user_profile, 'Denmark', 'muted-topic') stream_message_id = self.send_stream_message(sender_email, "Denmark", "hello") muted_stream_message_id = self.send_stream_message(sender_email, "Muted Stream", "hello") muted_topic_message_id = self.send_stream_message( sender_email, "Denmark", topic_name="muted-topic", content="hello", ) huddle_message_id = self.send_huddle_message( sender_email, [user_profile.email, othello.email], 'hello3', ) def get_unread_data() -> UnreadMessagesResult: raw_unread_data = get_raw_unread_data(user_profile) aggregated_data = aggregate_unread_data(raw_unread_data) return aggregated_data result = get_unread_data() # The count here reflects the count of unread messages that we will # report to users in the bankruptcy dialog, and for now it excludes unread messages # from muted treams, but it doesn't exclude unread messages from muted topics yet. self.assertEqual(result['count'], 4) unread_pm = result['pms'][0] self.assertEqual(unread_pm['sender_id'], sender_id) self.assertEqual(unread_pm['unread_message_ids'], [pm1_message_id, pm2_message_id]) self.assertTrue('sender_ids' not in unread_pm) unread_stream = result['streams'][0] self.assertEqual(unread_stream['stream_id'], get_stream('Denmark', user_profile.realm).id) self.assertEqual(unread_stream['topic'], 'muted-topic') self.assertEqual(unread_stream['unread_message_ids'], [muted_topic_message_id]) self.assertEqual(unread_stream['sender_ids'], [sender_id]) unread_stream = result['streams'][1] self.assertEqual(unread_stream['stream_id'], get_stream('Denmark', user_profile.realm).id) self.assertEqual(unread_stream['topic'], 'test') self.assertEqual(unread_stream['unread_message_ids'], [stream_message_id]) self.assertEqual(unread_stream['sender_ids'], [sender_id]) unread_stream = result['streams'][2] self.assertEqual(unread_stream['stream_id'], get_stream('Muted Stream', user_profile.realm).id) self.assertEqual(unread_stream['topic'], 'test') self.assertEqual(unread_stream['unread_message_ids'], [muted_stream_message_id]) self.assertEqual(unread_stream['sender_ids'], [sender_id]) huddle_string = ','.join(str(uid) for uid in sorted([sender_id, user_profile.id, othello.id])) unread_huddle = result['huddles'][0] self.assertEqual(unread_huddle['user_ids_string'], huddle_string) self.assertEqual(unread_huddle['unread_message_ids'], [huddle_message_id]) self.assertTrue('sender_ids' not in unread_huddle) self.assertEqual(result['mentions'], []) um = UserMessage.objects.get( user_profile_id=user_profile.id, message_id=stream_message_id ) um.flags |= UserMessage.flags.mentioned um.save() result = get_unread_data() self.assertEqual(result['mentions'], [stream_message_id]) class ClientDescriptorsTest(ZulipTestCase): def test_get_client_info_for_all_public_streams(self) -> None: hamlet = self.example_user('hamlet') realm = hamlet.realm queue_data = dict( all_public_streams=True, apply_markdown=True, client_gravatar=True, client_type_name='website', event_types=['message'], last_connection_time=time.time(), queue_timeout=0, realm_id=realm.id, user_profile_id=hamlet.id, ) client = allocate_client_descriptor(queue_data) message_event = dict( realm_id=realm.id, stream_name='whatever', ) client_info = get_client_info_for_message_event( message_event, users=[], ) self.assertEqual(len(client_info), 1) dct = client_info[client.event_queue.id] self.assertEqual(dct['client'].apply_markdown, True) self.assertEqual(dct['client'].client_gravatar, True) self.assertEqual(dct['client'].user_profile_id, hamlet.id) self.assertEqual(dct['flags'], []) self.assertEqual(dct['is_sender'], False) message_event = dict( realm_id=realm.id, stream_name='whatever', sender_queue_id=client.event_queue.id, ) client_info = get_client_info_for_message_event( message_event, users=[], ) dct = client_info[client.event_queue.id] self.assertEqual(dct['is_sender'], True) def test_get_client_info_for_normal_users(self) -> None: hamlet = self.example_user('hamlet') cordelia = self.example_user('cordelia') realm = hamlet.realm def test_get_info(apply_markdown: bool, client_gravatar: bool) -> None: clear_client_event_queues_for_testing() queue_data = dict( all_public_streams=False, apply_markdown=apply_markdown, client_gravatar=client_gravatar, client_type_name='website', event_types=['message'], last_connection_time=time.time(), queue_timeout=0, realm_id=realm.id, user_profile_id=hamlet.id, ) client = allocate_client_descriptor(queue_data) message_event = dict( realm_id=realm.id, stream_name='whatever', ) client_info = get_client_info_for_message_event( message_event, users=[ dict(id=cordelia.id), ], ) self.assertEqual(len(client_info), 0) client_info = get_client_info_for_message_event( message_event, users=[ dict(id=cordelia.id), dict(id=hamlet.id, flags=['mentioned']), ], ) self.assertEqual(len(client_info), 1) dct = client_info[client.event_queue.id] self.assertEqual(dct['client'].apply_markdown, apply_markdown) self.assertEqual(dct['client'].client_gravatar, client_gravatar) self.assertEqual(dct['client'].user_profile_id, hamlet.id) self.assertEqual(dct['flags'], ['mentioned']) self.assertEqual(dct['is_sender'], False) test_get_info(apply_markdown=False, client_gravatar=False) test_get_info(apply_markdown=True, client_gravatar=False) test_get_info(apply_markdown=False, client_gravatar=True) test_get_info(apply_markdown=True, client_gravatar=True) def test_process_message_event_with_mocked_client_info(self) -> None: hamlet = self.example_user("hamlet") class MockClient: def __init__(self, user_profile_id: int, apply_markdown: bool, client_gravatar: bool) -> None: self.user_profile_id = user_profile_id self.apply_markdown = apply_markdown self.client_gravatar = client_gravatar self.client_type_name = 'whatever' self.events = [] # type: List[Dict[str, Any]] def accepts_messages(self) -> bool: return True def accepts_event(self, event: Dict[str, Any]) -> bool: assert(event['type'] == 'message') return True def add_event(self, event: Dict[str, Any]) -> None: self.events.append(event) client1 = MockClient( user_profile_id=hamlet.id, apply_markdown=True, client_gravatar=False, ) client2 = MockClient( user_profile_id=hamlet.id, apply_markdown=False, client_gravatar=False, ) client3 = MockClient( user_profile_id=hamlet.id, apply_markdown=True, client_gravatar=True, ) client4 = MockClient( user_profile_id=hamlet.id, apply_markdown=False, client_gravatar=True, ) client_info = { 'client:1': dict( client=client1, flags=['starred'], ), 'client:2': dict( client=client2, flags=['has_alert_word'], ), 'client:3': dict( client=client3, flags=[], ), 'client:4': dict( client=client4, flags=[], ), } sender = hamlet message_event = dict( message_dict=dict( id=999, content='**hello**', rendered_content='hello', sender_id=sender.id, type='stream', client='website', # NOTE: Some of these fields are clutter, but some # will be useful when we let clients specify # that they can compute their own gravatar URLs. sender_email=sender.email, sender_realm_id=sender.realm_id, sender_avatar_source=UserProfile.AVATAR_FROM_GRAVATAR, sender_avatar_version=1, sender_is_mirror_dummy=None, raw_display_recipient=None, recipient_type=None, recipient_type_id=None, ), ) # Setting users to `[]` bypasses code we don't care about # for this test--we assume client_info is correct in our mocks, # and we are interested in how messages are put on event queue. users = [] # type: List[Any] with mock.patch('zerver.tornado.event_queue.get_client_info_for_message_event', return_value=client_info): process_message_event(message_event, users) # We are not closely examining avatar_url at this point, so # just sanity check them and then delete the keys so that # upcoming comparisons work. for client in [client1, client2]: message = client.events[0]['message'] self.assertIn('gravatar.com', message['avatar_url']) message.pop('avatar_url') self.assertEqual(client1.events, [ dict( type='message', message=dict( type='stream', sender_id=sender.id, sender_email=sender.email, id=999, content='hello', content_type='text/html', client='website', ), flags=['starred'], ), ]) self.assertEqual(client2.events, [ dict( type='message', message=dict( type='stream', sender_id=sender.id, sender_email=sender.email, id=999, content='**hello**', content_type='text/x-markdown', client='website', ), flags=['has_alert_word'], ), ]) self.assertEqual(client3.events, [ dict( type='message', message=dict( type='stream', sender_id=sender.id, sender_email=sender.email, avatar_url=None, id=999, content='hello', content_type='text/html', client='website', ), flags=[], ), ]) self.assertEqual(client4.events, [ dict( type='message', message=dict( type='stream', sender_id=sender.id, sender_email=sender.email, avatar_url=None, id=999, content='**hello**', content_type='text/x-markdown', client='website', ), flags=[], ), ]) class FetchQueriesTest(ZulipTestCase): def test_queries(self) -> None: user = self.example_user("hamlet") self.login(user.email) flush_per_request_caches() with queries_captured() as queries: with mock.patch('zerver.lib.events.always_want') as want_mock: fetch_initial_state_data( user_profile=user, event_types=None, queue_id='x', client_gravatar=False, ) self.assert_length(queries, 33) expected_counts = dict( alert_words=0, custom_profile_fields=1, default_streams=1, default_stream_groups=1, hotspots=0, message=1, muted_topics=1, pointer=0, presence=3, realm=0, realm_bot=1, realm_domains=1, realm_embedded_bots=0, realm_emoji=1, realm_filters=1, realm_user=3, realm_user_groups=2, recent_private_conversations=2, starred_messages=1, stream=2, stop_words=0, subscription=6, update_display_settings=0, update_global_notifications=0, update_message_flags=5, user_status=1, zulip_version=0, ) wanted_event_types = { item[0][0] for item in want_mock.call_args_list } self.assertEqual(wanted_event_types, set(expected_counts)) for event_type in sorted(wanted_event_types): count = expected_counts[event_type] flush_per_request_caches() with queries_captured() as queries: if event_type == 'update_message_flags': event_types = ['update_message_flags', 'message'] else: event_types = [event_type] fetch_initial_state_data( user_profile=user, event_types=event_types, queue_id='x', client_gravatar=False, ) self.assert_length(queries, count) class TestEventsRegisterAllPublicStreamsDefaults(ZulipTestCase): def setUp(self) -> None: self.user_profile = self.example_user('hamlet') self.email = self.user_profile.email def test_use_passed_all_public_true_default_false(self) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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(ZulipTestCase): def setUp(self) -> None: self.user_profile = self.example_user('hamlet') self.email = self.user_profile.email self.stream = get_stream('Verona', self.user_profile.realm) def test_use_passed_narrow_no_default(self) -> 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) -> 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) -> 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) -> None: self.user_profile.default_events_register_stream_id = None self.user_profile.save() result = _default_narrow(self.user_profile, []) self.assertEqual(result, []) class TestGetRawUserDataSystemBotRealm(ZulipTestCase): def test_get_raw_user_data_on_system_bot_realm(self) -> None: result = get_raw_user_data(get_realm("zulipinternal"), True) for bot_email in settings.CROSS_REALM_BOT_EMAILS: bot_profile = get_system_bot(bot_email) self.assertTrue(bot_profile.id in result) self.assertTrue(result[bot_profile.id]['is_cross_realm_bot'])