# See https://zulip.readthedocs.io/en/latest/subsystems/events-system.html for
# high-level documentation on how this system works.
#
# This module is closely integrated with zerver/lib/event_schema.py
# and zerver/lib/data_types.py systems for validating the schemas of
# events; it also uses the OpenAPI tools to validate our documentation.
import base64
import copy
import time
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from datetime import timedelta
from enum import Enum
from io import StringIO
from typing import Any
from unittest import mock

import orjson
import time_machine
from dateutil.parser import parse as dateparser
from django.utils.timezone import now as timezone_now
from typing_extensions import override

from zerver.actions.alert_words import do_add_alert_words, do_remove_alert_words
from zerver.actions.bots import (
    do_change_bot_owner,
    do_change_default_all_public_streams,
    do_change_default_events_register_stream,
    do_change_default_sending_stream,
)
from zerver.actions.channel_folders import (
    check_add_channel_folder,
    do_archive_channel_folder,
    do_change_channel_folder_description,
    do_change_channel_folder_name,
    do_unarchive_channel_folder,
    try_reorder_realm_channel_folders,
)
from zerver.actions.create_user import do_create_user, do_reactivate_user
from zerver.actions.custom_profile_fields import (
    check_remove_custom_profile_field_value,
    do_remove_realm_custom_profile_field,
    do_update_user_custom_profile_data_if_changed,
    try_add_realm_custom_profile_field,
    try_update_realm_custom_profile_field,
)
from zerver.actions.default_streams import (
    do_add_default_stream,
    do_add_streams_to_default_stream_group,
    do_change_default_stream_group_description,
    do_change_default_stream_group_name,
    do_create_default_stream_group,
    do_remove_default_stream,
    do_remove_default_stream_group,
    do_remove_streams_from_default_stream_group,
    lookup_default_stream_groups,
)
from zerver.actions.invites import (
    do_create_multiuse_invite_link,
    do_invite_users,
    do_revoke_multi_use_invite,
    do_revoke_user_invite,
)
from zerver.actions.message_delete import do_delete_messages
from zerver.actions.message_edit import (
    build_message_edit_request,
    do_update_embedded_data,
    do_update_message,
)
from zerver.actions.message_flags import do_update_message_flags
from zerver.actions.muted_users import do_mute_user, do_unmute_user
from zerver.actions.navigation_views import (
    do_add_navigation_view,
    do_remove_navigation_view,
    do_update_navigation_view,
)
from zerver.actions.onboarding_steps import do_mark_onboarding_step_as_read
from zerver.actions.presence import do_update_user_presence
from zerver.actions.reactions import do_add_reaction, do_remove_reaction
from zerver.actions.realm_domains import (
    do_add_realm_domain,
    do_change_realm_domain,
    do_remove_realm_domain,
)
from zerver.actions.realm_emoji import check_add_realm_emoji, do_remove_realm_emoji
from zerver.actions.realm_icon import do_change_icon_source
from zerver.actions.realm_linkifiers import (
    check_reorder_linkifiers,
    do_add_linkifier,
    do_remove_linkifier,
    do_update_linkifier,
)
from zerver.actions.realm_logo import do_change_logo_source
from zerver.actions.realm_playgrounds import check_add_realm_playground, do_remove_realm_playground
from zerver.actions.realm_settings import (
    do_change_realm_org_type,
    do_change_realm_permission_group_setting,
    do_change_realm_plan_type,
    do_deactivate_realm,
    do_set_push_notifications_enabled_end_timestamp,
    do_set_realm_authentication_methods,
    do_set_realm_moderation_request_channel,
    do_set_realm_new_stream_announcements_stream,
    do_set_realm_property,
    do_set_realm_signup_announcements_stream,
    do_set_realm_user_default_setting,
    do_set_realm_zulip_update_announcements_stream,
)
from zerver.actions.saved_snippets import (
    do_create_saved_snippet,
    do_delete_saved_snippet,
    do_edit_saved_snippet,
)
from zerver.actions.scheduled_messages import (
    check_schedule_message,
    delete_scheduled_message,
    edit_scheduled_message,
)
from zerver.actions.streams import (
    bulk_add_subscriptions,
    bulk_remove_subscriptions,
    do_change_stream_description,
    do_change_stream_folder,
    do_change_stream_group_based_setting,
    do_change_stream_message_retention_days,
    do_change_stream_permission,
    do_change_subscription_property,
    do_deactivate_stream,
    do_rename_stream,
    do_set_stream_property,
    do_unarchive_stream,
)
from zerver.actions.submessage import do_add_submessage
from zerver.actions.typing import (
    check_send_typing_notification,
    do_send_direct_message_edit_typing_notification,
    do_send_stream_message_edit_typing_notification,
    do_send_stream_typing_notification,
)
from zerver.actions.user_groups import (
    add_subgroups_to_user_group,
    bulk_add_members_to_user_groups,
    bulk_remove_members_from_user_groups,
    check_add_user_group,
    do_change_user_group_permission_setting,
    do_deactivate_user_group,
    do_reactivate_user_group,
    do_update_user_group_description,
    do_update_user_group_name,
    remove_subgroups_from_user_group,
)
from zerver.actions.user_settings import (
    do_change_avatar_fields,
    do_change_full_name,
    do_change_user_delivery_email,
    do_change_user_setting,
    do_regenerate_api_key,
)
from zerver.actions.user_status import do_update_user_status
from zerver.actions.user_topics import do_set_user_topic_visibility_policy
from zerver.actions.users import (
    do_change_is_imported_stub,
    do_deactivate_user,
    do_update_outgoing_webhook_service,
)
from zerver.actions.video_calls import do_set_zoom_token
from zerver.lib.drafts import DraftData, do_create_drafts, do_delete_draft, do_edit_draft
from zerver.lib.event_schema import (
    check_alert_words,
    check_attachment_add,
    check_attachment_remove,
    check_attachment_update,
    check_channel_folder_add,
    check_channel_folder_reorder,
    check_channel_folder_update,
    check_custom_profile_fields,
    check_default_stream_groups,
    check_default_streams,
    check_delete_message,
    check_direct_message,
    check_draft_add,
    check_draft_remove,
    check_draft_update,
    check_has_zoom_token,
    check_heartbeat,
    check_invites_changed,
    check_legacy_presence,
    check_message,
    check_modern_presence,
    check_muted_topics,
    check_muted_users,
    check_navigation_view_add,
    check_navigation_view_remove,
    check_navigation_view_update,
    check_onboarding_steps,
    check_push_device,
    check_reaction_add,
    check_reaction_remove,
    check_realm_bot_add,
    check_realm_bot_delete,
    check_realm_bot_update,
    check_realm_deactivated,
    check_realm_default_update,
    check_realm_domains_add,
    check_realm_domains_change,
    check_realm_domains_remove,
    check_realm_emoji_update,
    check_realm_export,
    check_realm_export_consent,
    check_realm_linkifiers,
    check_realm_playgrounds,
    check_realm_update,
    check_realm_update_dict,
    check_realm_user_add,
    check_realm_user_remove,
    check_realm_user_update,
    check_saved_snippets_add,
    check_saved_snippets_remove,
    check_saved_snippets_update,
    check_scheduled_message_add,
    check_scheduled_message_remove,
    check_scheduled_message_update,
    check_stream_create,
    check_stream_delete,
    check_stream_update,
    check_submessage,
    check_subscription_add,
    check_subscription_peer_add,
    check_subscription_peer_remove,
    check_subscription_remove,
    check_subscription_update,
    check_typing_edit_message_start,
    check_typing_edit_message_stop,
    check_typing_start,
    check_typing_stop,
    check_update_message,
    check_update_message_flags_add,
    check_update_message_flags_remove,
    check_user_group_add,
    check_user_group_add_members,
    check_user_group_add_subgroups,
    check_user_group_remove,
    check_user_group_remove_members,
    check_user_group_remove_subgroups,
    check_user_group_update,
    check_user_settings_update,
    check_user_status,
    check_user_topic,
)
from zerver.lib.events import apply_events, fetch_initial_state_data, post_process_state
from zerver.lib.exceptions import InvalidBouncerPublicKeyError
from zerver.lib.markdown import render_message_markdown
from zerver.lib.mention import MentionBackend, MentionData
from zerver.lib.muted_users import get_mute_object
from zerver.lib.push_registration import (
    RegisterPushDeviceToBouncerQueueItem,
    handle_register_push_device_to_bouncer,
)
from zerver.lib.streams import check_update_all_streams_active_status, user_has_metadata_access
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    create_dummy_file,
    get_subscription,
    get_test_image_file,
    read_test_image_file,
    reset_email_visibility_to_everyone_in_zulip_realm,
    stdout_suppressed,
)
from zerver.lib.timestamp import convert_to_UTC, datetime_to_timestamp
from zerver.lib.topic import TOPIC_NAME
from zerver.lib.types import (
    ProfileDataElementUpdateDict,
    UserGroupMembersData,
    UserGroupMembersDict,
)
from zerver.lib.upload import upload_message_attachment
from zerver.lib.user_groups import (
    UserGroupMembershipDetails,
    get_group_setting_value_for_api,
    get_role_based_system_groups_dict,
)
from zerver.models import (
    Attachment,
    CustomProfileField,
    ImageAttachment,
    Message,
    MultiuseInvite,
    NamedUserGroup,
    PreregistrationUser,
    PushDevice,
    Realm,
    RealmAuditLog,
    RealmDomain,
    RealmExport,
    RealmFilter,
    RealmPlayground,
    RealmUserDefault,
    SavedSnippet,
    Service,
    Stream,
    UserMessage,
    UserPresence,
    UserProfile,
    UserStatus,
    UserTopic,
)
from zerver.models.bots import get_bot_services
from zerver.models.clients import get_client
from zerver.models.groups import SystemGroups
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.recipients import get_or_create_direct_message_group
from zerver.models.streams import StreamTopicsPolicyEnum, get_stream
from zerver.models.users import get_user_by_delivery_email
from zerver.openapi.openapi import validate_against_openapi_schema
from zerver.tornado.django_api import send_event_rollback_unsafe
from zerver.tornado.event_queue import (
    allocate_client_descriptor,
    clear_client_event_queues_for_testing,
    create_heartbeat_event,
    mark_clients_to_reload,
    send_restart_events,
    send_web_reload_client_events,
)
from zerver.views.realm_playgrounds import access_playground_by_id
from zerver.worker.thumbnail import ensure_thumbnails


