event_schema: Extract check_realm_update_dict.

tools/check-node-fixtures

@@ -73,6 +73,8 @@ EXEMPT_OPENAPI_NAMES = [
     "realm_filters_event",
     # bots, delivery_email, profile_data
     "realm_user_add_event",
+    # openapi is incomplete
+    "realm_update_dict_event",
 ]
 
 

zerver/lib/event_schema.py

@@ -713,6 +713,108 @@ def check_realm_update(var_name: str, event: Dict[str, object], prop: str,) -> N
         raise AssertionError(f"Unexpected property type {property_type}")
 
 
+authentication_dict = DictType(
+    required_keys=[
+        ("Google", bool),
+        ("Dev", bool),
+        ("LDAP", bool),
+        ("GitHub", bool),
+        ("Email", bool),
+    ]
+)
+
+authentication_data = DictType(
+    required_keys=[
+        # this single-key dictionary is an annoying
+        # consequence of us using property_type of
+        # "default" for legacy reasons
+        ("authentication_methods", authentication_dict),
+    ]
+)
+
+icon_data = DictType(
+    required_keys=[
+        # force vertical
+        ("icon_url", str),
+        ("icon_source", str),
+    ]
+)
+
+logo_data = DictType(
+    required_keys=[
+        # force vertical
+        ("logo_url", str),
+        ("logo_source", str),
+    ]
+)
+
+message_edit_data = DictType(
+    required_keys=[
+        ("allow_message_editing", bool),
+        ("message_content_edit_limit_seconds", int),
+        ("allow_community_topic_editing", bool),
+    ]
+)
+
+night_logo_data = DictType(
+    required_keys=[
+        # force vertical
+        ("night_logo_url", str),
+        ("night_logo_source", str),
+    ]
+)
+
+update_dict_data = UnionType(
+    [
+        # force vertical
+        authentication_data,
+        icon_data,
+        logo_data,
+        message_edit_data,
+        night_logo_data,
+    ]
+)
+
+realm_update_dict_event = event_dict_type(
+    required_keys=[
+        ("type", Equals("realm")),
+        ("op", Equals("update_dict")),
+        ("property", EnumType(["default", "icon", "logo", "night_logo"])),
+        ("data", update_dict_data),
+    ]
+)
+_check_realm_update_dict = make_checker(realm_update_dict_event)
+
+
+def check_realm_update_dict(
+    # handle union types
+    var_name: str,
+    event: Dict[str, object],
+) -> None:
+    _check_realm_update_dict(var_name, event)
+
+    if event["property"] == "default":
+        assert isinstance(event["data"], dict)
+
+        if "allow_message_editing" in event["data"]:
+            sub_type = message_edit_data
+        elif "authentication_methods" in event["data"]:
+            sub_type = authentication_data
+        else:
+            raise AssertionError("unhandled fields in data")
+
+    elif event["property"] == "icon":
+        sub_type = icon_data
+    elif event["property"] == "logo":
+        sub_type = logo_data
+    elif event["property"] == "night_logo":
+        sub_type = night_logo_data
+    else:
+        raise AssertionError("unhandled property: {event['property']}")
+
+    check_data(sub_type, f"{var_name}['data']", event["data"])
+
+
 realm_user_type = DictType(
     required_keys=[
         ("user_id", int),

zerver/tests/test_events.py

@@ -112,6 +112,7 @@ from zerver.lib.event_schema import (
     check_realm_export,
     check_realm_filters,
     check_realm_update,
+    check_realm_update_dict,
     check_realm_user_add,
     check_realm_user_update,
     check_stream_create,
@@ -991,21 +992,6 @@ class NormalActionsTest(BaseAction):
         assert isinstance(events[1]['person']['avatar_url_medium'], str)
 
     def test_change_realm_authentication_methods(self) -> None:
-        schema_checker = check_events_dict([
-            ('type', equals('realm')),
-            ('op', equals('update_dict')),
-            ('property', equals('default')),
-            ('data', check_dict_only([
-                ('authentication_methods', check_dict_only([
-                    ('Google', check_bool),
-                    ('Dev', check_bool),
-                    ('LDAP', check_bool),
-                    ('GitHub', check_bool),
-                    ('Email', check_bool),
-                ])),
-            ])),
-        ])
-
         def fake_backends() -> Any:
             backends = (
                 'zproject.backends.DevAuthBackend',
@@ -1031,7 +1017,7 @@ class NormalActionsTest(BaseAction):
                         self.user_profile.realm,
                         auth_method_dict))
 
-            schema_checker('events[0]', events[0])
+            check_realm_update_dict('events[0]', events[0])
 
     def test_change_pin_stream(self) -> None:
         schema_checker = check_events_dict([
@@ -1093,16 +1079,6 @@ class NormalActionsTest(BaseAction):
                 schema_checker('events[0]', events[0])
 
     def test_change_realm_message_edit_settings(self) -> None:
-        schema_checker = 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),
@@ -1112,7 +1088,7 @@ class NormalActionsTest(BaseAction):
                                                      allow_message_editing,
                                                      message_content_edit_limit_seconds,
                                                      False))
-            schema_checker('events[0]', events[0])
+            check_realm_update_dict('events[0]', events[0])
 
     def test_change_realm_notifications_stream(self) -> None:
 
@@ -1387,44 +1363,17 @@ class NormalActionsTest(BaseAction):
     def test_change_realm_icon_source(self) -> None:
         action = lambda: do_change_icon_source(self.user_profile.realm, Realm.ICON_UPLOADED)
         events = self.verify_action(action, state_change_expected=True)
-        schema_checker = check_events_dict([
+        check_realm_update_dict('events[0]', events[0])
-            ('type', equals('realm')),
-            ('op', equals('update_dict')),
-            ('property', equals('icon')),
-            ('data', check_dict_only([
-                ('icon_url', check_string),
-                ('icon_source', check_string),
-            ])),
-        ])
-        schema_checker('events[0]', events[0])
 
     def test_change_realm_day_mode_logo_source(self) -> None:
         action = lambda: do_change_logo_source(self.user_profile.realm, Realm.LOGO_UPLOADED, False, acting_user=self.user_profile)
         events = self.verify_action(action, state_change_expected=True)
-        schema_checker = check_events_dict([
+        check_realm_update_dict('events[0]', events[0])
-            ('type', equals('realm')),
-            ('op', equals('update_dict')),
-            ('property', equals('logo')),
-            ('data', check_dict_only([
-                ('logo_url', check_string),
-                ('logo_source', check_string),
-            ])),
-        ])
-        schema_checker('events[0]', events[0])
 
     def test_change_realm_night_mode_logo_source(self) -> None:
         action = lambda: do_change_logo_source(self.user_profile.realm, Realm.LOGO_UPLOADED, True, acting_user=self.user_profile)
         events = self.verify_action(action, state_change_expected=True)
-        schema_checker = check_events_dict([
+        check_realm_update_dict('events[0]', events[0])
-            ('type', equals('realm')),
-            ('op', equals('update_dict')),
-            ('property', equals('night_logo')),
-            ('data', check_dict_only([
-                ('night_logo_url', check_string),
-                ('night_logo_source', check_string),
-            ])),
-        ])
-        schema_checker('events[0]', events[0])
 
     def test_change_bot_default_all_public_streams(self) -> None:
         bot = self.create_bot('test')