from datetime import timedelta
from operator import itemgetter
from typing import Literal
from unittest import mock

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

from zerver.actions.message_edit import get_mentions_for_message_updates
from zerver.actions.realm_settings import (
    do_change_realm_permission_group_setting,
    do_change_realm_plan_type,
    do_set_realm_property,
)
from zerver.actions.streams import do_change_stream_group_based_setting, do_deactivate_stream
from zerver.actions.user_groups import add_subgroups_to_user_group, check_add_user_group
from zerver.actions.user_settings import do_change_user_setting
from zerver.actions.user_topics import do_set_user_topic_visibility_policy
from zerver.lib import utils
from zerver.lib.message import messages_for_ids
from zerver.lib.message_cache import MessageDict
from zerver.lib.stream_topic import StreamTopicTarget
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import most_recent_message, queries_captured
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.lib.topic import TOPIC_NAME
from zerver.lib.utils import assert_is_not_none
from zerver.models import (
    Attachment,
    Message,
    NamedUserGroup,
    Realm,
    Subscription,
    UserProfile,
    UserTopic,
)
from zerver.models.groups import SystemGroups
from zerver.models.messages import UserMessage
from zerver.models.realms import MessageEditHistoryVisibilityPolicyEnum, get_realm
from zerver.models.streams import get_stream


