import random
import re
from collections.abc import Sequence
from datetime import timedelta
from email.headerregistry import Address
from unittest import mock
from unittest.mock import patch

import lxml.html
import orjson
from django.conf import settings
from django.core import mail
from django.core.mail.message import EmailMultiAlternatives
from django.test import override_settings
from django.utils.timezone import now as timezone_now
from django_stubs_ext import StrPromise

from zerver.actions.create_user import do_create_user
from zerver.actions.message_send import internal_send_private_message
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.email_notifications import (
    MissedMessageData,
    fix_emojis,
    fix_spoilers_in_html,
    handle_missedmessage_emails,
    include_realm_name_in_missedmessage_emails_subject,
    relative_to_full_url,
)
from zerver.lib.emoji import get_emoji_file_name
from zerver.lib.send_email import FromAddress
from zerver.lib.test_classes import ZulipTestCase
from zerver.models import Message, UserMessage, UserProfile, UserTopic
from zerver.models.realm_emoji import get_name_keyed_dict_for_active_realm_emoji
from zerver.models.realms import get_realm
from zerver.models.recipients import get_or_create_direct_message_group
from zerver.models.scheduled_jobs import NotificationTriggers
from zerver.models.streams import get_stream


class TestMessageNotificationEmails(ZulipTestCase):
    def test_read_message(self) -> None:
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")
        self.login("cordelia")
        result = self.client_post(
            "/json/messages",
            {
                "type": "private",
                "content": "Test message",
                "to": orjson.dumps([hamlet.email]).decode(),
            },
        )
        self.assert_json_success(result)
        message = self.get_last_message()

        # The message is marked as read for the sender (Cordelia) by the message send codepath.
        # We obviously should not send notifications to someone for messages they sent themselves.
        with mock.patch(
            "zerver.lib.email_notifications.do_send_missedmessage_events_reply_in_zulip"
        ) as m:
            handle_missedmessage_emails(
                cordelia.id,
                {message.id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)},
            )
        m.assert_not_called()

        # If the notification is processed before Hamlet reads the message, he should get the email.
        with mock.patch(
            "zerver.lib.email_notifications.do_send_missedmessage_events_reply_in_zulip"
        ) as m:
            handle_missedmessage_emails(
                hamlet.id,
                {message.id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)},
            )
        m.assert_called_once()

        # If Hamlet reads the message before receiving the email notification, we should not sent him
        # an email.
        usermessage = UserMessage.objects.get(
            user_profile=hamlet,
            message=message,
        )
        usermessage.flags.read = True
        usermessage.save()
        with mock.patch(
            "zerver.lib.email_notifications.do_send_missedmessage_events_reply_in_zulip"
        ) as m:
            handle_missedmessage_emails(
                hamlet.id,
                {message.id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)},
            )
        m.assert_not_called()

    def test_demo_organization_owner_email_not_set(self) -> None:
        realm = get_realm("zulip")
        realm.demo_organization_scheduled_deletion_date = timezone_now() + timedelta(days=30)
        realm.save()

        # Demo organization owner's don't have an email address set initially
        desdemona = self.example_user("desdemona")
        desdemona.delivery_email = ""
        desdemona.save()

        notification_bot = self.notification_bot(realm)
        internal_send_private_message(
            sender=notification_bot,
            recipient_user=desdemona,
            content="Notification bot message",
        )
        message = self.get_last_message()
        self.assertEqual(message.sender.id, notification_bot.id)
        self.assertEqual(message.content, "Notification bot message")

        with mock.patch(
            "zerver.lib.email_notifications.do_send_missedmessage_events_reply_in_zulip"
        ) as m:
            handle_missedmessage_emails(
                desdemona.id,
                {message.id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)},
            )
        m.assert_not_called()

    def normalize_string(self, s: str | StrPromise) -> str:
        s = s.strip()
        return re.sub(r"\s+", " ", s)

    def _get_tokens(self) -> list[str]:
        return ["mm" + str(random.getrandbits(32)) for _ in range(30)]

    def _test_cases(
        self,
        msg_id: int,
        verify_body_include: list[str],
        email_subject: str,
        verify_html_body: bool = False,
        show_message_content: bool = True,
        verify_body_does_not_include: Sequence[str] = [],
        trigger: str = "",
        mentioned_user_group_id: int | None = None,
    ) -> None:
        hamlet = self.example_user("hamlet")
        tokens = self._get_tokens()
        with patch("zerver.lib.email_mirror.generate_missed_message_token", side_effect=tokens):
            self.handle_missedmessage_emails(
                hamlet.id,
                {
                    msg_id: MissedMessageData(
                        trigger=trigger, mentioned_user_group_id=mentioned_user_group_id
                    )
                },
            )
        if settings.EMAIL_GATEWAY_PATTERN != "":
            reply_to_addresses = [settings.EMAIL_GATEWAY_PATTERN % (t,) for t in tokens]
            reply_to_emails = [
                str(Address(display_name="Zulip", addr_spec=address))
                for address in reply_to_addresses
            ]
        else:
            reply_to_emails = ["noreply@testserver"]
        msg = mail.outbox[0]
        assert isinstance(msg, EmailMultiAlternatives)
        from_email = str(
            Address(display_name="testserver notifications", addr_spec=FromAddress.NOREPLY)
        )
        self.assert_length(mail.outbox, 1)
        self.assertEqual(self.email_envelope_from(msg), settings.NOREPLY_EMAIL_ADDRESS)
        self.assertEqual(self.email_display_from(msg), from_email)
        self.assertEqual(msg.subject, email_subject)
        self.assert_length(msg.reply_to, 1)
        self.assertIn(msg.reply_to[0], reply_to_emails)
        if verify_html_body:
            for text in verify_body_include:
                assert isinstance(msg.alternatives[0][0], str)
                html = self.normalize_string(msg.alternatives[0][0])
                self.assertIn(text, html)
        else:
            for text in verify_body_include:
                self.assertIn(text, self.normalize_string(msg.body))
        for text in verify_body_does_not_include:
            self.assertNotIn(text, self.normalize_string(msg.body))

        self.assertEqual(msg.extra_headers["List-Id"], "Zulip Dev <zulip.testserver>")

    def _realm_name_in_missed_message_email_subject(
        self, realm_name_in_notifications: bool
    ) -> None:
        msg_id = self.send_personal_message(
            self.example_user("othello"),
            self.example_user("hamlet"),
            "Extremely personal message!",
        )
        verify_body_include = ["Extremely personal message!"]
        email_subject = "DMs with Othello, the Moor of Venice"

        if realm_name_in_notifications:
            email_subject = "DMs with Othello, the Moor of Venice [Zulip Dev]"
        self._test_cases(msg_id, verify_body_include, email_subject)

    def _extra_context_in_missed_stream_messages_mention(
        self, show_message_content: bool = True
    ) -> None:
        for i in range(11):
            self.send_stream_message(
                self.example_user("othello"),
                "Denmark",
                content=str(i),
                topic_name="test" if i % 2 == 0 else "TEST",
            )
        self.send_stream_message(self.example_user("othello"), "Denmark", "11", topic_name="test2")
        msg_id = self.send_stream_message(
            self.example_user("othello"), "denmark", "@**King Hamlet**"
        )

        if show_message_content:
            verify_body_include = [
                "Othello, the Moor of Venice: > 1 > 2 > 3 > 4 > 5 > 6 > 7 > 8 > 9 > 10 > @**King Hamlet** -- ",
                "You are receiving this because you were personally mentioned.",
            ]
            email_subject = "#Denmark > test"
            verify_body_does_not_include: list[str] = []
        else:
            # Test in case if message content in missed email message are disabled.
            verify_body_include = [
                "This email does not include message content because you have chosen to ",
                "http://zulip.testserver/help/email-notifications#hide-message-content ",
                "View or reply in Zulip Dev Zulip",
                " Manage email preferences: http://zulip.testserver/#settings/notifications",
            ]

            email_subject = "New messages"
            verify_body_does_not_include = [
                "Denmark > test",
                "Othello, the Moor of Venice",
                "1 2 3 4 5 6 7 8 9 10 @**King Hamlet**",
                "private",
                "group",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
            trigger=NotificationTriggers.MENTION,
        )

    def _extra_context_in_missed_stream_messages_topic_wildcard_mention_in_followed_topic(
        self,
        show_message_content: bool = True,
        *,
        receiver_is_participant: bool,
    ) -> None:
        for i in range(1, 3):
            self.send_stream_message(self.example_user("othello"), "Denmark", content=str(i))
        self.send_stream_message(self.example_user("othello"), "Denmark", "11", topic_name="test2")

        if receiver_is_participant:
            self.send_stream_message(self.example_user("hamlet"), "Denmark", content="hello")

        msg_id = self.send_stream_message(self.example_user("othello"), "Denmark", "@**topic**")
        trigger = NotificationTriggers.TOPIC_WILDCARD_MENTION_IN_FOLLOWED_TOPIC
        if not receiver_is_participant:
            trigger = NotificationTriggers.STREAM_EMAIL

        if show_message_content:
            # If Hamlet (receiver) is not a topic participant, @topic doesn't mention him,
            # so he won't receive added context (previous messages) in the email.
            if receiver_is_participant:
                verify_body_include = [
                    "Othello, the Moor of Venice: > 1 > 2 King Hamlet: > hello Othello, the Moor of Venice: > @**topic** -- ",
                    "You are receiving this because all topic participants were mentioned in #Denmark > test.",
                ]
            else:
                verify_body_include = [
                    "Othello, the Moor of Venice: > @**topic** -- ",
                    "You are receiving this because you have email notifications enabled for #Denmark.",
                ]
            email_subject = "#Denmark > test"
            verify_body_does_not_include: list[str] = []
        else:
            # Test in case if message content in missed email message are disabled.
            verify_body_include = [
                "This email does not include message content because you have chosen to ",
                "http://zulip.testserver/help/email-notifications#hide-message-content ",
                "View or reply in Zulip Dev Zulip",
                " Manage email preferences: http://zulip.testserver/#settings/notifications",
            ]
            email_subject = "New messages"
            verify_body_does_not_include = [
                "Othello, the Moor of Venice",
                "1 2 3 4 5 @**topic**",
                "private",
                "group",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
            trigger=trigger,
        )

    def _extra_context_in_missed_stream_messages_stream_wildcard_mention_in_followed_topic(
        self, show_message_content: bool = True
    ) -> None:
        for i in range(1, 6):
            self.send_stream_message(self.example_user("othello"), "Denmark", content=str(i))
        self.send_stream_message(self.example_user("othello"), "Denmark", "11", topic_name="test2")
        msg_id = self.send_stream_message(self.example_user("othello"), "Denmark", "@**all**")

        if show_message_content:
            verify_body_include = [
                "Othello, the Moor of Venice: > 1 > 2 > 3 > 4 > 5 > @**all** -- ",
                "You are receiving this because you have wildcard mention notifications enabled for topics you follow.",
            ]
            email_subject = "#Denmark > test"
            verify_body_does_not_include: list[str] = []
        else:
            # Test in case if message content in missed email message are disabled.
            verify_body_include = [
                "This email does not include message content because you have chosen to ",
                "http://zulip.testserver/help/email-notifications#hide-message-content ",
                "View or reply in Zulip Dev Zulip",
                " Manage email preferences: http://zulip.testserver/#settings/notifications",
            ]
            email_subject = "New messages"
            verify_body_does_not_include = [
                "Denmark > test",
                "Othello, the Moor of Venice",
                "1 2 3 4 5 @**all**",
                "private",
                "group",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
            trigger=NotificationTriggers.STREAM_WILDCARD_MENTION_IN_FOLLOWED_TOPIC,
        )

    def _extra_context_in_missed_stream_messages_topic_wildcard_mention(
        self,
        show_message_content: bool = True,
        *,
        receiver_is_participant: bool,
    ) -> None:
        for i in range(1, 3):
            self.send_stream_message(self.example_user("othello"), "Denmark", content=str(i))
        self.send_stream_message(self.example_user("othello"), "Denmark", "11", topic_name="test2")

        if receiver_is_participant:
            self.send_stream_message(self.example_user("hamlet"), "Denmark", content="hello")

        msg_id = self.send_stream_message(self.example_user("othello"), "denmark", "@**topic**")
        trigger = NotificationTriggers.TOPIC_WILDCARD_MENTION
        if not receiver_is_participant:
            trigger = NotificationTriggers.STREAM_EMAIL

        if show_message_content:
            # If Hamlet (receiver) is not a topic participant, @topic doesn't mention him,
            # so he won't receive added context (previous messages) in the email.
            if receiver_is_participant:
                verify_body_include = [
                    "Othello, the Moor of Venice: > 1 > 2 King Hamlet: > hello Othello, the Moor of Venice: > @**topic** -- ",
                    "You are receiving this because all topic participants were mentioned in #Denmark > test.",
                ]
            else:
                verify_body_include = [
                    "Othello, the Moor of Venice: > @**topic** -- ",
                    "You are receiving this because you have email notifications enabled for #Denmark.",
                ]
            email_subject = "#Denmark > test"
            verify_body_does_not_include: list[str] = []
        else:
            # Test in case if message content in missed email message are disabled.
            verify_body_include = [
                "This email does not include message content because you have chosen to ",
                "http://zulip.testserver/help/email-notifications#hide-message-content ",
                "View or reply in Zulip Dev Zulip",
                " Manage email preferences: http://zulip.testserver/#settings/notifications",
            ]
            email_subject = "New messages"
            verify_body_does_not_include = [
                "Othello, the Moor of Venice",
                "1 2 3 4 5 @**topic**",
                "private",
                "group",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
            trigger=trigger,
        )

    def _extra_context_in_missed_stream_messages_stream_wildcard_mention(
        self, show_message_content: bool = True
    ) -> None:
        for i in range(1, 6):
            self.send_stream_message(self.example_user("othello"), "Denmark", content=str(i))
        self.send_stream_message(self.example_user("othello"), "Denmark", "11", topic_name="test2")
        msg_id = self.send_stream_message(self.example_user("othello"), "denmark", "@**all**")

        if show_message_content:
            verify_body_include = [
                "Othello, the Moor of Venice: > 1 > 2 > 3 > 4 > 5 > @**all** -- ",
                "You are receiving this because everyone was mentioned in #Denmark.",
            ]
            email_subject = "#Denmark > test"
            verify_body_does_not_include: list[str] = []
        else:
            # Test in case if message content in missed email message are disabled.
            verify_body_include = [
                "This email does not include message content because you have chosen to ",
                "http://zulip.testserver/help/email-notifications#hide-message-content ",
                "View or reply in Zulip Dev Zulip",
                " Manage email preferences: http://zulip.testserver/#settings/notifications",
            ]
            email_subject = "New messages"
            verify_body_does_not_include = [
                "Denmark > test",
                "Othello, the Moor of Venice",
                "1 2 3 4 5 @**all**",
                "private",
                "group",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
            trigger=NotificationTriggers.STREAM_WILDCARD_MENTION,
        )

    def _extra_context_in_missed_stream_messages_email_notify(self) -> None:
        for i in range(11):
            self.send_stream_message(self.example_user("othello"), "Denmark", content=str(i))
        self.send_stream_message(self.example_user("othello"), "Denmark", "11", topic_name="test2")
        msg_id = self.send_stream_message(self.example_user("othello"), "denmark", "12")
        verify_body_include = [
            "Othello, the Moor of Venice: > 12 -- ",
            "You are receiving this because you have email notifications enabled for #Denmark.",
        ]
        email_subject = "#Denmark > test"
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            trigger=NotificationTriggers.STREAM_EMAIL,
        )

    def _extra_context_in_missed_stream_messages_mention_two_senders(
        self,
    ) -> None:
        cordelia = self.example_user("cordelia")
        self.subscribe(cordelia, "Denmark")

        for i in range(3):
            self.send_stream_message(cordelia, "Denmark", str(i))
        msg_id = self.send_stream_message(
            self.example_user("othello"), "Denmark", "@**King Hamlet**"
        )
        verify_body_include = [
            "Cordelia, Lear's daughter: > 0 > 1 > 2 Othello, the Moor of Venice: > @**King Hamlet** -- ",
            "You are receiving this because you were personally mentioned.",
        ]
        email_subject = "#Denmark > test"
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            trigger=NotificationTriggers.MENTION,
        )

    def _resolved_topic_missed_stream_messages_thread_friendly(self) -> None:
        topic_name = "threading and so forth"
        othello_user = self.example_user("othello")
        msg_id = -1
        for i in range(3):
            msg_id = self.send_stream_message(
                othello_user,
                "Denmark",
                content=str(i),
                topic_name=topic_name,
            )

        self.assert_json_success(self.resolve_topic_containing_message(othello_user, msg_id))

        verify_body_include = [
            "Othello, the Moor of Venice: > 2 -- ",
            "You are receiving this because you have email notifications enabled for #Denmark.",
        ]
        email_subject = "[resolved] #Denmark > threading and so forth"
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            trigger=NotificationTriggers.STREAM_EMAIL,
        )

    def _extra_context_in_missed_personal_messages(
        self,
        show_message_content: bool = True,
        message_content_disabled_by_user: bool = False,
        message_content_disabled_by_realm: bool = False,
    ) -> None:
        msg_id = self.send_personal_message(
            self.example_user("othello"),
            self.example_user("hamlet"),
            "Extremely personal message!",
        )

        if show_message_content:
            verify_body_include = ["> Extremely personal message!"]
            email_subject = "DMs with Othello, the Moor of Venice"
            verify_body_does_not_include: list[str] = []
        else:
            if message_content_disabled_by_realm:
                verify_body_include = [
                    "This email does not include message content because your organization",
                    "http://zulip.testserver/help/hide-message-content-in-emails",
                    "View or reply in Zulip Dev Zulip",
                    " Manage email preferences: http://zulip.testserver/#settings/notifications",
                ]
            elif message_content_disabled_by_user:
                verify_body_include = [
                    "This email does not include message content because you have chosen to ",
                    "http://zulip.testserver/help/email-notifications#hide-message-content ",
                    "View or reply in Zulip Dev Zulip",
                    " Manage email preferences: http://zulip.testserver/#settings/notifications",
                ]
            email_subject = "New messages"
            verify_body_does_not_include = [
                "Othello, the Moor of Venice",
                "Extremely personal message!",
                "mentioned",
                "group",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
        )

    def _reply_to_email_in_missed_personal_messages(self) -> None:
        msg_id = self.send_personal_message(
            self.example_user("othello"),
            self.example_user("hamlet"),
            "Extremely personal message!",
        )
        verify_body_include = ["Reply to this email directly, or view it in Zulip Dev Zulip"]
        email_subject = "DMs with Othello, the Moor of Venice"
        self._test_cases(msg_id, verify_body_include, email_subject)

    def _reply_warning_in_missed_personal_messages(self) -> None:
        msg_id = self.send_personal_message(
            self.example_user("othello"),
            self.example_user("hamlet"),
            "Extremely personal message!",
        )
        verify_body_include = ["Do not reply to this email."]
        email_subject = "DMs with Othello, the Moor of Venice"
        self._test_cases(msg_id, verify_body_include, email_subject)

    def _extra_context_in_missed_group_direct_messages_two_others(
        self, show_message_content: bool = True
    ) -> None:
        msg_id = self.send_group_direct_message(
            self.example_user("othello"),
            [
                self.example_user("hamlet"),
                self.example_user("iago"),
            ],
            "Group personal message!",
        )

        if show_message_content:
            verify_body_include = [
                "Othello, the Moor of Venice: > Group personal message! -- Reply"
            ]
            email_subject = "Group DMs with Iago and Othello, the Moor of Venice"
            verify_body_does_not_include: list[str] = []
        else:
            verify_body_include = [
                "This email does not include message content because you have chosen to ",
                "http://zulip.testserver/help/email-notifications#hide-message-content ",
                "View or reply in Zulip Dev Zulip",
                " Manage email preferences: http://zulip.testserver/#settings/notifications",
            ]
            email_subject = "New messages"
            verify_body_does_not_include = [
                "Iago",
                "Othello, the Moor of Venice Othello, the Moor of Venice",
                "Group personal message!",
                "mentioned",
                "Reply to this email directly, or view it in Zulip Dev Zulip",
            ]
        self._test_cases(
            msg_id,
            verify_body_include,
            email_subject,
            show_message_content=show_message_content,
            verify_body_does_not_include=verify_body_does_not_include,
        )

    def _extra_context_in_missed_group_direct_messages_three_others(self) -> None:
        msg_id = self.send_group_direct_message(
            self.example_user("othello"),
            [
                self.example_user("hamlet"),
                self.example_user("iago"),
                self.example_user("cordelia"),
            ],
            "Group personal message!",
        )

        verify_body_include = ["Othello, the Moor of Venice: > Group personal message! -- Reply"]
        email_subject = (
            "Group DMs with Cordelia, Lear's daughter, Iago, and Othello, the Moor of Venice"
        )
        self._test_cases(msg_id, verify_body_include, email_subject)

    def _extra_context_in_missed_group_direct_messages_many_others(self) -> None:
        msg_id = self.send_group_direct_message(
            self.example_user("othello"),
            [
                self.example_user("hamlet"),
                self.example_user("iago"),
                self.example_user("cordelia"),
                self.example_user("prospero"),
            ],
            "Group personal message!",
        )

        verify_body_include = ["Othello, the Moor of Venice: > Group personal message! -- Reply"]
        email_subject = "Group DMs with Cordelia, Lear's daughter, Iago, and 2 others"
        self._test_cases(msg_id, verify_body_include, email_subject)

    def _deleted_message_in_missed_stream_messages(self) -> None:
        msg_id = self.send_stream_message(
            self.example_user("othello"), "denmark", "@**King Hamlet** to be deleted"
        )

        hamlet = self.example_user("hamlet")
        self.login("othello")
        result = self.client_patch("/json/messages/" + str(msg_id), {"content": " "})
        self.assert_json_success(result)
        handle_missedmessage_emails(
            hamlet.id, {msg_id: MissedMessageData(trigger=NotificationTriggers.MENTION)}
        )
        self.assert_length(mail.outbox, 0)

    def _deleted_message_in_missed_personal_messages(self) -> None:
        msg_id = self.send_personal_message(
            self.example_user("othello"),
            self.example_user("hamlet"),
            "Extremely personal message! to be deleted!",
        )

        hamlet = self.example_user("hamlet")
        self.login("othello")
        result = self.client_patch("/json/messages/" + str(msg_id), {"content": " "})
        self.assert_json_success(result)
        handle_missedmessage_emails(
            hamlet.id, {msg_id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)}
        )
        self.assert_length(mail.outbox, 0)

    def _deleted_message_in_missed_group_direct_messages(self) -> None:
        msg_id = self.send_group_direct_message(
            self.example_user("othello"),
            [
                self.example_user("hamlet"),
                self.example_user("iago"),
            ],
            "Group personal message!",
        )

        hamlet = self.example_user("hamlet")
        iago = self.example_user("iago")
        self.login("othello")
        result = self.client_patch("/json/messages/" + str(msg_id), {"content": " "})
        self.assert_json_success(result)
        handle_missedmessage_emails(
            hamlet.id, {msg_id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)}
        )
        self.assert_length(mail.outbox, 0)
        handle_missedmessage_emails(
            iago.id, {msg_id: MissedMessageData(trigger=NotificationTriggers.DIRECT_MESSAGE)}
        )
        self.assert_length(mail.outbox, 0)

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

        hamlet_only = check_add_user_group(
            get_realm("zulip"), "hamlet_only", [hamlet], acting_user=hamlet
        )
        hamlet_and_cordelia = check_add_user_group(
            get_realm("zulip"), "hamlet_and_cordelia", [hamlet, cordelia], acting_user=hamlet
        )

        hamlet_only_message_id = self.send_stream_message(othello, "Denmark", "@*hamlet_only*")
        hamlet_and_cordelia_message_id = self.send_stream_message(
            othello, "Denmark", "@*hamlet_and_cordelia*"
        )

        self.handle_missedmessage_emails(
            hamlet.id,
            {
                hamlet_only_message_id: MissedMessageData(
                    trigger=NotificationTriggers.MENTION, mentioned_user_group_id=hamlet_only.id
                ),
                hamlet_and_cordelia_message_id: MissedMessageData(
                    trigger=NotificationTriggers.MENTION,
                    mentioned_user_group_id=hamlet_and_cordelia.id,
                ),
            },
        )

        expected_email_include = [
            "Othello, the Moor of Venice: > @*hamlet_only* > @*hamlet_and_cordelia* -- ",
            "You are receiving this because @hamlet_only was mentioned.",
        ]

        for text in expected_email_include:
            self.assertIn(text, self.normalize_string(mail.outbox[0].body))

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

        hamlet_and_cordelia = check_add_user_group(
            get_realm("zulip"), "hamlet_and_cordelia", [hamlet, cordelia], acting_user=hamlet
        )

        user_group_mentioned_message_id = self.send_stream_message(
            othello, "Denmark", "@*hamlet_and_cordelia*"
        )
        personal_mentioned_message_id = self.send_stream_message(
            othello, "Denmark", "@**King Hamlet**"
