import asyncio
import base64
import email.parser
import email.policy
import os
from contextlib import suppress
from datetime import timedelta
from email.headerregistry import Address
from email.message import EmailMessage, MIMEPart
from smtplib import SMTPException, SMTPSenderRefused
from unittest import mock

import orjson
import time_machine
from aiosmtpd.smtp import SMTP
from asgiref.sync import sync_to_async
from django.conf import settings
from django.core import mail
from django.core.mail.backends.locmem import EmailBackend
from django.test import override_settings
from django.utils.timezone import now as timezone_now

from zerver.actions.realm_settings import do_deactivate_realm
from zerver.actions.streams import do_change_stream_group_based_setting, do_deactivate_stream
from zerver.actions.users import do_deactivate_user
from zerver.lib.email_mirror import (
    RateLimitedRealmMirror,
    create_missed_message_address,
    filter_footer,
    generate_missed_message_token,
    get_missed_message_token_from_address,
    is_forwarded,
    is_missed_message_address,
    log_error,
    process_message,
    process_missed_message,
    redact_email_address,
    strip_from_subject,
)
from zerver.lib.email_mirror_helpers import (
    ZulipEmailForwardError,
    decode_email_address,
    encode_email_address,
    get_channel_email_token,
    get_email_gateway_message_string_from_address,
)
from zerver.lib.email_mirror_server import ZulipMessageHandler, send_to_postmaster
from zerver.lib.email_notifications import convert_html_to_markdown
from zerver.lib.send_email import FromAddress
from zerver.lib.streams import ensure_stream
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import most_recent_message, most_recent_usermessage
from zerver.models import Attachment, Recipient, Stream, UserProfile
from zerver.models.groups import NamedUserGroup, SystemGroups
from zerver.models.messages import Message
from zerver.models.realms import get_realm
from zerver.models.streams import StreamTopicsPolicyEnum, get_stream
from zerver.models.users import get_system_bot

logger_name = "zerver.lib.email_mirror"