class EditMessageTest(ZulipTestCase):
    def check_message(self, msg_id: int, topic_name: str, content: str) -> None:
        # Make sure we saved the message correctly to the DB.
        msg = Message.objects.select_related("realm").get(id=msg_id)
        self.assertEqual(msg.topic_name(), topic_name)
        self.assertEqual(msg.content, content)

        """
        We assume our caller just edited a message.

        Next, we will make sure we properly cached the messages.  We still have
        to do a query to hydrate recipient info, but we won't need to hit the
        zerver_message table.
        """

        with queries_captured(keep_cache_warm=True) as queries:
            (fetch_message_dict,) = messages_for_ids(
                message_ids=[msg.id],
                user_message_flags={msg_id: []},
                search_fields={},
                apply_markdown=False,
                client_gravatar=False,
                allow_empty_topic_name=True,
                message_edit_history_visibility_policy=MessageEditHistoryVisibilityPolicyEnum.all.value,
                user_profile=None,
                realm=msg.realm,
            )

        self.assert_length(queries, 1)
        for query in queries:
            self.assertNotIn("message", query.sql)

        self.assertEqual(
            fetch_message_dict[TOPIC_NAME],
            msg.topic_name(),
        )
        self.assertEqual(
            fetch_message_dict["content"],
            msg.content,
        )
        self.assertEqual(
            fetch_message_dict["sender_id"],
            msg.sender_id,
        )

        if msg.edit_history:
            message_edit_history = orjson.loads(msg.edit_history)
            for item in message_edit_history:
                if "prev_rendered_content_version" in item:
                    del item["prev_rendered_content_version"]

            self.assertEqual(
                fetch_message_dict["edit_history"],
                message_edit_history,
            )

    def check_message_flags(
        self, message_id: int, user_ids: list[int], flag: str, check_present: bool
    ) -> None:
        # Make sure we updated the message flags correctly to the DB.
        for user_id in user_ids:
            um = UserMessage.objects.get(
                user_profile_id=user_id,
                message_id=message_id,
            )
            if check_present:
                self.assertIn(flag, um.flags_list())
            else:
                self.assertNotIn(flag, um.flags_list())

    def test_edit_message_no_changes(self) -> None:
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("hamlet"), "Denmark", topic_name="editing", content="before edit"
        )
        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {},
        )
        self.assert_json_error(result, "Nothing to change")

    # Right now, we prevent users from editing widgets.
    def test_edit_submessage(self) -> None:
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="/poll Games?\nYES\nNO",
        )
        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "content": "/poll Games?\nYES\nNO\nMaybe",
            },
        )
        self.assert_json_error(result, "Widgets cannot be edited.")

    def test_query_count_on_messages_to_encoded_cache(self) -> None:
        # `messages_to_encoded_cache` method is used by the mechanisms
        # tested in this class. Hence, its performance is tested here.
        # Generate 2 messages
        user = self.example_user("hamlet")
        realm = user.realm
        self.login_user(user)
        stream_name = "public_stream"
        self.subscribe(user, stream_name)
        message_ids = []
        message_ids.append(self.send_stream_message(user, stream_name, "Message one"))
        user_2 = self.example_user("cordelia")
        self.subscribe(user_2, stream_name)
        message_ids.append(self.send_stream_message(user_2, stream_name, "Message two"))
        self.subscribe(self.notification_bot(realm), stream_name)
        message_ids.append(
            self.send_stream_message(self.notification_bot(realm), stream_name, "Message three")
        )
        messages = [
            Message.objects.select_related(*Message.DEFAULT_SELECT_RELATED).get(id=message_id)
            for message_id in message_ids
        ]

        # Check number of queries performed
        # 1 query for realm_id per message = 3
        # 1 query each for reactions & submessage for all messages = 2
        # 1 query for linkifiers
        # 1 query for display recipients
        with self.assert_database_query_count(7):
            MessageDict.messages_to_encoded_cache(messages)

        realm_id = 2  # Fetched from stream object
        # Check number of queries performed with realm_id
        with self.assert_database_query_count(3):
            MessageDict.messages_to_encoded_cache(messages, realm_id)

    def test_save_message(self) -> None:
        """This is also tested by a client test, but here we can verify
        the cache against the database"""
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("hamlet"), "Denmark", topic_name="editing", content="before edit"
        )
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "after edit",
            },
        )
        self.assert_json_success(result)
        self.check_message(msg_id, topic_name="editing", content="after edit")

        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "topic": "edited",
            },
        )
        self.assert_json_success(result)
        self.assertEqual(Message.objects.get(id=msg_id).topic_name(), "edited")

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

        msg_id = self.send_personal_message(
            from_user=hamlet, to_user=cordelia, content="Outgoing direct message"
        )
        result = self.client_get("/json/messages/" + str(msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "Outgoing direct message")
        self.assertEqual(response_dict["message"]["id"], msg_id)
        self.assertEqual(response_dict["message"]["recipient_id"], cordelia.recipient_id)
        self.assertEqual(response_dict["message"]["flags"], ["read"])
        self.assertEqual(response_dict["message"][TOPIC_NAME], "")

        msg_id = self.send_personal_message(
            from_user=cordelia, to_user=hamlet, content="Incoming direct message"
        )
        result = self.client_get("/json/messages/" + str(msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "Incoming direct message")
        self.assertEqual(response_dict["message"]["id"], msg_id)
        # Incoming DMs show the recipient_id that outgoing DMs would.
        self.assertEqual(response_dict["message"]["recipient_id"], cordelia.recipient_id)
        self.assertEqual(response_dict["message"]["flags"], [])
        self.assertEqual(response_dict["message"][TOPIC_NAME], "")

        # Send message to web-public stream where hamlet is not subscribed.
        # This will test case of user having no `UserMessage` but having access
        # to message.
        web_public_stream = self.make_stream("web-public-stream", is_web_public=True)
        self.subscribe(self.example_user("cordelia"), web_public_stream.name)
        web_public_stream_msg_id = self.send_stream_message(
            self.example_user("cordelia"), web_public_stream.name, content="web-public message"
        )
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "web-public message")
        self.assertEqual(response_dict["message"]["id"], web_public_stream_msg_id)
        self.assertEqual(response_dict["message"]["flags"], ["read", "historical"])

        # Spectator should be able to fetch message in web-public stream.
        self.logout()
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "web-public message")
        self.assertEqual(response_dict["message"]["id"], web_public_stream_msg_id)

        # Verify default is apply_markdown=True
        self.assertEqual(response_dict["message"]["content"], "<p>web-public message</p>")

        # Verify apply_markdown=False works correctly.
        result = self.client_get(
            "/json/messages/" + str(web_public_stream_msg_id), {"apply_markdown": "false"}
        )
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "web-public message")
        self.assertEqual(response_dict["message"]["content"], "web-public message")

        with self.settings(WEB_PUBLIC_STREAMS_ENABLED=False):
            result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", status_code=401
        )

        # Test error cases
        self.login("hamlet")
        result = self.client_get("/json/messages/999999")
        self.assert_json_error(result, "Invalid message(s)")

        self.login("cordelia")
        result = self.client_get(f"/json/messages/{msg_id}")
        self.assert_json_success(result)

        self.login("othello")
        result = self.client_get(f"/json/messages/{msg_id}")
        self.assert_json_error(result, "Invalid message(s)")

    def test_fetch_raw_message_spectator(self) -> None:
        user_profile = self.example_user("iago")
        self.login("iago")
        web_public_stream = self.make_stream("web-public-stream", is_web_public=True)
        self.subscribe(user_profile, web_public_stream.name)

        web_public_stream_msg_id = self.send_stream_message(
            user_profile, web_public_stream.name, content="web-public message"
        )

        non_web_public_stream = self.make_stream("non-web-public-stream")
        self.subscribe(user_profile, non_web_public_stream.name)
        non_web_public_stream_msg_id = self.send_stream_message(
            user_profile, non_web_public_stream.name, content="non-web-public message"
        )

        # Generate a direct message to use in verification.
        private_message_id = self.send_personal_message(user_profile, user_profile)

        invalid_message_id = private_message_id + 1000

        self.logout()

        # Confirm WEB_PUBLIC_STREAMS_ENABLED is enforced.
        with self.settings(WEB_PUBLIC_STREAMS_ENABLED=False):
            result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )

        do_set_realm_property(
            user_profile.realm, "enable_spectator_access", False, acting_user=None
        )
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )
        do_set_realm_property(user_profile.realm, "enable_spectator_access", True, acting_user=None)

        # Verify success with web-public stream and default SELF_HOSTED plan type.
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "web-public message")
        self.assertEqual(response_dict["message"]["flags"], ["read"])

        # Verify LIMITED plan type does not allow web-public access.
        do_change_realm_plan_type(user_profile.realm, Realm.PLAN_TYPE_LIMITED, acting_user=None)
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )

        do_set_realm_property(user_profile.realm, "enable_spectator_access", True, acting_user=None)
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )

        # Verify works with STANDARD_FREE plan type too.
        do_change_realm_plan_type(
            user_profile.realm, Realm.PLAN_TYPE_STANDARD_FREE, acting_user=None
        )
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "web-public message")

        # Verify direct messages are rejected.
        result = self.client_get("/json/messages/" + str(private_message_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )

        # Verify an actual public stream is required.
        result = self.client_get("/json/messages/" + str(non_web_public_stream_msg_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )

        # Verify invalid message IDs are rejected with the same error message.
        result = self.client_get("/json/messages/" + str(invalid_message_id))
        self.assert_json_error(
            result, "Not logged in: API authentication or user session required", 401
        )

        # Verify success with deactivated streams.
        do_deactivate_stream(web_public_stream, acting_user=None)
        result = self.client_get("/json/messages/" + str(web_public_stream_msg_id))
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["raw_content"], "web-public message")
        self.assertEqual(response_dict["message"]["flags"], ["read"])

    def test_fetch_raw_message_stream_wrong_realm(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        stream = self.make_stream("public_stream")
        self.subscribe(user_profile, stream.name)
        msg_id = self.send_stream_message(
            user_profile, stream.name, topic_name="test", content="test"
        )
        result = self.client_get(f"/json/messages/{msg_id}")
        self.assert_json_success(result)

        mit_user = self.mit_user("sipbtest")
        self.login_user(mit_user)
        result = self.client_get(f"/json/messages/{msg_id}", subdomain="zephyr")
        self.assert_json_error(result, "Invalid message(s)")

    def test_fetch_raw_message_private_stream(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        stream = self.make_stream("private_stream", invite_only=True)
        self.subscribe(user_profile, stream.name)
        msg_id = self.send_stream_message(
            user_profile, stream.name, topic_name="test", content="test"
        )
        result = self.client_get(f"/json/messages/{msg_id}")
        self.assert_json_success(result)
        self.login("othello")
        result = self.client_get(f"/json/messages/{msg_id}")
        self.assert_json_error(result, "Invalid message(s)")

    def test_edit_message_no_permission(self) -> None:
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("iago"), "Denmark", topic_name="editing", content="before edit"
        )
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "content after edit",
            },
        )
        self.assert_json_error(result, "You don't have permission to edit this message")

        self.login("iago")
        realm = get_realm("zulip")
        do_set_realm_property(realm, "allow_message_editing", False, acting_user=None)
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "content after edit",
            },
        )
        self.assert_json_error(result, "Your organization has turned off message editing")

    def test_edit_message_no_content(self) -> None:
        self.login("hamlet")
        # Check message edit in stream for no content.
        msg_id = self.send_stream_message(
            self.example_user("hamlet"), "Denmark", topic_name="editing", content="before edit"
        )
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": " ",
            },
        )
        self.assert_json_success(result)
        content = Message.objects.filter(id=msg_id).values_list("content", flat=True)[0]
        self.assertEqual(content, "(deleted)")

        # Check message edit in DMs for no content.
        msg_id = self.send_personal_message(
            from_user=self.example_user("hamlet"), to_user=self.example_user("cordelia")
        )
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": " ",
            },
        )
        self.assert_json_success(result)
        content = Message.objects.filter(id=msg_id).values_list("content", flat=True)[0]
        self.assertEqual(content, "(deleted)")

    def test_edit_message_in_unsubscribed_private_stream(self) -> None:
        hamlet = self.example_user("hamlet")
        self.login("hamlet")

        stream = self.make_stream(
            "privatestream", invite_only=True, history_public_to_subscribers=False
        )
        self.subscribe(hamlet, "privatestream")
        msg_id = self.send_stream_message(
            hamlet, "privatestream", topic_name="editing", content="before edit"
        )

        # Ensure the user originally could edit the message. This ensures the
        # loss of the ability is caused by unsubscribing, rather than something
        # else wrong with the test's setup/assumptions.
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "test can edit before unsubscribing",
            },
        )
        self.assert_json_success(result)

        self.unsubscribe(hamlet, "privatestream")
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "after unsubscribing",
            },
        )
        self.assert_json_error(result, "Invalid message(s)")
        content = Message.objects.get(id=msg_id).content
        self.assertEqual(content, "test can edit before unsubscribing")

        hamlet_group = check_add_user_group(
            hamlet.realm,
            "prospero_group",
            [hamlet],
            acting_user=hamlet,
        )
        do_change_stream_group_based_setting(
            stream,
            "can_add_subscribers_group",
            hamlet_group,
            acting_user=hamlet,
        )
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "having content access after unsubscribing",
            },
        )
        self.assert_json_success(result)
        content = Message.objects.get(id=msg_id).content
        self.assertEqual(content, "having content access after unsubscribing")

    def test_edit_message_guest_in_unsubscribed_public_stream(self) -> None:
        guest_user = self.example_user("polonius")
        self.login("polonius")
        self.assertEqual(guest_user.role, UserProfile.ROLE_GUEST)

        self.make_stream("publicstream", invite_only=False)
        self.subscribe(guest_user, "publicstream")
        msg_id = self.send_stream_message(
            guest_user, "publicstream", topic_name="editing", content="before edit"
        )

        # Ensure the user originally could edit the message.
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "test can edit before unsubscribing",
            },
        )
        self.assert_json_success(result)

        self.unsubscribe(guest_user, "publicstream")
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": "after unsubscribing",
            },
        )
        self.assert_json_error(result, "Invalid message(s)")
        content = Message.objects.get(id=msg_id).content
        self.assertEqual(content, "test can edit before unsubscribing")

    def test_edit_message_history_disabled(self) -> None:
        user_profile = self.example_user("hamlet")
        do_set_realm_property(
            user_profile.realm,
            "message_edit_history_visibility_policy",
            MessageEditHistoryVisibilityPolicyEnum.none,
            acting_user=None,
        )
        self.login("hamlet")

        # Single-line edit
        msg_id_1 = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="content before edit",
        )

        new_content_1 = "content after edit"
        result_1 = self.client_patch(
            f"/json/messages/{msg_id_1}",
            {
                "content": new_content_1,
            },
        )
        self.assert_json_success(result_1)

        result = self.client_get(f"/json/messages/{msg_id_1}/history")
        self.assert_json_error(result, "Message edit history is disabled in this organization")

        # Now verify that if we fetch the message directly, there's no
        # edit history data attached.
        message_fetch_result = self.client_get(
            f"/json/messages/{msg_id_1}",
        )
        self.assert_json_success(message_fetch_result)
        message_dict = orjson.loads(message_fetch_result.content)["message"]
        self.assertNotIn("edit_history", message_dict)
        # We still have a last edit timestamp present.
        self.assertIn("last_edit_timestamp", message_dict)

    def test_edit_message_history(self) -> None:
        self.login("hamlet")

        # Single-line edit
        msg_id_1 = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="content before edit",
        )
        new_content_1 = "content after edit"
        result_1 = self.client_patch(
            f"/json/messages/{msg_id_1}",
            {
                "content": new_content_1,
            },
        )
        self.assert_json_success(result_1)

        message_edit_history_1 = self.client_get(f"/json/messages/{msg_id_1}/history")
        json_response_1 = orjson.loads(message_edit_history_1.content)
        message_history_1 = json_response_1["message_history"]

        # Check content of message after edit.
        self.assertEqual(message_history_1[0]["rendered_content"], "<p>content before edit</p>")
        self.assertEqual(message_history_1[1]["rendered_content"], "<p>content after edit</p>")
        self.assertEqual(
            message_history_1[1]["content_html_diff"],
            (
                "<div><p>content "
                '<span class="highlight_text_inserted">after</span> '
                '<span class="highlight_text_deleted">before</span>'
                " edit</p></div>"
            ),
        )
        # Check content of message before edit.
        self.assertEqual(
            message_history_1[1]["prev_rendered_content"], "<p>content before edit</p>"
        )

        # Edits on new lines
        msg_id_2 = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="content before edit, line 1\n\ncontent before edit, line 3",
        )
        new_content_2 = (
            "content before edit, line 1\ncontent after edit, line 2\ncontent before edit, line 3"
        )
        result_2 = self.client_patch(
            f"/json/messages/{msg_id_2}",
            {
                "content": new_content_2,
            },
        )
        self.assert_json_success(result_2)

        message_edit_history_2 = self.client_get(f"/json/messages/{msg_id_2}/history")
        json_response_2 = orjson.loads(message_edit_history_2.content)
        message_history_2 = json_response_2["message_history"]

        self.assertEqual(
            message_history_2[0]["rendered_content"],
            "<p>content before edit, line 1</p>\n<p>content before edit, line 3</p>",
        )
        self.assertEqual(
            message_history_2[1]["rendered_content"],
            (
                "<p>content before edit, line 1<br>\n"
                "content after edit, line 2<br>\n"
                "content before edit, line 3</p>"
            ),
        )
        self.assertEqual(
            message_history_2[1]["content_html_diff"],
            (
                "<div><p>content before edit, line 1<br> "
                'content <span class="highlight_text_inserted">after edit, line 2<br> '
                "content</span> before edit, line 3</p></div>"
            ),
        )
        self.assertEqual(
            message_history_2[1]["prev_rendered_content"],
            "<p>content before edit, line 1</p>\n<p>content before edit, line 3</p>",
        )

    def test_edit_direct_message_history(self) -> None:
        msg_id = self.send_personal_message(
            self.example_user("hamlet"),
            self.example_user("othello"),
            content="content before edit",
        )

        self.login("hamlet")
        new_content = "content after edit"
        result = self.client_patch(
            f"/json/messages/{msg_id}",
            {
                "content": new_content,
            },
        )
        self.assert_json_success(result)

        message_edit_history = self.client_get(f"/json/messages/{msg_id}/history")
        json_response = orjson.loads(message_edit_history.content)
        message_history = json_response["message_history"]
        self.assertEqual(message_history[0]["content"], "content before edit")
        self.assertEqual(message_history[0]["rendered_content"], "<p>content before edit</p>")
        self.assertEqual(message_history[0]["topic"], "")

        self.assertEqual(message_history[1]["prev_content"], "content before edit")
        self.assertEqual(message_history[1]["prev_rendered_content"], "<p>content before edit</p>")
        self.assertEqual(message_history[1]["content"], "content after edit")
        self.assertEqual(message_history[1]["rendered_content"], "<p>content after edit</p>")
        self.assertEqual(message_history[1]["topic"], "")

        self.assertEqual(
            message_history[1]["content_html_diff"],
            '<div><p>content <span class="highlight_text_inserted">after</span> <span class="highlight_text_deleted">before</span> edit</p></div>',
        )

    def test_empty_message_edit(self) -> None:
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="We will edit this to render as empty.",
        )
        # Edit that manually to simulate a rendering bug
        message = Message.objects.get(id=msg_id)
        message.rendered_content = ""
        message.save(update_fields=["rendered_content"])

        self.assert_json_success(
            self.client_patch(
                "/json/messages/" + str(msg_id),
                {
                    "content": "We will edit this to also render as empty.",
                },
            )
        )
        # And again tweak to simulate a rendering bug
        message = Message.objects.get(id=msg_id)
        message.rendered_content = ""
        message.save(update_fields=["rendered_content"])

        history = self.client_get("/json/messages/" + str(msg_id) + "/history")
        message_history = orjson.loads(history.content)["message_history"]
        self.assertEqual(message_history[0]["rendered_content"], "")
        self.assertEqual(message_history[1]["rendered_content"], "")
        self.assertEqual(message_history[1]["content_html_diff"], "<div></div>")

    def test_edit_link(self) -> None:
        # Link editing
        self.login("hamlet")
        msg_id_1 = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="Here is a link to [zulip](www.zulip.org).",
        )
        new_content_1 = "Here is a link to [zulip](www.zulipchat.com)."
        result_1 = self.client_patch(
            f"/json/messages/{msg_id_1}",
            {
                "content": new_content_1,
            },
        )
        self.assert_json_success(result_1)

        message_edit_history_1 = self.client_get(f"/json/messages/{msg_id_1}/history")
        json_response_1 = orjson.loads(message_edit_history_1.content)
        message_history_1 = json_response_1["message_history"]

        # Check content of message after edit.
        self.assertEqual(
            message_history_1[0]["rendered_content"],
            '<p>Here is a link to <a href="http://www.zulip.org">zulip</a>.</p>',
        )
        self.assertEqual(
            message_history_1[1]["rendered_content"],
            '<p>Here is a link to <a href="http://www.zulipchat.com">zulip</a>.</p>',
        )
        self.assertEqual(
            message_history_1[1]["content_html_diff"],
            (
                '<div><p>Here is a link to <a href="http://www.zulipchat.com"'
                ">zulip "
                '<span class="highlight_text_inserted"> Link: http://www.zulipchat.com'
                '</span> </a> <span class="highlight_text_inserted">.'
                '</span> <span class="highlight_text_deleted"> Link: http://www.zulip.org .'
                "</span> </p></div>"
            ),
        )

    def test_edit_history_unedited(self) -> None:
        self.login("hamlet")

        msg_id = self.send_stream_message(
            self.example_user("hamlet"),
            "Denmark",
            topic_name="editing",
            content="This message has not been edited.",
        )

        result = self.client_get(f"/json/messages/{msg_id}/history")

        message_history = self.assert_json_success(result)["message_history"]
        self.assert_length(message_history, 1)

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

