event_schema: Extract check_realm_export.

These are all trivial transformations.

Note that we don't insist timestamps are
floats; the NumberType class allows ints
too.

zerver/lib/event_schema.py

@@ -416,6 +416,58 @@ def check_realm_bot_update(
     assert {"user_id", field} == set(event["bot"].keys())
 
 
+export_type = DictType(
+    required_keys=[
+        ("id", int),
+        ("export_time", NumberType()),
+        ("acting_user_id", int),
+        ("export_url", OptionalType(str)),
+        ("deleted_timestamp", OptionalType(NumberType())),
+        ("failed_timestamp", OptionalType(NumberType())),
+        ("pending", bool),
+    ]
+)
+
+realm_export_event = event_dict_type(
+    required_keys=[
+        # force vertical
+        ("type", Equals("realm_export")),
+        ("exports", ListType(export_type),),
+    ]
+)
+_check_realm_export = make_checker(realm_export_event)
+
+
+def check_realm_export(
+    var_name: str,
+    event: Dict[str, object],
+    has_export_url: bool,
+    has_deleted_timestamp: bool,
+    has_failed_timestamp: bool,
+) -> None:
+    """
+    Check the overall event first, knowing it has some
+    optional types.
+    """
+    _check_realm_export(var_name, event)
+
+    """
+    Theoretically we can have multiple exports, but until
+    that happens in practice, we assume our tests are
+    gonna have one export.
+    """
+    assert isinstance(event["exports"], list)
+    assert len(event["exports"]) == 1
+    export = event["exports"][0]
+
+    """
+    Now verify which fields have non-None values.
+    """
+    assert has_export_url == (export["export_url"] is not None)
+    assert has_deleted_timestamp == (export["deleted_timestamp"] is not None)
+    assert has_failed_timestamp == (export["failed_timestamp"] is not None)
+
+
 plan_type_extra_data_type = DictType(
     required_keys=[
         # force vertical

zerver/tests/test_events.py

@@ -103,6 +103,7 @@ from zerver.lib.event_schema import (
     check_realm_bot_delete,
     check_realm_bot_remove,
     check_realm_bot_update,
+    check_realm_export,
     check_realm_update,
     check_realm_user_update,
     check_stream_create,
@@ -1843,32 +1844,6 @@ class NormalActionsTest(BaseAction):
         schema_checker('events[0]', events[0])
 
     def test_notify_realm_export(self) -> None:
-        pending_schema_checker = check_events_dict([
-            ('type', equals('realm_export')),
-            ('exports', check_list(check_dict_only([
-                ('id', check_int),
-                ('export_time', check_float),
-                ('acting_user_id', check_int),
-                ('export_url', equals(None)),
-                ('deleted_timestamp', equals(None)),
-                ('failed_timestamp', equals(None)),
-                ('pending', check_bool),
-            ]))),
-        ])
-
-        schema_checker = check_events_dict([
-            ('type', equals('realm_export')),
-            ('exports', check_list(check_dict_only([
-                ('id', check_int),
-                ('export_time', check_float),
-                ('acting_user_id', check_int),
-                ('export_url', check_string),
-                ('deleted_timestamp', equals(None)),
-                ('failed_timestamp', equals(None)),
-                ('pending', check_bool),
-            ]))),
-        ])
-
         do_change_user_role(self.user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
         self.login_user(self.user_profile)
 
@@ -1882,60 +1857,40 @@ class NormalActionsTest(BaseAction):
                 'INFO:root:Completed data export for zulip in' in info_logs.output[0]
             )
 
-        # We first notify when an export is initiated,
+        # We get two realm_export events for this action, where the first
-        pending_schema_checker('events[0]', events[0])
+        # is missing the export_url (because it's pending).
+        check_realm_export(
+            "events[0]",
+            events[0],
+            has_export_url=False,
+            has_deleted_timestamp=False,
+            has_failed_timestamp=False,
+        )
 
-        # The second event is then a message from notification-bot.
+        check_realm_export(
-        schema_checker('events[2]', events[2])
+            "events[2]",
+            events[2],
+            has_export_url=True,
+            has_deleted_timestamp=False,
+            has_failed_timestamp=False,
+        )
 
         # Now we check the deletion of the export.
-        deletion_schema_checker = check_events_dict([
-            ('type', equals('realm_export')),
-            ('exports', check_list(check_dict_only([
-                ('id', check_int),
-                ('export_time', check_float),
-                ('acting_user_id', check_int),
-                ('export_url', equals(None)),
-                ('deleted_timestamp', check_float),
-                ('failed_timestamp', equals(None)),
-                ('pending', check_bool),
-            ]))),
-        ])
-
         audit_log_entry = RealmAuditLog.objects.filter(
             event_type=RealmAuditLog.REALM_EXPORTED).first()
         events = self.verify_action(
             lambda: self.client_delete(f'/json/export/realm/{audit_log_entry.id}'),
             state_change_expected=False, num_events=1)
-        deletion_schema_checker('events[0]', events[0])
+
+        check_realm_export(
+            "events[0]",
+            events[0],
+            has_export_url=False,
+            has_deleted_timestamp=True,
+            has_failed_timestamp=False,
+        )
 
     def test_notify_realm_export_on_failure(self) -> None:
-        pending_schema_checker = check_events_dict([
-            ('type', equals('realm_export')),
-            ('exports', check_list(check_dict_only([
-                ('id', check_int),
-                ('export_time', check_float),
-                ('acting_user_id', check_int),
-                ('export_url', equals(None)),
-                ('deleted_timestamp', equals(None)),
-                ('failed_timestamp', equals(None)),
-                ('pending', check_bool),
-            ]))),
-        ])
-
-        failed_schema_checker = check_events_dict([
-            ('type', equals('realm_export')),
-            ('exports', check_list(check_dict_only([
-                ('id', check_int),
-                ('export_time', check_float),
-                ('acting_user_id', check_int),
-                ('export_url', equals(None)),
-                ('deleted_timestamp', equals(None)),
-                ('failed_timestamp', check_float),
-                ('pending', check_bool),
-            ]))),
-        ])
-
         do_change_user_role(self.user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
         self.login_user(self.user_profile)
 
@@ -1952,9 +1907,21 @@ class NormalActionsTest(BaseAction):
             # independent of time bit by not matching exact log but only part of it.
             self.assertTrue("ERROR:root:Data export for zulip failed after" in error_log.output[0])
 
-        pending_schema_checker('events[0]', events[0])
+        # We get two events for the export.
-
+        check_realm_export(
-        failed_schema_checker('events[1]', events[1])
+            "events[0]",
+            events[0],
+            has_export_url=False,
+            has_deleted_timestamp=False,
+            has_failed_timestamp=False,
+        )
+        check_realm_export(
+            "events[1]",
+            events[1],
+            has_export_url=False,
+            has_deleted_timestamp=False,
+            has_failed_timestamp=True,
+        )
 
     def test_has_zoom_token(self) -> None:
         schema_checker = check_events_dict([