from datetime import timedelta
from email.headerregistry import Address
from typing import Any
from unittest import mock

import orjson
from django.conf import settings
from django.db.models import Q
from django.test import override_settings
from django.utils.timezone import now as timezone_now

from zerver.actions.create_realm import do_create_realm
from zerver.actions.create_user import do_create_user
from zerver.actions.message_delete import do_delete_messages
from zerver.actions.message_send import (
    build_message_send_dict,
    check_message,
    check_send_stream_message,
    do_send_messages,
    extract_private_recipients,
    extract_stream_indicator,
    internal_prep_private_message,
    internal_prep_stream_message_by_name,
    internal_send_group_direct_message,
    internal_send_private_message,
    internal_send_stream_message,
    internal_send_stream_message_by_name,
    send_rate_limited_pm_notification_to_bot_owner,
)
from zerver.actions.realm_settings import (
    do_change_realm_permission_group_setting,
    do_set_realm_property,
)
from zerver.actions.streams import (
    do_change_stream_group_based_setting,
    do_change_stream_permission,
    do_deactivate_stream,
    do_set_stream_property,
)
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.users import do_change_can_forge_sender, do_deactivate_user
from zerver.lib.addressee import Addressee
from zerver.lib.exceptions import (
    DirectMessageInitiationError,
    DirectMessagePermissionError,
    JsonableError,
    MessagesNotAllowedInEmptyTopicError,
    TopicsNotAllowedError,
)
from zerver.lib.message import get_raw_unread_data, get_recent_private_conversations
from zerver.lib.message_cache import MessageDict
from zerver.lib.per_request_cache import flush_per_request_caches
from zerver.lib.stream_subscription import create_stream_subscription
from zerver.lib.streams import create_stream_if_needed
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    get_user_messages,
    make_client,
    message_stream_count,
    most_recent_message,
    most_recent_usermessage,
    reset_email_visibility_to_everyone_in_zulip_realm,
)
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.lib.types import UserGroupMembersData
from zerver.models import (
    Message,
    NamedUserGroup,
    Realm,
    RealmDomain,
    Recipient,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
)
from zerver.models.constants import MAX_TOPIC_NAME_LENGTH
from zerver.models.groups import SystemGroups
from zerver.models.realms import RealmTopicsPolicyEnum, get_realm
from zerver.models.recipients import get_direct_message_group, get_or_create_direct_message_group
from zerver.models.streams import StreamTopicsPolicyEnum, get_stream
from zerver.models.users import (
    get_system_bot,
    get_user,
    get_user_by_delivery_email,
    is_cross_realm_bot_email,
)
from zerver.views.message_send import InvalidMirrorInputError


