from datetime import timedelta
from unittest import mock

import orjson
from django.utils.timezone import now as timezone_now

from zerver.actions.message_delete import do_delete_messages
from zerver.actions.realm_settings import (
    do_change_realm_permission_group_setting,
    do_set_realm_property,
)
from zerver.actions.streams import (
    do_change_stream_group_based_setting,
    do_change_stream_permission,
    do_set_stream_property,
)
from zerver.actions.user_groups import check_add_user_group
from zerver.lib.message import has_message_access
from zerver.lib.streams import (
    can_access_stream_metadata_user_ids,
    update_stream_active_status_for_realm,
)
from zerver.lib.test_classes import ZulipTestCase, get_topic_messages
from zerver.lib.test_helpers import queries_captured
from zerver.lib.topic import RESOLVED_TOPIC_PREFIX
from zerver.lib.types import UserGroupMembersData
from zerver.lib.url_encoding import stream_message_url
from zerver.lib.user_groups import UserGroupMembershipDetails
from zerver.models import Message, NamedUserGroup, Stream, UserMessage, UserProfile
from zerver.models.groups import SystemGroups
from zerver.models.realms import get_realm
from zerver.models.streams import StreamTopicsPolicyEnum, get_stream
from zerver.tornado.django_api import send_event_on_commit