class TestEncodeDecode(ZulipTestCase):
    def _assert_options(
        self,
        options: dict[str, bool],
        show_sender: bool = False,
        include_footer: bool = False,
        include_quotes: bool = False,
        prefer_text: bool = True,
    ) -> None:
        self.assertEqual(show_sender, ("show_sender" in options) and options["show_sender"])
        self.assertEqual(
            include_footer, ("include_footer" in options) and options["include_footer"]
        )
        self.assertEqual(
            include_quotes, ("include_quotes" in options) and options["include_quotes"]
        )
        self.assertEqual(prefer_text, options.get("prefer_text", True))

    def test_encode_decode(self) -> None:
        realm = get_realm("zulip")
        stream_name = "dev. help"
        stream = ensure_stream(realm, stream_name, acting_user=None)
        hamlet = self.example_user("hamlet")
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        email_address = encode_email_address(stream.name, email_token)
        self.assertEqual(email_address, f"dev-help.{email_token}@testserver")

        # The default form of the email address (with an option - "include-footer"):
        token, options = decode_email_address(f"dev-help.{email_token}.include-footer@testserver")
        self._assert_options(options, include_footer=True)
        self.assertEqual(token, email_token)

        # Using + instead of . as the separator is also supported for backwards compatibility,
        # since that was the original form of addresses that we used:
        token, options = decode_email_address(f"dev-help+{email_token}+include-footer@testserver")
        self._assert_options(options, include_footer=True)
        self.assertEqual(token, email_token)

        token, options = decode_email_address(email_address)
        self._assert_options(options)
        self.assertEqual(token, email_token)

        # We also handle mixing + and . but it shouldn't be recommended to users.
        email_address_all_options = (
            "dev-help.{}+include-footer.show-sender+include-quotes@testserver"
        )
        email_address_all_options = email_address_all_options.format(email_token)
        token, options = decode_email_address(email_address_all_options)
        self._assert_options(options, show_sender=True, include_footer=True, include_quotes=True)
        self.assertEqual(token, email_token)

        email_address = email_address.replace("@testserver", "@zulip.org")
        email_address_all_options = email_address_all_options.replace("@testserver", "@zulip.org")
        with self.assertRaises(ZulipEmailForwardError):
            decode_email_address(email_address)

        with self.assertRaises(ZulipEmailForwardError):
            decode_email_address(email_address_all_options)

        with self.settings(EMAIL_GATEWAY_EXTRA_PATTERN_HACK="@zulip.org"):
            token, options = decode_email_address(email_address)
            self._assert_options(options)
            self.assertEqual(token, email_token)

            token, options = decode_email_address(email_address_all_options)
            self._assert_options(
                options, show_sender=True, include_footer=True, include_quotes=True
            )
            self.assertEqual(token, email_token)

        with self.assertRaises(ZulipEmailForwardError):
            decode_email_address("bogus")

    # Test stream name encoding changes introduced due to
    # https://github.com/zulip/zulip/issues/9840
    def test_encode_decode_nonlatin_alphabet_stream_name(self) -> None:
        realm = get_realm("zulip")
        stream_name = "Тестовы some ascii letters"
        stream = ensure_stream(realm, stream_name, acting_user=None)
        hamlet = self.example_user("hamlet")
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        email_address = encode_email_address(stream.name, email_token)

        msg_string = get_email_gateway_message_string_from_address(email_address)
        parts = msg_string.split("+")
        # Stream name should be completely stripped to '', so msg_string
        # should only have the email_token in it.
        self.assert_length(parts, 1)

        # Correctly decode the resulting address that doesn't have the stream name:
        token, show_sender = decode_email_address(email_address)
        self.assertFalse(show_sender)
        self.assertEqual(token, email_token)

        asciiable_stream_name = "ąężć"
        stream = ensure_stream(realm, asciiable_stream_name, acting_user=None)
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        email_address = encode_email_address(stream.name, email_token)
        self.assertTrue(email_address.startswith("aezc."))

    def test_decode_ignores_stream_name(self) -> None:
        stream = get_stream("Denmark", get_realm("zulip"))
        hamlet = self.example_user("hamlet")
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        stream_to_address = encode_email_address(stream.name, email_token)
        stream_to_address = stream_to_address.replace("denmark", "Some_name")

        # get the email_token:
        token = decode_email_address(stream_to_address)[0]
        self.assertEqual(token, email_token)

    def test_encode_with_show_sender(self) -> None:
        stream = get_stream("Denmark", get_realm("zulip"))
        hamlet = self.example_user("hamlet")
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        stream_to_address = encode_email_address(stream.name, email_token, show_sender=True)

        token, options = decode_email_address(stream_to_address)
        self._assert_options(options, show_sender=True)
        self.assertEqual(token, email_token)

    def test_decode_prefer_text_options(self) -> None:
        stream = get_stream("Denmark", get_realm("zulip"))
        hamlet = self.example_user("hamlet")
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        encode_email_address(stream.name, email_token)
        address_prefer_text = f"Denmark.{email_token}.prefer-text@testserver"
        address_prefer_html = f"Denmark.{email_token}.prefer-html@testserver"

        _token, options = decode_email_address(address_prefer_text)
        self._assert_options(options, prefer_text=True)

        _token, options = decode_email_address(address_prefer_html)
        self._assert_options(options, prefer_text=False)


class TestGetMissedMessageToken(ZulipTestCase):
    def test_get_missed_message_token(self) -> None:
        with self.settings(EMAIL_GATEWAY_PATTERN="%s@example.com"):
            address = "mm" + ("x" * 32) + "@example.com"
            self.assertTrue(is_missed_message_address(address))
            token = get_missed_message_token_from_address(address)
            self.assertEqual(token, "mm" + "x" * 32)

            # This next section was a bug at one point--we'd treat ordinary
            # user addresses that happened to begin with "mm" as being
            # the special mm+32chars tokens.
            address = "mmathers@example.com"
            self.assertFalse(is_missed_message_address(address))
            with self.assertRaises(ZulipEmailForwardError):
                get_missed_message_token_from_address(address)

            # Now test the case where we our address does not match the
            # EMAIL_GATEWAY_PATTERN.
            # This used to crash in an ugly way; we want to throw a proper
            # exception.
            address = "alice@not-the-domain-we-were-expecting.com"
            self.assertFalse(is_missed_message_address(address))
            with self.assertRaises(ZulipEmailForwardError):
                get_missed_message_token_from_address(address)