class MessagePOSTTest(ZulipTestCase):
    def _send_and_verify_message(
        self,
        user: UserProfile,
        stream_name: str,
        error_msg: str | None = None,
        *,
        allow_unsubscribed_sender: bool = False,
        topic_name: str = "test",
    ) -> None:
        if error_msg is None:
            msg_id = self.send_stream_message(
                user,
                stream_name,
                allow_unsubscribed_sender=allow_unsubscribed_sender,
                topic_name=topic_name,
            )
            result = self.api_get(user, "/api/v1/messages/" + str(msg_id))
            self.assert_json_success(result)
        else:
            with self.assertRaisesRegex(JsonableError, error_msg):
                self.send_stream_message(
                    user, stream_name, allow_unsubscribed_sender=allow_unsubscribed_sender
                )

    def test_message_to_stream_by_name(self) -> None:
        """
        Sending a message to a stream to which you are subscribed is
        successful.
        """
        recipient_type_name = ["stream", "channel"]
        self.login("hamlet")

        for recipient_type in recipient_type_name:
            result = self.client_post(
                "/json/messages",
                {
                    "type": recipient_type,
                    "to": orjson.dumps("Verona").decode(),
                    "content": "Test message",
                    "topic": "Test topic",
                },
            )
            self.assert_json_success(result)

    def test_api_message_to_stream_by_name(self) -> None:
        """
        Same as above, but for the API view
        """
        recipient_type_name = ["stream", "channel"]
        user = self.example_user("hamlet")

        for recipient_type in recipient_type_name:
            result = self.api_post(
                user,
                "/api/v1/messages",
                {
                    "type": recipient_type,
                    "to": orjson.dumps("Verona").decode(),
                    "content": "Test message",
                    "topic": "Test topic",
                },
            )
            self.assert_json_success(result)

    def test_message_to_stream_with_nonexistent_id(self) -> None:
        cordelia = self.example_user("cordelia")
        bot = self.create_test_bot(
            short_name="whatever",
            user_profile=cordelia,
        )
        result = self.api_post(
            bot,
            "/api/v1/messages",
            {
                "type": "channel",
                "to": orjson.dumps([99999]).decode(),
                "content": "Stream message by ID.",
                "topic": "Test topic for stream ID message",
            },
        )
        self.assert_json_error(result, "Channel with ID '99999' does not exist")

        msg = self.get_last_message()
        expected = (
            "Your bot `whatever-bot@zulip.testserver` tried to send a message to "
            "channel ID 99999, but there is no channel with that ID."
        )
        self.assertEqual(msg.content, expected)

    def test_message_to_stream_with_no_subscribers(self) -> None:
        """
        Sending a message to an empty stream succeeds, but sends a warning
        to the owner.
        """
        realm = get_realm("zulip")
        cordelia = self.example_user("cordelia")
        bot = self.create_test_bot(
            short_name="whatever",
            user_profile=cordelia,
        )
        stream = create_stream_if_needed(realm, "Acropolis")[0]
        result = self.api_post(
            bot,
            "/api/v1/messages",
            {
                "type": "channel",
                "to": orjson.dumps(stream.name).decode(),
                "content": "Stream message to an empty stream by name.",
                "topic": "Test topic for empty stream name message",
            },
        )
        self.assert_json_success(result)

        msg = self.get_last_message()
        expected = "Stream message to an empty stream by name."
        self.assertEqual(msg.content, expected)

        msg = self.get_second_to_last_message()
        expected = (
            "Your bot `whatever-bot@zulip.testserver` tried to send a message to "
            "channel #**Acropolis**. The channel exists but does not have any subscribers."
        )
        self.assertEqual(msg.content, expected)

    def test_message_to_stream_with_no_subscribers_by_id(self) -> None:
        """
        Sending a message to an empty stream succeeds, but sends a warning
        to the owner.
        """
        realm = get_realm("zulip")
        cordelia = self.example_user("cordelia")
        bot = self.create_test_bot(
            short_name="whatever",
            user_profile=cordelia,
        )
        stream = create_stream_if_needed(realm, "Acropolis")[0]
        result = self.api_post(
            bot,
            "/api/v1/messages",
            {
                "type": "channel",
                "to": orjson.dumps([stream.id]).decode(),
                "content": "Stream message to an empty stream by id.",
                "topic": "Test topic for empty stream id message",
            },
        )
        self.assert_json_success(result)

        msg = self.get_last_message()
        expected = "Stream message to an empty stream by id."
        self.assertEqual(msg.content, expected)

        msg = self.get_second_to_last_message()
        expected = (
            "Your bot `whatever-bot@zulip.testserver` tried to send a message to "
            "channel #**Acropolis**. The channel exists but does not have any subscribers."
        )
        self.assertEqual(msg.content, expected)

    def test_message_to_stream_by_id(self) -> None:
        """
        Sending a message to a stream (by stream ID) to which you are
        subscribed is successful.
        """
        recipient_type_name = ["stream", "channel"]
        self.login("hamlet")
        realm = get_realm("zulip")
        stream = get_stream("Verona", realm)

        for recipient_type in recipient_type_name:
            content = f"Stream message by ID, type parameter: {recipient_type}."
            result = self.client_post(
                "/json/messages",
                {
                    "type": recipient_type,
                    "to": orjson.dumps([stream.id]).decode(),
                    "content": content,
                    "topic": "Test topic for stream ID message",
                },
            )
            self.assert_json_success(result)
            sent_message = self.get_last_message()
            self.assertEqual(sent_message.content, content)

    def test_can_send_message_group_permission(self) -> None:
        realm = get_realm("zulip")

        desdemona = self.example_user("desdemona")
        iago = self.example_user("iago")
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")
        othello = self.example_user("othello")
        polonius = self.example_user("polonius")

        desdemona_owned_bot = self.create_test_bot(
            short_name="whatever1",
            full_name="whatever1",
            user_profile=desdemona,
        )
        iago_owned_bot = self.create_test_bot(
            short_name="whatever2",
            full_name="whatever2",
            user_profile=iago,
        )
        cordelia_owned_bot = self.create_test_bot(
            short_name="whatever3",
            full_name="whatever3",
            user_profile=cordelia,
        )
        othello_owned_bot = self.create_test_bot(
            short_name="whatever4",
            full_name="whatever4",
            user_profile=othello,
        )
        notification_bot = get_system_bot("notification-bot@zulip.com", realm.id)

        bot_without_owner = do_create_user(
            email="free-bot@zulip.testserver",
            password="",
            realm=realm,
            full_name="freebot",
            bot_type=UserProfile.DEFAULT_BOT,
            acting_user=None,
        )

        stream_name = "Verona"
        stream = get_stream(stream_name, realm)

        nobody_group = NamedUserGroup.objects.get(
            name=SystemGroups.NOBODY, realm_for_sharding=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", nobody_group, acting_user=iago
        )

        self._send_and_verify_message(
            desdemona, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            desdemona_owned_bot, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            bot_without_owner, stream_name, "You do not have permission to post in this channel."
        )

        # Cross realm bots should be allowed
        internal_send_stream_message(
            notification_bot, stream, "Test topic", "Test message by notification bot"
        )
        self.assertEqual(self.get_last_message().content, "Test message by notification bot")

        owners_group = NamedUserGroup.objects.get(
            name=SystemGroups.OWNERS, realm_for_sharding=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", owners_group, acting_user=iago
        )

        self._send_and_verify_message(
            iago, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            iago_owned_bot, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            bot_without_owner, stream_name, "You do not have permission to post in this channel."
        )

        self._send_and_verify_message(desdemona, stream_name)
        self._send_and_verify_message(desdemona_owned_bot, stream_name)

        # Cross realm bots should be allowed
        internal_send_stream_message(
            notification_bot, stream, "Test topic", "Test message by notification bot"
        )
        self.assertEqual(self.get_last_message().content, "Test message by notification bot")

        hamletcharacters_group = NamedUserGroup.objects.get(
            name="hamletcharacters", realm_for_sharding=realm
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", hamletcharacters_group, acting_user=iago
        )

        self._send_and_verify_message(
            desdemona, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            desdemona_owned_bot, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            iago, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            iago_owned_bot, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            bot_without_owner, stream_name, "You do not have permission to post in this channel."
        )

        self._send_and_verify_message(hamlet, stream_name)
        self._send_and_verify_message(cordelia, stream_name)
        self._send_and_verify_message(cordelia_owned_bot, stream_name)

        # Cross realm bots should be allowed
        internal_send_stream_message(
            notification_bot, stream, "Test topic", "Test message by notification bot"
        )
        self.assertEqual(self.get_last_message().content, "Test message by notification bot")

        setting_group_member_dict = UserGroupMembersData(
            direct_members=[othello.id], direct_subgroups=[owners_group.id]
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", setting_group_member_dict, acting_user=iago
        )

        self._send_and_verify_message(
            iago, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            iago_owned_bot, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            hamlet, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            cordelia, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            cordelia_owned_bot, stream_name, "You do not have permission to post in this channel."
        )
        self._send_and_verify_message(
            bot_without_owner, stream_name, "You do not have permission to post in this channel."
        )

        self._send_and_verify_message(desdemona, stream_name)
        self._send_and_verify_message(desdemona_owned_bot, stream_name)
        self._send_and_verify_message(othello, stream_name)
        self._send_and_verify_message(othello_owned_bot, stream_name)

        # Cross realm bots should be allowed
        internal_send_stream_message(
            notification_bot, stream, "Test topic", "Test message by notification bot"
        )
        self.assertEqual(self.get_last_message().content, "Test message by notification bot")

        everyone_group = NamedUserGroup.objects.get(
            name=SystemGroups.EVERYONE, realm_for_sharding=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", everyone_group, acting_user=iago
        )
        self._send_and_verify_message(othello, stream_name)
        self._send_and_verify_message(othello_owned_bot, stream_name)
        self._send_and_verify_message(iago, stream_name)
        self._send_and_verify_message(iago_owned_bot, stream_name)
        self._send_and_verify_message(polonius, stream_name)
        self._send_and_verify_message(bot_without_owner, stream_name)

        # Cross realm bots should be allowed
        internal_send_stream_message(
            notification_bot, stream, "Test topic", "Test message by notification bot"
        )
        self.assertEqual(self.get_last_message().content, "Test message by notification bot")

    def test_can_send_message_group_permission_for_streams(self) -> None:
        realm = get_realm("zulip")
        othello = self.example_user("othello")

        stream_name = "private_stream"
        stream = self.make_stream(
            stream_name, realm, invite_only=True, history_public_to_subscribers=True
        )
        self._send_and_verify_message(
            othello, stream_name, "Not authorized to send to channel 'private_stream"
        )

        othello_group_member_dict = UserGroupMembersData(
            direct_members=[othello.id], direct_subgroups=[]
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", othello_group_member_dict, acting_user=othello
        )

        self.subscribe(othello, stream_name)
        self._send_and_verify_message(othello, stream_name)
        self.unsubscribe(othello, stream_name)

        do_change_stream_group_based_setting(
            stream, "can_add_subscribers_group", othello_group_member_dict, acting_user=othello
        )
        self._send_and_verify_message(othello, stream_name, allow_unsubscribed_sender=True)

        # history_public_to_subscribers is False
        do_change_stream_permission(
            stream,
            invite_only=True,
            history_public_to_subscribers=False,
            is_web_public=False,
            acting_user=othello,
        )
        self._send_and_verify_message(
            othello,
            stream_name,
            "Not authorized to send to channel 'private_stream",
            allow_unsubscribed_sender=True,
        )

        # invite_only should not matter while sending message, since we
        # might add the ability for guests to join public channels via
        # `can_join_group` in the future and in that case
        # `history_public_to_subscribers` might be a relevant property
        # for public channels
        guest_user = self.example_user("polonius")
        guest_user_group_member_dict = UserGroupMembersData(
            direct_members=[guest_user.id], direct_subgroups=[]
        )
        do_change_stream_group_based_setting(
            stream, "can_send_message_group", guest_user_group_member_dict, acting_user=othello
        )

        do_change_stream_permission(
            stream,
            invite_only=False,
            history_public_to_subscribers=True,
            is_web_public=False,
            acting_user=othello,
        )
        self._send_and_verify_message(
            guest_user,
            stream_name,
            "Not authorized to send to channel 'private_stream",
            allow_unsubscribed_sender=True,
        )
        # Guest not subscribed to a public stream should be able to
        # send a message to that channel if they are part of both a
        # group providing content access and `can_send_message_group`
        # for that channel and `history_public_to_subscribers` is
        # True.
        #
        # But can_add_subscribers_group has !allow_everyone_group.
        do_change_stream_group_based_setting(
            stream, "can_add_subscribers_group", guest_user_group_member_dict, acting_user=othello
        )
        self._send_and_verify_message(
            guest_user,
            stream_name,
            "Not authorized to send to channel 'private_stream",
            allow_unsubscribed_sender=True,
        )

    def test_can_create_topic_group_permission(self) -> None:
        realm = get_realm("zulip")

        desdemona = self.example_user("desdemona")
        iago = self.example_user("iago")
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")
        othello = self.example_user("othello")
        polonius = self.example_user("polonius")

        desdemona_owned_bot = self.create_test_bot(
            short_name="whatever1",
            full_name="whatever1",
            user_profile=desdemona,
        )
        iago_owned_bot = self.create_test_bot(
            short_name="whatever2",
            full_name="whatever2",
            user_profile=iago,
        )
        cordelia_owned_bot = self.create_test_bot(
            short_name="whatever3",
            full_name="whatever3",
            user_profile=cordelia,
        )
        othello_owned_bot = self.create_test_bot(
            short_name="whatever4",
            full_name="whatever4",
            user_profile=othello,
        )
        notification_bot = get_system_bot("notification-bot@zulip.com", realm.id)

        bot_without_owner = do_create_user(
            email="free-bot@zulip.testserver",
            password="",
            realm=realm,
            full_name="freebot",
            bot_type=UserProfile.DEFAULT_BOT,
            acting_user=None,
        )
        can_create_topic_error_msg = (
            "You do not have permission to create new topics in this channel."
        )

        stream_name = "Verona"
        stream = get_stream(stream_name, realm)

        def check_sending_message_for_can_create_topic_group(
            sender: UserProfile,
            topic_name: str = "new_topic",
            expect_fail: bool = False,
        ) -> None:
            if expect_fail:
                self._send_and_verify_message(
                    sender,
                    stream_name,
                    can_create_topic_error_msg,
                    topic_name=topic_name,
                )
                return

            if is_cross_realm_bot_email(sender.delivery_email):
                internal_send_stream_message(
                    notification_bot, stream, topic_name, "Message from notification bot."
                )
                message = self.get_last_message()
                self.assertEqual(message.content, "Message from notification bot.")
            else:
                self._send_and_verify_message(sender, stream_name, topic_name=topic_name)

            sent_message = self.get_last_message()
            # Delete last sent message, so that the topic used for testing
            # new topics is empty.
            do_delete_messages(realm, [sent_message], acting_user=None)

        self._send_and_verify_message(desdemona, stream_name, topic_name="existing topic")

        # Set topic creation settings to nobody group.
        nobody_group = NamedUserGroup.objects.get(
            name=SystemGroups.NOBODY, realm=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            stream, "can_create_topic_group", nobody_group, acting_user=iago
        )

        check_sending_message_for_can_create_topic_group(desdemona, expect_fail=True)
        check_sending_message_for_can_create_topic_group(desdemona_owned_bot, expect_fail=True)
        check_sending_message_for_can_create_topic_group(bot_without_owner, expect_fail=True)

        # Cross realm bots should be allowed
        check_sending_message_for_can_create_topic_group(notification_bot)

        # Sending message to an existing topic should be allowed.
        check_sending_message_for_can_create_topic_group(desdemona, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            desdemona_owned_bot, topic_name="existing topic"
        )
        check_sending_message_for_can_create_topic_group(cordelia, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            bot_without_owner, topic_name="existing topic"
        )

        owners_group = NamedUserGroup.objects.get(
            name=SystemGroups.OWNERS, realm=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            stream, "can_create_topic_group", owners_group, acting_user=iago
        )

        check_sending_message_for_can_create_topic_group(iago, expect_fail=True)
        check_sending_message_for_can_create_topic_group(iago_owned_bot, expect_fail=True)
        check_sending_message_for_can_create_topic_group(bot_without_owner, expect_fail=True)

        check_sending_message_for_can_create_topic_group(desdemona)
        check_sending_message_for_can_create_topic_group(desdemona_owned_bot)

        # Cross realm bots should be allowed
        check_sending_message_for_can_create_topic_group(notification_bot)

        # Sending message to an existing topic should be allowed.
        check_sending_message_for_can_create_topic_group(iago, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            iago_owned_bot, topic_name="existing topic"
        )
        check_sending_message_for_can_create_topic_group(
            bot_without_owner, topic_name="existing topic"
        )

        hamletcharacters_group = NamedUserGroup.objects.get(name="hamletcharacters", realm=realm)
        do_change_stream_group_based_setting(
            stream, "can_create_topic_group", hamletcharacters_group, acting_user=iago
        )

        check_sending_message_for_can_create_topic_group(desdemona, expect_fail=True)
        check_sending_message_for_can_create_topic_group(desdemona_owned_bot, expect_fail=True)
        check_sending_message_for_can_create_topic_group(iago, expect_fail=True)
        check_sending_message_for_can_create_topic_group(iago_owned_bot, expect_fail=True)
        check_sending_message_for_can_create_topic_group(bot_without_owner, expect_fail=True)

        check_sending_message_for_can_create_topic_group(hamlet)
        check_sending_message_for_can_create_topic_group(cordelia)
        check_sending_message_for_can_create_topic_group(cordelia_owned_bot)

        # Cross realm bots should be allowed
        check_sending_message_for_can_create_topic_group(notification_bot)

        # Sending message to an existing topic should be allowed.
        check_sending_message_for_can_create_topic_group(desdemona, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            desdemona_owned_bot, topic_name="existing topic"
        )
        check_sending_message_for_can_create_topic_group(iago, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            iago_owned_bot, topic_name="existing topic"
        )
        check_sending_message_for_can_create_topic_group(
            bot_without_owner, topic_name="existing topic"
        )

        setting_group_member_dict = UserGroupMembersData(
            direct_members=[othello.id], direct_subgroups=[owners_group.id]
        )
        do_change_stream_group_based_setting(
            stream, "can_create_topic_group", setting_group_member_dict, acting_user=iago
        )

        check_sending_message_for_can_create_topic_group(iago, expect_fail=True)
        check_sending_message_for_can_create_topic_group(iago_owned_bot, expect_fail=True)
        check_sending_message_for_can_create_topic_group(hamlet, expect_fail=True)
        check_sending_message_for_can_create_topic_group(cordelia, expect_fail=True)
        check_sending_message_for_can_create_topic_group(cordelia_owned_bot, expect_fail=True)
        check_sending_message_for_can_create_topic_group(bot_without_owner, expect_fail=True)

        check_sending_message_for_can_create_topic_group(desdemona)
        check_sending_message_for_can_create_topic_group(desdemona_owned_bot)
        check_sending_message_for_can_create_topic_group(othello)
        check_sending_message_for_can_create_topic_group(othello_owned_bot)

        # Cross realm bots should be allowed
        check_sending_message_for_can_create_topic_group(notification_bot)

        # Sending message to an existing topic should be allowed.
        check_sending_message_for_can_create_topic_group(iago, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            iago_owned_bot, topic_name="existing topic"
        )
        check_sending_message_for_can_create_topic_group(hamlet, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(cordelia, topic_name="existing topic")
        check_sending_message_for_can_create_topic_group(
            cordelia_owned_bot, topic_name="existing topic"
        )
        check_sending_message_for_can_create_topic_group(
            bot_without_owner, topic_name="existing topic"
        )

        everyone_group = NamedUserGroup.objects.get(
            name=SystemGroups.EVERYONE, realm=realm, is_system_group=True
        )
        do_change_stream_group_based_setting(
            stream, "can_create_topic_group", everyone_group, acting_user=iago
        )
        check_sending_message_for_can_create_topic_group(othello)
        check_sending_message_for_can_create_topic_group(othello_owned_bot)
        check_sending_message_for_can_create_topic_group(iago)
        check_sending_message_for_can_create_topic_group(iago_owned_bot)
        check_sending_message_for_can_create_topic_group(polonius)
        check_sending_message_for_can_create_topic_group(bot_without_owner)

        # Cross realm bots should be allowed
        check_sending_message_for_can_create_topic_group(notification_bot)

    def test_api_message_with_default_to(self) -> None:
        """
        Sending messages without a to field should be sent to the default
        stream for the user_profile.
        """
        user = self.example_user("hamlet")
        user.default_sending_stream_id = get_stream("Verona", user.realm).id
        user.save()
        # The `to` field is required according to OpenAPI specification
        result = self.api_post(
            user,
            "/api/v1/messages",
            {
                "type": "channel",
                "content": "Test message no to",
                "topic": "Test topic",
            },
            intentionally_undocumented=True,
        )
        self.assert_json_success(result)

        sent_message = self.get_last_message()
        self.assertEqual(sent_message.content, "Test message no to")

    def test_message_to_nonexistent_stream(self) -> None:
        """
        Sending a message to a nonexistent stream fails.
        """
        self.login("hamlet")
        self.assertFalse(Stream.objects.filter(name="nonexistent_stream"))
        result = self.client_post(
            "/json/messages",
            {
                "type": "channel",
                "to": orjson.dumps("nonexistent_stream").decode(),
                "content": "Test message",
                "topic": "Test topic",
            },
        )
        self.assert_json_error(result, "Channel 'nonexistent_stream' does not exist")

    def test_message_to_nonexistent_stream_with_bad_characters(self) -> None:
        """
        Nonexistent stream name with bad characters should be escaped properly.
        """
        self.login("hamlet")
        self.assertFalse(Stream.objects.filter(name="""&<"'><non-existent>"""))
        result = self.client_post(
            "/json/messages",
            {
                "type": "channel",
                "to": orjson.dumps("""&<"'><non-existent>""").decode(),
                "content": "Test message",
                "topic": "Test topic",
            },
        )
        self.assert_json_error(
            result, "Channel '&amp;&lt;&quot;&#x27;&gt;&lt;non-existent&gt;' does not exist"
        )