class BaseAction(ZulipTestCase):
    """Core class for verifying the apply_event race handling logic as
    well as the event formatting logic of any function using send_event_rollback_unsafe.

    See https://zulip.readthedocs.io/en/latest/subsystems/events-system.html#testing
    for extensive design details for this testing system.
    """

    @override
    def setUp(self) -> None:
        super().setUp()
        self.user_profile = self.example_user("hamlet")

    @contextmanager
    def verify_action(
        self,
        *,
        event_types: list[str] | None = None,
        include_subscribers: bool = True,
        state_change_expected: bool = True,
        notification_settings_null: bool = False,
        client_gravatar: bool = True,
        user_avatar_url_field_optional: bool = False,
        slim_presence: bool = False,
        include_streams: bool = True,
        num_events: int = 1,
        bulk_message_deletion: bool = True,
        stream_typing_notifications: bool = True,
        user_settings_object: bool = False,
        pronouns_field_type_supported: bool = True,
        linkifier_url_template: bool = True,
        user_list_incomplete: bool = False,
        client_is_old: bool = False,
        include_deactivated_groups: bool = False,
        archived_channels: bool = False,
        allow_empty_topic_name: bool = True,
        simplified_presence_events: bool = False,
    ) -> Iterator[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,
                realm_id=self.user_profile.realm_id,
                event_types=event_types,
                client_type_name="website",
                apply_markdown=True,
                client_gravatar=client_gravatar,
                slim_presence=slim_presence,
                all_public_streams=False,
                queue_timeout=600,
                last_connection_time=time.time(),
                narrow=[],
                bulk_message_deletion=bulk_message_deletion,
                stream_typing_notifications=stream_typing_notifications,
                user_settings_object=user_settings_object,
                pronouns_field_type_supported=pronouns_field_type_supported,
                linkifier_url_template=linkifier_url_template,
                user_list_incomplete=user_list_incomplete,
                include_deactivated_groups=include_deactivated_groups,
                archived_channels=archived_channels,
                simplified_presence_events=simplified_presence_events,
            )
        )

        # 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,
            realm=self.user_profile.realm,
            event_types=event_types,
            client_gravatar=client_gravatar,
            user_avatar_url_field_optional=user_avatar_url_field_optional,
            slim_presence=slim_presence,
            include_subscribers=include_subscribers,
            include_streams=include_streams,
            pronouns_field_type_supported=pronouns_field_type_supported,
            linkifier_url_template=linkifier_url_template,
            user_list_incomplete=user_list_incomplete,
            include_deactivated_groups=include_deactivated_groups,
            archived_channels=archived_channels,
        )

        if client_is_old:
            mark_clients_to_reload([client.event_queue.id])

        events: list[dict[str, Any]] = []

        # We want even those `send_event_rollback_unsafe` calls which have
        # been hooked to `transaction.on_commit` to execute in tests.
        # See the comment in `ZulipTestCase.capture_send_event_calls`.
        with self.captureOnCommitCallbacks(execute=True):
            yield events

        self.user_profile.refresh_from_db()

        # Append to an empty list so the result is accessible through the
        # reference we just yielded.
        events += client.event_queue.contents()

        content = {
            "queue_id": "123.12",
            # The JSON wrapper helps in converting tuples to lists
            # as tuples aren't valid JSON structure.
            "events": orjson.loads(orjson.dumps(events)),
            "msg": "",
            "result": "success",
        }
        validate_against_openapi_schema(content, "/events", "get", "200")
        self.assert_length(events, num_events)
        initial_state = copy.deepcopy(hybrid_state)
        post_process_state(
            self.user_profile, initial_state, notification_settings_null, allow_empty_topic_name
        )
        before = orjson.dumps(initial_state)
        apply_events(
            self.user_profile,
            state=hybrid_state,
            events=events,
            fetch_event_types=None,
            client_gravatar=client_gravatar,
            slim_presence=slim_presence,
            include_subscribers=include_subscribers,
            linkifier_url_template=linkifier_url_template,
            user_list_incomplete=user_list_incomplete,
            include_deactivated_groups=include_deactivated_groups,
            archived_channels=archived_channels,
            simplified_presence_events=simplified_presence_events,
        )
        post_process_state(
            self.user_profile, hybrid_state, notification_settings_null, allow_empty_topic_name
        )
        after = orjson.dumps(hybrid_state)

        if state_change_expected:
            if before == after:  # nocoverage
                print(orjson.dumps(initial_state, option=orjson.OPT_INDENT_2).decode())
                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,
            realm=self.user_profile.realm,
            event_types=event_types,
            client_gravatar=client_gravatar,
            user_avatar_url_field_optional=user_avatar_url_field_optional,
            slim_presence=slim_presence,
            include_subscribers=include_subscribers,
            include_streams=include_streams,
            pronouns_field_type_supported=pronouns_field_type_supported,
            linkifier_url_template=linkifier_url_template,
            user_list_incomplete=user_list_incomplete,
            include_deactivated_groups=include_deactivated_groups,
            archived_channels=archived_channels,
            simplified_presence_events=simplified_presence_events,
        )
        post_process_state(
            self.user_profile, normal_state, notification_settings_null, allow_empty_topic_name
        )
        self.match_states(hybrid_state, normal_state, events)

    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:
            if "never_subscribed" in state:
                for u in state["never_subscribed"]:
                    if "subscribers" in u:
                        u["subscribers"].sort()
                    # this isn't guaranteed to match
                    del u["subscriber_count"]
            if "subscriptions" in state:
                for u in state["subscriptions"]:
                    if "subscribers" in u:
                        u["subscribers"].sort()
                    # this isn't guaranteed to match
                    del u["subscriber_count"]
                state["subscriptions"] = {u["name"]: u for u in state["subscriptions"]}
            if "unsubscribed" in state:
                for u in state["unsubscribed"]:
                    # this isn't guaranteed to match
                    del u["subscriber_count"]
                state["unsubscribed"] = {u["name"]: u for u in state["unsubscribed"]}
            if "streams" in state:
                for stream in state["streams"]:
                    if "subscriber_count" in stream:
                        # this isn't guaranteed to match
                        del stream["subscriber_count"]
            if "realm_bots" in state:
                state["realm_bots"] = {u["email"]: u for u in state["realm_bots"]}
            # Since time is different for every call, just fix the value
            state["server_timestamp"] = 0
            if "presence_last_update_id" in state:
                # We don't adjust presence_last_update_id via apply_events,
                # since events don't carry the relevant information.
                # Fix the value just like server_timestamp.
                state["presence_last_update_id"] = 0

        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, v1 in state1.items():
                if v1 != state2[k]:
                    print("\nkey = " + k)
                    try:
                        self.assertEqual({k: v1}, {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.
                """,
                flush=True,
            )

            raise AssertionError("Mismatching states")


class NormalActionsTest(BaseAction):
    def create_bot(self, email: str, **extras: Any) -> UserProfile:
        return self.create_test_bot(email, self.user_profile, **extras)

    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)
            with self.verify_action():
                self.send_stream_message(
                    self.example_user("cordelia"),
                    "Verona",
                    content,
                    skip_capture_on_commit_callbacks=True,
                )

    def test_automatically_follow_topic_where_mentioned(self) -> None:
        user = self.user_profile

        do_change_user_setting(
            user_profile=user,
            setting_name="automatically_follow_topics_where_mentioned",
            setting_value=True,
            acting_user=None,
        )

        def get_num_events() -> int:  # nocoverage
            try:
                user_topic = UserTopic.objects.get(
                    user_profile=user,
                    stream_id=get_stream("Verona", user.realm).id,
                    topic_name__iexact="test",
                )
                if user_topic.visibility_policy != UserTopic.VisibilityPolicy.FOLLOWED:
                    return 3
            except UserTopic.DoesNotExist:
                return 3
            return 1

        for i in range(3):
            content = "mentioning... @**" + user.full_name + "** hello " + str(i)
            with self.verify_action(num_events=get_num_events()):
                self.send_stream_message(
                    self.example_user("cordelia"),
                    "Verona",
                    content,
                    skip_capture_on_commit_callbacks=True,
                )

    def test_topic_wildcard_mentioned_send_message_events(self) -> None:
        for i in range(3):
            content = "mentioning... @**topic** hello " + str(i)
            with self.verify_action():
                self.send_stream_message(
                    self.example_user("cordelia"),
                    "Verona",
                    content,
                    skip_capture_on_commit_callbacks=True,
                )

    def test_stream_wildcard_mentioned_send_message_events(self) -> None:
        for i in range(3):
            content = "mentioning... @**all** hello " + str(i)
            with self.verify_action():
                self.send_stream_message(
                    self.example_user("cordelia"),
                    "Verona",
                    content,
                    skip_capture_on_commit_callbacks=True,
                )

    def test_pm_send_message_events(self) -> None:
        with self.verify_action() as events:
            self.send_personal_message(
                self.example_user("cordelia"),
                self.example_user("hamlet"),
                "hola",
                skip_capture_on_commit_callbacks=True,
            )
        self.assertEqual(events[0]["message"][TOPIC_NAME], "")

        # Verify direct message editing - content only edit
        pm = Message.objects.order_by("-id")[0]
        content = "new content"
        rendering_result = render_message_markdown(pm, content)
        prior_mention_user_ids: set[int] = set()
        mention_backend = MentionBackend(self.user_profile.realm_id)
        mention_data = MentionData(
            mention_backend=mention_backend,
            content=content,
            message_sender=self.example_user("cordelia"),
        )

        message_edit_request = build_message_edit_request(
            message=pm,
            user_profile=self.user_profile,
            propagate_mode="change_one",
            stream_id=None,
            topic_name=None,
            content=content,
        )
        with self.verify_action(state_change_expected=False) as events:
            do_update_message(
                self.user_profile,
                pm,
                message_edit_request,
                False,
                False,
                rendering_result,
                prior_mention_user_ids,
                mention_data,
            )
        check_update_message(
            "events[0]",
            events[0],
            is_stream_message=False,
            has_content=True,
            has_topic=False,
            has_new_stream_id=False,
            is_embedded_update_only=False,
        )

    def test_pm_send_message_events_via_direct_message_group(self) -> None:
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")

        # Create a direct message group with hamlet and cordelia
        get_or_create_direct_message_group(id_list=[hamlet.id, cordelia.id])

        with self.verify_action():
            self.send_group_direct_message(
                from_user=hamlet,
                to_users=[hamlet, cordelia],
                content="hola",
                skip_capture_on_commit_callbacks=True,
            )

    def test_direct_message_group_send_message_events(self) -> None:
        direct_message_group = [
            self.example_user("hamlet"),
            self.example_user("othello"),
        ]
        with self.verify_action() as events:
            self.send_group_direct_message(
                self.example_user("cordelia"),
                direct_message_group,
                "hola",
                skip_capture_on_commit_callbacks=True,
            )
        self.assertEqual(events[0]["message"][TOPIC_NAME], "")

    def test_user_creation_events_on_sending_messages(self) -> None:
        self.set_up_db_for_testing_user_access()
        polonius = self.example_user("polonius")
        cordelia = self.example_user("cordelia")

        self.user_profile = polonius

        # Test that guest will not receive creation event
        # for bots as they can access all the bots.
        bot = self.create_test_bot("test2", cordelia, full_name="Test bot")
        with self.verify_action(num_events=1) as events:
            self.send_personal_message(bot, polonius, "hola", skip_capture_on_commit_callbacks=True)
        check_direct_message("events[0]", events[0])

        with self.verify_action(num_events=2) as events:
            self.send_personal_message(
                cordelia, polonius, "hola", skip_capture_on_commit_callbacks=True
            )
        check_realm_user_add("events[0]", events[0])
        check_direct_message("events[1]", events[1])
        self.assertEqual(events[0]["person"]["user_id"], cordelia.id)

        othello = self.example_user("othello")
        desdemona = self.example_user("desdemona")

        with self.verify_action(num_events=3) as events:
            self.send_group_direct_message(
                othello, [polonius, desdemona, bot], "hola", skip_capture_on_commit_callbacks=True
            )
        check_realm_user_add("events[0]", events[0])
        check_realm_user_add("events[1]", events[1])
        check_direct_message("events[2]", events[2])
        user_creation_user_ids = {events[0]["person"]["user_id"], events[1]["person"]["user_id"]}
        self.assertEqual(user_creation_user_ids, {othello.id, desdemona.id})

    def test_stream_send_message_events(self) -> None:
        hamlet = self.user_profile
        for stream_name in ["Verona", "Denmark", "core team"]:
            stream = get_stream(stream_name, hamlet.realm)
            sub = get_subscription(stream.name, hamlet)
            do_change_subscription_property(hamlet, sub, stream, "is_muted", True, acting_user=None)

        def verify_events_generated_and_reset_visibility_policy(
            events: list[dict[str, Any]], stream_name: str, topic_name: str
        ) -> None:
            # event-type: muted_topics
            check_muted_topics("events[0]", events[0])
            # event-type: user_topic
            check_user_topic("events[1]", events[1])

            if events[2]["type"] == "message":
                check_message("events[2]", events[2])
            else:
                # event-type: reaction
                check_reaction_add("events[2]", events[2])

            # Reset visibility policy
            do_set_user_topic_visibility_policy(
                hamlet,
                get_stream(stream_name, hamlet.realm),
                topic_name,
                visibility_policy=UserTopic.VisibilityPolicy.INHERIT,
            )

        # Events generated during send message action depends on the 'automatically_follow_topics_policy'
        # and 'automatically_unmute_topics_in_muted_streams_policy' settings. Here we test all the
        # possible combinations.

        # action: participation
        # 'automatically_follow_topics_policy' | 'automatically_unmute_topics_in_muted_streams_policy' | visibility_policy
        #         ON_PARTICIPATION             |                    ON_INITIATION                      |     FOLLOWED
        #         ON_PARTICIPATION             |                   ON_PARTICIPATION                    |     FOLLOWED
        #         ON_PARTICIPATION             |                       ON_SEND                         |     FOLLOWED
        #         ON_PARTICIPATION             |                        NEVER                          |     FOLLOWED
        message_id = self.send_stream_message(hamlet, "Verona", "hello", "topic")
        message = Message.objects.get(id=message_id)
        do_change_user_setting(
            user_profile=hamlet,
            setting_name="automatically_follow_topics_policy",
            setting_value=UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
            acting_user=None,
        )
        for setting_value in UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_CHOICES:
            do_change_user_setting(
                user_profile=hamlet,
                setting_name="automatically_unmute_topics_in_muted_streams_policy",
                setting_value=setting_value,
                acting_user=None,
            )
            # Three events are generated:
            # 2 for following the topic and 1 for adding reaction.
            with self.verify_action(client_gravatar=False, num_events=3) as events:
                do_add_reaction(hamlet, message, "tada", "1f389", "unicode_emoji")
            verify_events_generated_and_reset_visibility_policy(events, "Verona", "topic")
            do_remove_reaction(hamlet, message, "1f389", "unicode_emoji")

        # action: send
        # 'automatically_follow_topics_policy' | 'automatically_unmute_topics_in_muted_streams_policy' | visibility_policy
        #                ON_SEND               |                    ON_INITIATION                      |     FOLLOWED
        #                ON_SEND               |                   ON_PARTICIPATION                    |     FOLLOWED