class TestFilterFooter(ZulipTestCase):
    def test_filter_footer(self) -> None:
        text = """Test message
        --Not a delimiter--
        More message
        --
        Footer"""
        expected_output = """Test message
        --Not a delimiter--
        More message"""
        result = filter_footer(text)
        self.assertEqual(result, expected_output)

    def test_filter_footer_many_parts(self) -> None:
        text = """Test message
        --
        Part1
        --
        Part2"""
        result = filter_footer(text)
        # Multiple possible footers, don't strip
        self.assertEqual(result, text)


class TestStreamEmailMessages(ZulipTestCase):
    def create_incoming_valid_message(
        self, msgtext: str, stream: Stream, include_quotes: bool
    ) -> EmailMessage:
        hamlet = self.example_user("hamlet")
        email_token = get_channel_email_token(stream, creator=hamlet, sender=hamlet)
        address = Address(addr_spec=encode_email_address(stream.name, email_token))
        email_username = address.username + "+show-sender"
        if include_quotes:
            email_username += "+include-quotes"
        stream_to_address = Address(username=email_username, domain=address.domain).addr_spec

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content(msgtext)
        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")
        return incoming_valid_message

    def test_receive_stream_email_messages_success(self) -> None:
        # build dummy messages for stream
        # test valid incoming stream message is processed properly
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_address = encode_email_address(stream.name, email_token)

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        # Hamlet is subscribed to this stream so should see the email message from Othello.
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, "TestStreamEmailMessages body")
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    # Test receiving an email with the address on an UnstructuredHeader
    # (e.g. Envelope-To) instead of an AddressHeader (e.g. To).
    # https://github.com/zulip/zulip/issues/15864
    def test_receive_stream_email_messages_other_header_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_address = encode_email_address(stream.name, email_token)

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        # Simulate a mailing list
        incoming_valid_message["To"] = "foo-mailinglist@example.com"
        incoming_valid_message["Envelope-To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        # Hamlet is subscribed to this stream so should see the email message from Othello.
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, "TestStreamEmailMessages body")
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    def test_receive_stream_email_messages_blank_subject_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_address = encode_email_address(stream.name, email_token)

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = ""
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        # Hamlet is subscribed to this stream so should see the email message from Othello.
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, "TestStreamEmailMessages body")
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), "Email with no subject")

    def test_receive_stream_email_messages_subject_with_nonprintable_chars(
        self,
    ) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_address = encode_email_address(stream.name, email_token)

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = "Test \u0000 subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        message = most_recent_message(user_profile)

        self.assertEqual(message.topic_name(), "Test  subject")

        # Now check that a subject that will be stripped to the empty string
        # is handled correctly.
        incoming_valid_message.replace_header("Subject", "\u0000")
        process_message(incoming_valid_message)
        message = most_recent_message(user_profile)

        self.assertEqual(message.topic_name(), "Email with no subject")

    def test_receive_stream_email_messages_subject_channel_no_topics(
        self,
    ) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)
        stream.topics_policy = StreamTopicsPolicyEnum.empty_topic_only.value
        stream.save()

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_address = encode_email_address(stream.name, email_token)

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = "Test subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        message = most_recent_message(user_profile)

        self.assertEqual(message.topic_name(), "")
        self.assertEqual(
            message.content, "**Subject:** Test subject\n\nTestStreamEmailMessages body"
        )

    def test_receive_private_stream_email_messages_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.make_stream("private_stream", invite_only=True)
        self.subscribe(user_profile, "private_stream")
        stream = get_stream("private_stream", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_address = encode_email_address(stream.name, email_token)

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        # Hamlet is subscribed to this stream so should see the email message from Othello.
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, "TestStreamEmailMessages body")
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    def test_receive_stream_email_multiple_recipient_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        # stream address is angle-addr within multiple addresses
        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        stream_to_addresses = [
            "A.N. Other <another@example.org>",
            f"Denmark <{encode_email_address(stream.name, email_token)}>",
        ]

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")

        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = ", ".join(stream_to_addresses)
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)

        # Hamlet is subscribed to this stream so should see the email message from Othello.
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, "TestStreamEmailMessages body")
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    def test_receive_stream_email_deactivated_stream(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        msgtext = "TestStreamEmailMessages Body"
        incoming_valid_message = self.create_incoming_valid_message(
            msgtext, stream, include_quotes=False
        )

        do_deactivate_stream(stream, acting_user=None)
        last_message_id = Message.objects.latest("id").id

        with self.assertLogs(logger_name, level="INFO") as m:
            process_message(incoming_valid_message)
        self.assertEqual(
            m.output,
            [
                f"INFO:{logger_name}:Failed to process email to {stream.name} ({stream.realm.string_id}): "
                f"Not authorized to send to channel '{stream.name}'",
            ],
        )
        self.assertEqual(Message.objects.latest("id").id, last_message_id)

    def test_receive_stream_email_show_sender_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        msgtext = "TestStreamEmailMessages Body"
        incoming_valid_message = self.create_incoming_valid_message(
            msgtext, stream, include_quotes=False
        )
        process_message(incoming_valid_message)
        message = most_recent_message(user_profile)

        self.assertEqual(
            message.content,
            "**From:** {}\n\n{}".format(self.example_email("hamlet"), msgtext),
        )
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    def test_receive_stream_email_forwarded_success(self) -> None:
        msgtext = """
Hello! Here is a message I am forwarding to this list.
I hope you enjoy reading it!
-Glen

From: John Doe johndoe@wherever
To: A Zulip-subscribed mailing list somelist@elsewhere
Subject: Some subject

Here is the original email. It is full of text
and other things
-John
"""
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        def send_and_check_contents(
            msgtext: str, stream: Stream, include_quotes: bool, expected_body: str
        ) -> None:
            incoming_valid_message = self.create_incoming_valid_message(
                msgtext, stream, include_quotes
            )
            process_message(incoming_valid_message)
            message = most_recent_message(user_profile)
            expected = "**From:** {}\n\n{}".format(self.example_email("hamlet"), expected_body)
            self.assertEqual(message.content, expected.strip())
            self.assert_message_stream_name(message, stream.name)
            self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

        # include_quotes=True: expect the From:... to be preserved
        send_and_check_contents(msgtext, stream, include_quotes=True, expected_body=msgtext)

        # include_quotes=False: expect the From:... to be stripped
        send_and_check_contents(
            msgtext,
            stream,
            include_quotes=False,
            expected_body="Hello! Here is a message I am forwarding to this list.\nI hope you enjoy reading it!\n-Glen",
        )

    def test_receive_stream_email_show_sender_utf8_encoded_sender(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        address = Address(addr_spec=encode_email_address(stream.name, email_token))
        email_username = address.username + "+show-sender"
        stream_to_address = Address(username=email_username, domain=address.domain).addr_spec

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("TestStreamEmailMessages body")
        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = (
            "Test =?utf-8?b?VXNlcsOzxIXEmQ==?= <=?utf-8?q?hamlet=5F=C4=99?=@zulip.com>"
        )
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)
        message = most_recent_message(user_profile)

        self.assertEqual(
            message.content,
            "**From:** {}\n\n{}".format(
                "Test Useróąę <hamlet_ę@zulip.com>", "TestStreamEmailMessages body"
            ),
        )
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    def test_receive_stream_email_include_footer_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        address = Address(addr_spec=encode_email_address(stream.name, email_token))
        email_username = address.username + "+include-footer"
        stream_to_address = Address(username=email_username, domain=address.domain).addr_spec

        text = """Test message
        --
        Footer"""

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content(text)
        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, text)
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])

    def test_receive_stream_email_include_quotes_success(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.subscribe(user_profile, "Denmark")
        stream = get_stream("Denmark", user_profile.realm)

        email_token = get_channel_email_token(stream, creator=user_profile, sender=user_profile)
        address = Address(addr_spec=encode_email_address(stream.name, email_token))
        email_username = address.username + "+include-quotes"
        stream_to_address = Address(username=email_username, domain=address.domain).addr_spec

        text = """Reply

        -----Original Message-----

        Quote"""

        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content(text)
        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
        incoming_valid_message["From"] = self.example_email("hamlet")
        incoming_valid_message["To"] = stream_to_address
        incoming_valid_message["Reply-to"] = self.example_email("othello")

        process_message(incoming_valid_message)
        message = most_recent_message(user_profile)

        self.assertEqual(message.content, text)
        self.assert_message_stream_name(message, stream.name)
        self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])


