diff --git a/zerver/lib/event_schema.py b/zerver/lib/event_schema.py index b877a82d03..d145d0d42e 100644 --- a/zerver/lib/event_schema.py +++ b/zerver/lib/event_schema.py @@ -16,6 +16,7 @@ from zerver.lib.validator import ( check_none_or, check_string, check_union, + check_url, equals, ) from zerver.models import Realm, Stream, UserProfile @@ -98,6 +99,84 @@ check_optional_value = check_union( ] ) +_check_bot_services_outgoing = check_dict_only( + required_keys=[ + # force vertical + ("base_url", check_url), + ("interface", check_int), + ("token", check_string), + ] +) + +# We use a strict check here, because our tests +# don't specifically focus on seeing how +# flexible we can make the types be for config_data. +_ad_hoc_config_data_schema = equals(dict(foo="bar")) + +_check_bot_services_embedded = check_dict_only( + required_keys=[ + # force vertical + ("service_name", check_string), + ("config_data", _ad_hoc_config_data_schema), + ] +) + +# Note that regular bots just get an empty list of services, +# so the sub_validator for check_list won't matter for them. +_check_bot_services = check_list( + check_union( + [ + # force vertical + _check_bot_services_outgoing, + _check_bot_services_embedded, + ] + ), +) + +_check_bot = check_dict_only( + required_keys=[ + ("user_id", check_int), + ("api_key", check_string), + ("avatar_url", check_string), + ("bot_type", check_int), + ("default_all_public_streams", check_bool), + ("default_events_register_stream", check_none_or(check_string)), + ("default_sending_stream", check_none_or(check_string)), + ("email", check_string), + ("full_name", check_string), + ("is_active", check_bool), + ("owner_id", check_int), + ("services", _check_bot_services), + ] +) + +_check_realm_bot_add = check_events_dict( + required_keys=[ + # force vertical + ("type", equals("realm_bot")), + ("op", equals("add")), + ("bot", _check_bot), + ] +) + + +def check_realm_bot_add(var_name: str, event: Dict[str, Any],) -> None: + _check_realm_bot_add(var_name, event) + + bot_type = event["bot"]["bot_type"] + + services_field = f"{var_name}['bot']['services']" + services = event["bot"]["services"] + + if bot_type == UserProfile.DEFAULT_BOT: + equals([])(services_field, services) + elif bot_type == UserProfile.OUTGOING_WEBHOOK_BOT: + check_list(_check_bot_services_outgoing, length=1)(services_field, services) + elif bot_type == UserProfile.EMBEDDED_BOT: + check_list(_check_bot_services_embedded, length=1)(services_field, services) + else: + raise AssertionError(f"Unknown bot_type: {bot_type}") + """ realm/update events are flexible for values; diff --git a/zerver/tests/test_events.py b/zerver/tests/test_events.py index e863e15e1b..0e9211035f 100644 --- a/zerver/tests/test_events.py +++ b/zerver/tests/test_events.py @@ -92,6 +92,7 @@ from zerver.lib.actions import ( from zerver.lib.event_schema import ( basic_stream_fields, check_events_dict, + check_realm_bot_add, check_realm_update, check_stream_create, check_stream_update, @@ -1725,48 +1726,9 @@ class NormalActionsTest(BaseAction): schema_checker('events[0]', events[0]) def test_create_bot(self) -> None: - # We use a strict check here, because this test - # isn't specifically focused on seeing how - # flexible we can make the types be for config_data. - ad_hoc_config_data_schema = equals(dict(foo='bar')) - - def get_bot_created_checker(bot_type: str) -> Validator[object]: - if bot_type == "GENERIC_BOT": - # Generic bots don't really understand the concept of - # "services", so we just enforce that we get an empty list. - check_services: Validator[List[object]] = equals([]) - elif bot_type == "OUTGOING_WEBHOOK_BOT": - check_services = check_list(check_dict_only([ - ('base_url', check_url), - ('interface', check_int), - ('token', check_string), - ]), length=1) - elif bot_type == "EMBEDDED_BOT": - check_services = check_list(check_dict_only([ - ('service_name', check_string), - ('config_data', ad_hoc_config_data_schema), - ]), length=1) - return check_events_dict([ - ('type', equals('realm_bot')), - ('op', equals('add')), - ('bot', check_dict_only([ - ('email', check_string), - ('user_id', check_int), - ('bot_type', 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_id', check_int), - ('services', check_services), - ])), - ]) action = lambda: self.create_bot('test') events = self.verify_action(action, num_events=2) - get_bot_created_checker(bot_type="GENERIC_BOT")('events[1]', events[1]) + check_realm_bot_add('events[1]', events[1]) action = lambda: self.create_bot('test_outgoing_webhook', full_name='Outgoing Webhook Bot', @@ -1776,7 +1738,7 @@ class NormalActionsTest(BaseAction): events = self.verify_action(action, num_events=2) # The third event is the second call of notify_created_bot, which contains additional # data for services (in contrast to the first call). - get_bot_created_checker(bot_type="OUTGOING_WEBHOOK_BOT")('events[1]', events[1]) + check_realm_bot_add('events[1]', events[1]) action = lambda: self.create_bot('test_embedded', full_name='Embedded Bot', @@ -1784,7 +1746,7 @@ class NormalActionsTest(BaseAction): config_data=ujson.dumps({'foo': 'bar'}), bot_type=UserProfile.EMBEDDED_BOT) events = self.verify_action(action, num_events=2) - get_bot_created_checker(bot_type="EMBEDDED_BOT")('events[1]', events[1]) + check_realm_bot_add('events[1]', events[1]) def test_change_bot_full_name(self) -> None: bot = self.create_bot('test') @@ -1918,30 +1880,12 @@ class NormalActionsTest(BaseAction): change_bot_owner_checker_bot('events[0]', events[0]) change_bot_owner_checker_user('events[1]', events[1]) - change_bot_owner_checker_bot = check_events_dict([ - ('type', equals('realm_bot')), - ('op', equals('add')), - ('bot', check_dict_only([ - ('email', check_string), - ('user_id', check_int), - ('bot_type', 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_id', check_int), - ('services', equals([])), - ])), - ]) previous_owner = self.example_user('aaron') self.user_profile = self.example_user('hamlet') bot = self.create_test_bot('test2', previous_owner, full_name='Test2 Testerson') action = lambda: do_change_bot_owner(bot, self.user_profile, previous_owner) events = self.verify_action(action, num_events=2) - change_bot_owner_checker_bot('events[0]', events[0]) + check_realm_bot_add('events[0]', events[0]) change_bot_owner_checker_user('events[1]', events[1]) def test_do_update_outgoing_webhook_service(self) -> None: @@ -1983,32 +1927,11 @@ class NormalActionsTest(BaseAction): bot_deactivate_checker('events[1]', events[1]) def test_do_reactivate_user(self) -> None: - bot_reactivate_checker = check_events_dict([ - ('type', equals('realm_bot')), - ('op', equals('add')), - ('bot', check_dict_only([ - ('email', check_string), - ('user_id', check_int), - ('bot_type', 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_id', check_none_or(check_int)), - ('services', check_list(check_dict_only([ - ('base_url', check_url), - ('interface', check_int), - ]))), - ])), - ]) bot = self.create_bot('test') do_deactivate_user(bot) action = lambda: do_reactivate_user(bot) events = self.verify_action(action, num_events=2) - bot_reactivate_checker('events[1]', events[1]) + check_realm_bot_add('events[1]', events[1]) def test_do_mark_hotspot_as_read(self) -> None: self.user_profile.tutorial_status = UserProfile.TUTORIAL_WAITING