class MessageMoveStreamTest(ZulipTestCase):
    def assert_has_usermessage(self, user_profile_id: int, message_id: int) -> None:
        self.assertEqual(
            UserMessage.objects.filter(
                user_profile_id=user_profile_id, message_id=message_id
            ).exists(),
            True,
        )

    def assert_lacks_usermessage(self, user_profile_id: int, message_id: int) -> None:
        self.assertEqual(
            UserMessage.objects.filter(
                user_profile_id=user_profile_id, message_id=message_id
            ).exists(),
            False,
        )

    def assert_move_message(
        self,
        user: str,
        orig_stream: Stream,
        orig_topic_name: str = "test",
        stream_id: int | None = None,
        topic_name: str | None = None,
        expected_error: str | None = None,
    ) -> None:
        user_profile = self.example_user(user)
        self.subscribe(user_profile, orig_stream.name)
        message_id = self.send_stream_message(
            user_profile, orig_stream.name, topic_name=orig_topic_name
        )

        params_dict: dict[str, str | int] = {}
        if stream_id is not None:
            params_dict["stream_id"] = stream_id
        if topic_name is not None:
            params_dict["topic"] = topic_name

        result = self.api_patch(
            user_profile,
            "/api/v1/messages/" + str(message_id),
            params_dict,
        )
        if expected_error is not None:
            self.assert_json_error(result, expected_error)
        else:
            self.assert_json_success(result)

    def prepare_move_topics(
        self,
        user_email: str,
        old_stream: str,
        new_stream: str,
        topic_name: str,
        language: str | None = None,
    ) -> tuple[UserProfile, Stream, Stream, int, int]:
        user_profile = self.example_user(user_email)
        if language is not None:
            user_profile.default_language = language
            user_profile.save(update_fields=["default_language"])

        self.login(user_email)
        try:
            stream = get_stream(old_stream, user_profile.realm)
            messages = get_topic_messages(user_profile, stream, "test")
            do_delete_messages(user_profile.realm, messages, acting_user=None)
        except Stream.DoesNotExist:
            stream = self.make_stream(old_stream)
        try:
            stream_to = get_stream(new_stream, user_profile.realm)
            messages = get_topic_messages(user_profile, stream_to, "test")
            do_delete_messages(user_profile.realm, messages, acting_user=None)
        except Stream.DoesNotExist:
            stream_to = self.make_stream(new_stream)
        self.subscribe(user_profile, stream.name)
        self.subscribe(user_profile, stream_to.name)
        msg_id = self.send_stream_message(
            user_profile, stream.name, topic_name=topic_name, content="First"
        )
        msg_id_lt = self.send_stream_message(
            user_profile, stream.name, topic_name=topic_name, content="Second"
        )

        self.send_stream_message(user_profile, stream.name, topic_name=topic_name, content="third")

        return (user_profile, stream, stream_to, msg_id, msg_id_lt)

    def test_move_message_cant_move_private_message(self) -> None:
        hamlet = self.example_user("hamlet")
        self.login("hamlet")
        cordelia = self.example_user("cordelia")
        msg_id = self.send_personal_message(hamlet, cordelia)

        verona = get_stream("Verona", hamlet.realm)

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "stream_id": verona.id,
            },
        )

        self.assert_json_error(result, "Direct messages cannot be moved to channels.")

    def test_move_message_to_stream_with_content(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, _msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
                "content": "Not allowed",
            },
        )
        self.assert_json_error(result, "Cannot change message content while changing channel")

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 3)

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 0)

    def test_change_all_propagate_mode_for_moving_old_messages(self) -> None:
        user_profile = self.example_user("hamlet")
        realm = user_profile.realm
        id1 = self.send_stream_message(user_profile, "Denmark", topic_name="topic1")
        id2 = self.send_stream_message(user_profile, "Denmark", topic_name="topic1")
        id3 = self.send_stream_message(user_profile, "Denmark", topic_name="topic1")
        id4 = self.send_stream_message(user_profile, "Denmark", topic_name="topic1")
        self.send_stream_message(user_profile, "Denmark", topic_name="topic1")

        members_system_group = NamedUserGroup.objects.get(
            name=SystemGroups.MEMBERS, realm_for_sharding=realm, is_system_group=True
        )

        do_change_realm_permission_group_setting(
            realm,
            "can_move_messages_between_channels_group",
            members_system_group,
            acting_user=None,
        )

        message = Message.objects.get(id=id1)
        message.date_sent -= timedelta(days=10)
        message.save()

        message = Message.objects.get(id=id2)
        message.date_sent -= timedelta(days=8)
        message.save()

        message = Message.objects.get(id=id3)
        message.date_sent -= timedelta(days=5)
        message.save()

        verona = get_stream("Verona", user_profile.realm)
        denmark = get_stream("Denmark", user_profile.realm)
        old_topic_name = "topic1"
        old_stream = denmark

        def test_moving_all_topic_messages(
            new_topic_name: str | None = None, new_stream: Stream | None = None
        ) -> None:
            self.login("hamlet")
            params_dict: dict[str, str | int] = {
                "propagate_mode": "change_all",
                "send_notification_to_new_thread": "false",
            }

            if new_topic_name is not None:
                params_dict["topic"] = new_topic_name
            else:
                new_topic_name = old_topic_name

            if new_stream is not None:
                params_dict["stream_id"] = new_stream.id
            else:
                new_stream = old_stream

            result = self.client_patch(
                f"/json/messages/{id4}",
                params_dict,
            )
            self.assert_json_error(
                result,
                "You only have permission to move the 3/5 most recent messages in this topic.",
            )
            # Check message count in old topic and/or stream.
            messages = get_topic_messages(user_profile, old_stream, old_topic_name)
            self.assert_length(messages, 5)

            # Check message count in new topic and/or stream.
            messages = get_topic_messages(user_profile, new_stream, new_topic_name)
            self.assert_length(messages, 0)

            json = orjson.loads(result.content)
            first_message_id_allowed_to_move = json["first_message_id_allowed_to_move"]

            params_dict["propagate_mode"] = "change_later"
            result = self.client_patch(
                f"/json/messages/{first_message_id_allowed_to_move}",
                params_dict,
            )
            self.assert_json_success(result)

            # Check message count in old topic and/or stream.
            messages = get_topic_messages(user_profile, old_stream, old_topic_name)
            self.assert_length(messages, 2)

            # Check message count in new topic and/or stream.
            messages = get_topic_messages(user_profile, new_stream, new_topic_name)
            self.assert_length(messages, 3)

            self.login("shiva")
            # Move these messages to the original topic and stream, to test the case
            # when user is moderator.
            result = self.client_patch(
                f"/json/messages/{id4}",
                {
                    "topic": old_topic_name,
                    "stream_id": old_stream.id,
                    "propagate_mode": "change_all",
                    "send_notification_to_new_thread": "false",
                },
            )

            params_dict["propagate_mode"] = "change_all"
            result = self.client_patch(
                f"/json/messages/{id4}",
                params_dict,
            )
            self.assert_json_success(result)

            # Check message count in old topic and/or stream.
            messages = get_topic_messages(user_profile, old_stream, old_topic_name)
            self.assert_length(messages, 0)

            # Check message count in new topic and/or stream.
            messages = get_topic_messages(user_profile, new_stream, new_topic_name)
            self.assert_length(messages, 5)

        # Test only topic editing case.
        test_moving_all_topic_messages(new_topic_name="topic edited")

        # Move these messages to the original topic to test the next case.
        self.client_patch(
            f"/json/messages/{id4}",
            {
                "topic": old_topic_name,
                "propagate_mode": "change_all",
                "send_notification_to_new_thread": "false",
            },
        )

        # Test only stream editing case
        test_moving_all_topic_messages(new_stream=verona)

        # Move these messages to the original stream to test the next case.
        self.client_patch(
            f"/json/messages/{id4}",
            {
                "stream_id": denmark.id,
                "propagate_mode": "change_all",
                "send_notification_to_new_thread": "false",
            },
        )

        # Set time limit for moving messages between streams to 2 weeks.
        do_set_realm_property(
            user_profile.realm,
            "move_messages_between_streams_limit_seconds",
            604800 * 2,
            acting_user=None,
        )

        # Test editing both topic and stream together.
        test_moving_all_topic_messages(new_topic_name="edited", new_stream=verona)

        # Move these messages to the original stream and topic to test the next case.
        self.client_patch(
            f"/json/messages/{id4}",
            {
                "stream_id": denmark.id,
                "topic": old_topic_name,
                "propagate_mode": "change_all",
                "send_notification_to_new_thread": "false",
            },
        )

        # Test editing both topic and stream with no limit set.
        self.login("hamlet")
        do_set_realm_property(
            user_profile.realm,
            "move_messages_within_stream_limit_seconds",
            None,
            acting_user=None,
        )
        do_set_realm_property(
            user_profile.realm,
            "move_messages_between_streams_limit_seconds",
            None,
            acting_user=None,
        )

        new_stream = verona
        new_topic_name = "edited"
        result = self.client_patch(
            f"/json/messages/{id4}",
            {
                "topic": new_topic_name,
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
                "send_notification_to_new_thread": "false",
            },
        )
        self.assert_json_success(result)
        # Check message count in old topic and/or stream.
        messages = get_topic_messages(user_profile, old_stream, old_topic_name)
        self.assert_length(messages, 0)

        # Check message count in new topic and/or stream.
        messages = get_topic_messages(user_profile, new_stream, new_topic_name)
        self.assert_length(messages, 5)

    def test_move_message_to_stream(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, _msg_id_lt) = self.prepare_move_topics(
            "iago",
            "test move stream",
            "new stream",
            "test",
            # Set the user's translation language to German to test that
            # it is overridden by the realm's default language.
            "de",
        )

        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
                "send_notification_to_old_thread": "true",
            },
            HTTP_ACCEPT_LANGUAGE="de",
        )

        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 1)
        self.assertEqual(
            messages[0].content,
            f"This topic was moved to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 4)
        self.assertEqual(
            messages[3].content,
            f"This topic was moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_preexisting_topic(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, _msg_id_lt) = self.prepare_move_topics(
            "iago",
            "test move stream",
            "new stream",
            "test",
            # Set the user's translation language to German to test that
            # it is overridden by the realm's default language.
            "de",
        )

        self.send_stream_message(
            sender=self.example_user("iago"),
            stream_name="new stream",
            topic_name="test",
            content="Always here",
        )

        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
                "send_notification_to_old_thread": "true",
            },
            HTTP_ACCEPT_LANGUAGE="de",
        )

        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 1)
        self.assertEqual(
            messages[0].content,
            f"This topic was moved to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 5)
        self.assertEqual(
            messages[4].content,
            f"3 messages were moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_realm_admin_cant_move_to_another_realm(self) -> None:
        user_profile = self.example_user("iago")
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.login("iago")

        lear_realm = get_realm("lear")
        new_stream = self.make_stream("new", lear_realm)

        msg_id = self.send_stream_message(user_profile, "Verona", topic_name="test123")

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
            },
        )

        self.assert_json_error(result, "Invalid channel ID")

    def test_move_message_realm_admin_cant_move_to_private_stream_without_subscription(
        self,
    ) -> None:
        user_profile = self.example_user("iago")
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.login("iago")

        new_stream = self.make_stream("new", invite_only=True)
        msg_id = self.send_stream_message(user_profile, "Verona", topic_name="test123")

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
            },
        )

        self.assert_json_error(result, "Invalid channel ID")

    def test_move_message_realm_admin_cant_move_from_private_stream_without_subscription(
        self,
    ) -> None:
        user_profile = self.example_user("iago")
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.login("iago")

        self.make_stream("privatestream", invite_only=True)
        self.subscribe(user_profile, "privatestream")
        msg_id = self.send_stream_message(user_profile, "privatestream", topic_name="test123")
        self.unsubscribe(user_profile, "privatestream")

        verona = get_stream("Verona", user_profile.realm)

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "stream_id": verona.id,
                "propagate_mode": "change_all",
            },
        )

        self.assert_json_error(
            result,
            "Invalid message(s)",
        )

    def test_move_message_from_private_stream_message_access_checks(
        self,
    ) -> None:
        hamlet = self.example_user("hamlet")
        user_profile = self.example_user("iago")
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.login("iago")

        private_stream = self.make_stream(
            "privatestream", invite_only=True, history_public_to_subscribers=False
        )
        self.subscribe(hamlet, "privatestream")
        original_msg_id = self.send_stream_message(hamlet, "privatestream", topic_name="test123")
        self.subscribe(user_profile, "privatestream")
        new_msg_id = self.send_stream_message(user_profile, "privatestream", topic_name="test123")

        # Now we unsub and hamlet sends a new message (we won't have access to it even after re-subbing!)
        self.unsubscribe(user_profile, "privatestream")
        new_inaccessible_msg_id = self.send_stream_message(
            hamlet, "privatestream", topic_name="test123"
        )

        # Re-subscribe and send another message:
        self.subscribe(user_profile, "privatestream")
        newest_msg_id = self.send_stream_message(
            user_profile, "privatestream", topic_name="test123"
        )

        verona = get_stream("Verona", user_profile.realm)

        result = self.client_patch(
            "/json/messages/" + str(new_msg_id),
            {
                "stream_id": verona.id,
                "propagate_mode": "change_all",
            },
        )

        self.assert_json_success(result)
        self.assertEqual(Message.objects.get(id=new_msg_id).recipient_id, verona.recipient_id)
        self.assertEqual(Message.objects.get(id=newest_msg_id).recipient_id, verona.recipient_id)
        # The original message and the new, inaccessible message weren't moved,
        # because user_profile doesn't have access to them.
        self.assertEqual(
            Message.objects.get(id=original_msg_id).recipient_id, private_stream.recipient_id
        )
        self.assertEqual(
            Message.objects.get(id=new_inaccessible_msg_id).recipient_id,
            private_stream.recipient_id,
        )

    def test_move_message_to_stream_change_later(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        result = self.client_patch(
            f"/json/messages/{msg_id_later}",
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_later",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 2)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[1].content,
            f"2 messages were moved from this topic to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 3)
        self.assertEqual(messages[0].id, msg_id_later)
        self.assertEqual(
            messages[2].content,
            f"2 messages were moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_preexisting_topic_change_later(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        self.send_stream_message(
            sender=self.example_user("iago"),
            stream_name="new stream",
            topic_name="test",
            content="Always here",
        )

        result = self.client_patch(
            f"/json/messages/{msg_id_later}",
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_later",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 2)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[1].content,
            f"2 messages were moved from this topic to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 4)
        self.assertEqual(messages[0].id, msg_id_later)
        self.assertEqual(
            messages[3].content,
            f"2 messages were moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_stream_change_later_all_moved(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, _msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_later",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 1)
        self.assertEqual(
            messages[0].content,
            f"This topic was moved to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 4)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[3].content,
            f"This topic was moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_preexisting_topic_change_later_all_moved(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, _msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        self.send_stream_message(
            sender=self.example_user("iago"),
            stream_name="new stream",
            topic_name="test",
            content="Always here",
        )

        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_later",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 1)
        self.assertEqual(
            messages[0].content,
            f"This topic was moved to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 5)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[4].content,
            f"3 messages were moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_stream_change_one(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        result = self.client_patch(
            "/json/messages/" + str(msg_id_later),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_one",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 3)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[2].content,
            f"A message was moved from this topic to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        message = {
            "id": msg_id_later,
            "stream_id": new_stream.id,
            "display_recipient": new_stream.name,
            "topic": "test",
        }
        moved_message_link = stream_message_url(messages[1].realm, message)
        self.assert_length(messages, 2)
        self.assertEqual(messages[0].id, msg_id_later)
        self.assertEqual(
            messages[1].content,
            f"[A message]({moved_message_link}) was moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_preexisting_topic_change_one(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        self.send_stream_message(
            sender=self.example_user("iago"),
            stream_name="new stream",
            topic_name="test",
            content="Always here",
        )

        result = self.client_patch(
            "/json/messages/" + str(msg_id_later),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_one",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 3)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[2].content,
            f"A message was moved from this topic to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        message = {
            "id": msg_id_later,
            "stream_id": new_stream.id,
            "display_recipient": new_stream.name,
            "topic": "test",
        }
        moved_message_link = stream_message_url(messages[2].realm, message)
        self.assert_length(messages, 3)
        self.assertEqual(messages[0].id, msg_id_later)
        self.assertEqual(
            messages[2].content,
            f"[A message]({moved_message_link}) was moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_stream_change_all(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        result = self.client_patch(
            "/json/messages/" + str(msg_id_later),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 1)
        self.assertEqual(
            messages[0].content,
            f"This topic was moved to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 4)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[3].content,
            f"This topic was moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_to_preexisting_topic_change_all(self) -> None:
        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
            "iago", "test move stream", "new stream", "test"
        )

        self.send_stream_message(
            sender=self.example_user("iago"),
            stream_name="new stream",
            topic_name="test",
            content="Always here",
        )

        result = self.client_patch(
            "/json/messages/" + str(msg_id_later),
            {
                "stream_id": new_stream.id,
                "propagate_mode": "change_all",
                "send_notification_to_old_thread": "true",
            },
        )
        self.assert_json_success(result)

        messages = get_topic_messages(user_profile, old_stream, "test")
        self.assert_length(messages, 1)
        self.assertEqual(
            messages[0].content,
            f"This topic was moved to #**new stream>test** by @_**Iago|{user_profile.id}**.",
        )

        messages = get_topic_messages(user_profile, new_stream, "test")
        self.assert_length(messages, 5)
        self.assertEqual(messages[0].id, msg_id)
        self.assertEqual(
            messages[4].content,
            f"3 messages were moved here from #**test move stream>test** by @_**Iago|{user_profile.id}**.",
        )

    def test_move_message_between_streams_policy_setting(self) -> None:
        othello = self.example_user("othello")
        cordelia = self.example_user("cordelia")
        realm = othello.realm

        def check_move_message_according_to_permission(
            username: str, expect_fail: bool = False
        ) -> None:
            (user_profile, old_stream, new_stream, msg_id, _msg_id_later) = (
                self.prepare_move_topics(username, "old_stream", "new_stream", "test")
            )
            result = self.client_patch(
                "/json/messages/" + str(msg_id),
                {
                    "stream_id": new_stream.id,
                    "propagate_mode": "change_all",
                },
            )

            if expect_fail:
                self.assert_json_error(result, "You don't have permission to move this message")