class TestChannelEmailMessagesPermissions(ZulipTestCase):
    def create_incoming_valid_message(self, channel_email_address: str) -> EmailMessage:
        incoming_valid_message = EmailMessage()
        incoming_valid_message.set_content("message body")
        incoming_valid_message["Subject"] = "test subject"
        incoming_valid_message["To"] = channel_email_address
        return incoming_valid_message

    def test_valid_sender_id(self) -> None:
        hamlet = self.example_user("hamlet")
        realm = get_realm("zulip")
        channel = get_stream("Denmark", realm)
        self.login("hamlet")

        # Sender is the current user itself.
        result = self.client_get(
            f"/json/streams/{channel.id}/email_address", {"sender_id": hamlet.id}
        )
        self.assert_json_success(result)

        # Sender is the Email gateway bot.
        email_gateway_bot = get_system_bot(settings.EMAIL_GATEWAY_BOT, realm.id)
        result = self.client_get(
            f"/json/streams/{channel.id}/email_address", {"sender_id": email_gateway_bot.id}
        )
        self.assert_json_success(result)

        # Sender is a bot owned by the current user.
        bot = self.create_test_bot("test2", hamlet, full_name="Test bot")
        assert bot.bot_owner is not None
        self.assertEqual(bot.bot_owner.id, hamlet.id)
        result = self.client_get(f"/json/streams/{channel.id}/email_address", {"sender_id": bot.id})
        self.assert_json_success(result)

        # Sender is a random user ID. (None of the above three cases)
        othello = self.example_user("othello")
        result = self.client_get(
            f"/json/streams/{channel.id}/email_address", {"sender_id": othello.id}
        )
        self.assert_json_error(result, "No such bot")

    def test_creator_with_send_message_permission(self) -> None:
        hamlet = self.example_user("hamlet")
        realm = get_realm("zulip")
        channel = get_stream("Denmark", realm)

        self.set_user_role(hamlet, UserProfile.ROLE_MODERATOR)
        moderators_group = NamedUserGroup.objects.get(
            name=SystemGroups.MODERATORS, realm_for_sharding=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            channel, "can_send_message_group", moderators_group, acting_user=hamlet
        )

        # Sender is the current user itself.
        email_token = get_channel_email_token(channel, creator=hamlet, sender=hamlet)
        channel_email_address = encode_email_address(channel.name, email_token)
        incoming_valid_message = self.create_incoming_valid_message(channel_email_address)

        with self.assertLogs(logger_name, level="INFO") as m:
            process_message(incoming_valid_message)
        self.assertEqual(
            m.output,
            [
                f"INFO:{logger_name}:Successfully processed email to {channel.name} ({realm.string_id})"
            ],
        )

        # Sender is the Email gateway bot.
        email_gateway_bot = get_system_bot(settings.EMAIL_GATEWAY_BOT, realm.id)
        email_token = get_channel_email_token(channel, creator=hamlet, sender=email_gateway_bot)
        channel_email_address = encode_email_address(channel.name, email_token)
        incoming_valid_message = self.create_incoming_valid_message(channel_email_address)

        with self.assertLogs(logger_name, level="INFO") as m:
            process_message(incoming_valid_message)
        self.assertEqual(
            m.output,
            [
                f"INFO:{logger_name}:Successfully processed email to {channel.name} ({realm.string_id})"
            ],
        )

