paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-10-29 02:53:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

event_schema: Extract check_realm_bot_update.

Browse Source
This commit is contained in:
Steve Howell
2020-07-08 15:47:56 +00:00
committed by Tim Abbott
parent f5c4ee4477
commit 96f5ab1c87
2 changed files with 44 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
zerver/lib/event_schema.py
View File

@@ -178,6 +178,39 @@ def check_realm_bot_add(var_name: str, event: Dict[str, Any],) -> None:
raise AssertionError(f"Unknown bot_type: {bot_type}")
_check_bot_for_update = check_dict_only(
required_keys=[
# force vertical
("user_id", check_int),
],
optional_keys=[
("api_key", check_string),
("avatar_url", check_string),
("default_all_public_streams", check_bool),
("default_events_register_stream", check_none_or(check_string)),
("default_sending_stream", check_none_or(check_string)),
("full_name", check_string),
("owner_id", check_int),
("services", _check_bot_services),
],
)
_check_realm_bot_update = check_events_dict(
required_keys=[
("type", equals("realm_bot")),
("op", equals("update")),
("bot", _check_bot_for_update),
]
)
def check_realm_bot_update(var_name: str, event: Dict[str, Any], field: str,) -> None:
# Check the overall schema first.
_check_realm_bot_update(var_name, event)
assert {"user_id", field} == set(event["bot"].keys())
"""
realm/update events are flexible for values;
we will use a more strict checker to check

51
zerver/tests/test_events.py
View File

@@ -93,6 +93,7 @@ from zerver.lib.event_schema import (
basic_stream_fields,
check_events_dict,
check_realm_bot_add,
check_realm_bot_update,
check_realm_update,
check_stream_create,
check_stream_update,
@@ -297,16 +298,6 @@ class NormalActionsTest(BaseAction):
def create_bot(self, email: str, **extras: Any) -> UserProfile:
return self.create_test_bot(email, self.user_profile, **extras)
def realm_bot_schema(self, field_name: str, check: Validator[object]) -> Validator[Dict[str, object]]:
return check_events_dict([
('type', equals('realm_bot')),
('op', equals('update')),
('bot', check_dict_only([
('user_id', check_int),
(field_name, check),
])),
])
def test_mentioned_send_message_events(self) -> None:
user = self.example_user('hamlet')
@@ -1752,19 +1743,19 @@ class NormalActionsTest(BaseAction):
bot = self.create_bot('test')
action = lambda: do_change_full_name(bot, 'New Bot Name', self.user_profile)
events = self.verify_action(action, num_events=2)
self.realm_bot_schema('full_name', check_string)('events[1]', events[1])
check_realm_bot_update('events[1]', events[1], 'full_name')
def test_regenerate_bot_api_key(self) -> None:
bot = self.create_bot('test')
action = lambda: do_regenerate_api_key(bot, self.user_profile)
events = self.verify_action(action)
self.realm_bot_schema('api_key', check_string)('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'api_key')
def test_change_bot_avatar_source(self) -> None:
bot = self.create_bot('test')
action = lambda: do_change_avatar_fields(bot, bot.AVATAR_FROM_USER, acting_user=self.user_profile)
events = self.verify_action(action, num_events=2)
self.realm_bot_schema('avatar_url', check_string)('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'avatar_url')
self.assertEqual(events[1]['type'], 'realm_user')
def test_change_realm_icon_source(self) -> None:
@@ -1813,7 +1804,7 @@ class NormalActionsTest(BaseAction):
bot = self.create_bot('test')
action = lambda: do_change_default_all_public_streams(bot, True)
events = self.verify_action(action)
self.realm_bot_schema('default_all_public_streams', check_bool)('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'default_all_public_streams')
def test_change_bot_default_sending_stream(self) -> None:
bot = self.create_bot('test')
@@ -1821,11 +1812,11 @@ class NormalActionsTest(BaseAction):
action = lambda: do_change_default_sending_stream(bot, stream)
events = self.verify_action(action)
self.realm_bot_schema('default_sending_stream', check_string)('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'default_sending_stream')
action = lambda: do_change_default_sending_stream(bot, None)
events = self.verify_action(action)
self.realm_bot_schema('default_sending_stream', equals(None))('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'default_sending_stream')
def test_change_bot_default_events_register_stream(self) -> None:
bot = self.create_bot('test')
@@ -1833,11 +1824,11 @@ class NormalActionsTest(BaseAction):
action = lambda: do_change_default_events_register_stream(bot, stream)
events = self.verify_action(action)
self.realm_bot_schema('default_events_register_stream', check_string)('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'default_events_register_stream')
action = lambda: do_change_default_events_register_stream(bot, None)
events = self.verify_action(action)
self.realm_bot_schema('default_events_register_stream', equals(None))('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'default_events_register_stream')
def test_change_bot_owner(self) -> None:
change_bot_owner_checker_user = check_events_dict([
@@ -1849,20 +1840,12 @@ class NormalActionsTest(BaseAction):
])),
])
change_bot_owner_checker_bot = check_events_dict([
('type', equals('realm_bot')),
('op', equals('update')),
('bot', check_dict_only([
('user_id', check_int),
('owner_id', check_int),
])),
])
self.user_profile = self.example_user('iago')
owner = self.example_user('hamlet')
bot = self.create_bot('test')
action = lambda: do_change_bot_owner(bot, owner, self.user_profile)
events = self.verify_action(action, num_events=2)
change_bot_owner_checker_bot('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'owner_id')
change_bot_owner_checker_user('events[1]', events[1])
change_bot_owner_checker_bot = check_events_dict([
@@ -1889,18 +1872,6 @@ class NormalActionsTest(BaseAction):
change_bot_owner_checker_user('events[1]', events[1])
def test_do_update_outgoing_webhook_service(self) -> None:
update_outgoing_webhook_service_checker = check_events_dict([
('type', equals('realm_bot')),
('op', equals('update')),
('bot', check_dict_only([
('user_id', check_int),
('services', check_list(check_dict_only([
('base_url', check_url),
('interface', check_int),
('token', check_string),
]))),
])),
])
self.user_profile = self.example_user('iago')
bot = self.create_test_bot('test', self.user_profile,
full_name='Test Bot',
@@ -1910,7 +1881,7 @@ class NormalActionsTest(BaseAction):
)
action = lambda: do_update_outgoing_webhook_service(bot, 2, 'http://hostname.domain2.com')
events = self.verify_action(action)
update_outgoing_webhook_service_checker('events[0]', events[0])
check_realm_bot_update('events[0]', events[0], 'services')
def test_do_deactivate_user(self) -> None:
bot_deactivate_checker = check_events_dict([