import time
from collections.abc import Callable, Collection
from typing import Any
from unittest import mock

import orjson
from django.http import HttpRequest, HttpResponse
from typing_extensions import override

from zerver.actions.message_send import internal_send_private_message
from zerver.actions.muted_users import do_mute_user
from zerver.actions.streams import do_change_subscription_property
from zerver.actions.user_groups import 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.cache import cache_delete, get_muting_users_cache_key
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import HostRequestMock, dummy_handler, mock_queue_publish
from zerver.models import PushDevice, Recipient, Subscription, UserProfile, UserTopic
from zerver.models.streams import get_stream
from zerver.tornado.event_queue import (
    ClientDescriptor,
    access_client_descriptor,
    allocate_client_descriptor,
    maybe_enqueue_notifications,
    missedmessage_hook,
    persistent_queue_filename,
)
from zerver.tornado.views import cleanup_event_queue, get_events


class MaybeEnqueueNotificationsTest(ZulipTestCase):
    def test_maybe_enqueue_notifications(self) -> None:
        # We've already tested the "when to send notifications" logic as part of the
        # notification_data module.
        # This test is for verifying whether `maybe_enqueue_notifications` returns the
        # `already_notified` data correctly.
        params = self.get_maybe_enqueue_notifications_parameters(
            message_id=1, user_id=1, acting_user_id=2
        )

        with mock_queue_publish(
            "zerver.tornado.event_queue.queue_json_publish_rollback_unsafe"
        ) as mock_queue_json_publish:
            notified = maybe_enqueue_notifications(**params)
            mock_queue_json_publish.assert_not_called()

        with mock_queue_publish(
            "zerver.tornado.event_queue.queue_json_publish_rollback_unsafe"
        ) as mock_queue_json_publish:
            params["user_notifications_data"] = self.create_user_notifications_data_object(
                user_id=1, dm_push_notify=True, dm_email_notify=True
            )
            notified = maybe_enqueue_notifications(**params)
            self.assertTrue(mock_queue_json_publish.call_count, 2)

            queues_pushed = [entry[0][0] for entry in mock_queue_json_publish.call_args_list]
            self.assertIn("missedmessage_mobile_notifications", queues_pushed)
            self.assertIn("missedmessage_emails", queues_pushed)

            self.assertTrue(notified["email_notified"])
            self.assertTrue(notified["push_notified"])

        with mock_queue_publish(
            "zerver.tornado.event_queue.queue_json_publish_rollback_unsafe"
        ) as mock_queue_json_publish:
            params = self.get_maybe_enqueue_notifications_parameters(
                message_id=1,
                acting_user_id=2,
                user_id=3,
                mention_push_notify=True,
                mention_email_notify=True,
                mentioned_user_group_id=33,
            )
            notified = maybe_enqueue_notifications(**params)
            self.assertTrue(mock_queue_json_publish.call_count, 2)

            push_notice = mock_queue_json_publish.call_args_list[0][0][1]
            self.assertEqual(push_notice["mentioned_user_group_id"], 33)

            email_notice = mock_queue_json_publish.call_args_list[1][0][1]
            self.assertEqual(email_notice["mentioned_user_group_id"], 33)


class StreamWatchersTest(ZulipTestCase):
    def test_stream_watchers(self) -> None:
        """
        We used to have a bug with stream_watchers, where we set their flags to
        None.
        """
        cordelia = self.example_user("cordelia")
        hamlet = self.example_user("hamlet")
        realm = hamlet.realm
        stream_name = "Denmark"

        self.subscribe(cordelia, stream_name)
        self.unsubscribe(hamlet, stream_name)

        queue_data = dict(
            all_public_streams=True,
            apply_markdown=True,
            client_gravatar=True,
            client_type_name="home grown API program",
            event_types=["message"],
            last_connection_time=time.time(),
            queue_timeout=0,
            realm_id=realm.id,
            user_profile_id=hamlet.id,
        )

        client = allocate_client_descriptor(queue_data)

        self.send_stream_message(cordelia, stream_name)

        self.assert_length(client.event_queue.contents(), 1)

        # This next line of code should silently succeed and basically do
        # nothing under the covers.  This test is here to prevent a bug
        # from re-appearing.
        missedmessage_hook(
            user_profile_id=hamlet.id,
            client=client,
            last_for_client=True,
        )


class MissedMessageHookTest(ZulipTestCase):
    """Tests what arguments missedmessage_hook passes into maybe_enqueue_notifications.
    Combined with the previous test, this ensures that the missedmessage_hook is correct"""

    def tornado_call(
        self,
        view_func: Callable[[HttpRequest, UserProfile], HttpResponse],
        user_profile: UserProfile,
        post_data: dict[str, Any],
    ) -> HttpResponse:
        request = HostRequestMock(post_data, user_profile, tornado_handler=dummy_handler)
        return view_func(request, user_profile)

    def allocate_event_queue(self, user: UserProfile) -> ClientDescriptor:
        result = self.tornado_call(
            get_events,
            user,
            {
                "apply_markdown": orjson.dumps(True).decode(),
                "client_gravatar": orjson.dumps(True).decode(),
                "event_types": orjson.dumps(["message"]).decode(),
                "user_client": "website",
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        self.assert_json_success(result)
        queue_id = orjson.loads(result.content)["queue_id"]
        return access_client_descriptor(user.id, queue_id)

    def destroy_event_queue(self, user: UserProfile, queue_id: str) -> None:
        # Ignore this when running mypy, since mypy expects queue_id to be specified when calling the function
        # but it is actually extracted from the request body using the typed_endpoint decorator.
        result = self.tornado_call(cleanup_event_queue, user, {"queue_id": queue_id})  # type: ignore[arg-type]
        self.assert_json_success(result)

    def assert_maybe_enqueue_notifications_call_args(
        self,
        args_dict: Collection[Any],
        message_id: int,
        user_id: int,
        **kwargs: Any,
    ) -> None:
        expected_args_dict = self.get_maybe_enqueue_notifications_parameters(
            user_id=user_id,
            acting_user_id=self.example_user("iago").id,
            message_id=message_id,
            **kwargs,
        )
        self.assertEqual(args_dict, expected_args_dict)

    def change_subscription_properties(self, properties: dict[str, bool]) -> None:
        stream = get_stream("Denmark", self.user_profile.realm)
        sub = Subscription.objects.get(
            user_profile=self.user_profile,
            recipient__type=Recipient.STREAM,
            recipient__type_id=stream.id,
        )
        for property_name, value in properties.items():
            do_change_subscription_property(
                self.user_profile, sub, stream, property_name, value, acting_user=None
            )

    @override
    def setUp(self) -> None:
        super().setUp()
        self.user_profile = self.example_user("hamlet")
        self.cordelia = self.example_user("cordelia")
        do_change_user_setting(
            self.user_profile, "enable_online_push_notifications", False, acting_user=None
        )
        self.register_push_device(self.user_profile.id)
        self.iago = self.example_user("iago")
        self.client_descriptor = self.allocate_event_queue(self.user_profile)
        self.assertTrue(self.client_descriptor.event_queue.empty())

        self.login_user(self.user_profile)

    @override
    def tearDown(self) -> None:
        self.destroy_event_queue(self.user_profile, self.client_descriptor.event_queue.id)
        super().tearDown()

    def test_basic(self) -> None:
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            # To test the missed_message hook, we first need to send a message
            msg_id = self.send_stream_message(self.iago, "Denmark")

            # Verify that nothing happens if you call it as not the
            # "last client descriptor", in which case the function
            # short-circuits, since the `missedmessage_hook` handler
            # for garbage-collection is only for the user's last queue.
            missedmessage_hook(self.user_profile.id, self.client_descriptor, False)
            mock_enqueue.assert_not_called()

            # Now verify that we called the appropriate enqueue function
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_direct_message(self) -> None:
        # By default, email and push notifications should be sent for direct messages
        msg_id = self.send_personal_message(self.iago, self.user_profile)
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                dm_email_notify=True,
                dm_push_notify=True,
                already_notified={"email_notified": True, "push_notified": True},
            )

    def test_enable_offline_email_notifications_setting(self) -> None:
        # When `enable_offline_email_notifications` is off, email notifications
        # should not be sent for direct messages
        do_change_user_setting(
            self.user_profile, "enable_offline_email_notifications", False, acting_user=None
        )
        msg_id = self.send_personal_message(self.iago, self.user_profile)
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                dm_email_notify=False,
                dm_push_notify=True,
                already_notified={"email_notified": False, "push_notified": True},
            )

    def test_mention(self) -> None:
        # By default, email and push notifications should be sent for mentions
        msg_id = self.send_stream_message(
            self.example_user("iago"), "Denmark", content="@**King Hamlet** what's up?"
        )
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                mention_push_notify=True,
                mention_email_notify=True,
                already_notified={"email_notified": True, "push_notified": True},
            )

    def test_enable_offline_push_notifications_setting(self) -> None:
        # When `enable_offline_push_notifications` is off, push notifications should not be sent for mentions
        do_change_user_setting(
            self.user_profile, "enable_offline_push_notifications", False, acting_user=None
        )
        msg_id = self.send_personal_message(self.iago, self.user_profile)
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                dm_email_notify=True,
                dm_push_notify=False,
                already_notified={"email_notified": True, "push_notified": False},
            )

    def test_no_push_device_registered(self) -> None:
        # When `enable_offline_push_notifications` is `true` but no push device registered,
        # push notifications should not be sent.
        do_change_user_setting(
            self.user_profile, "enable_offline_push_notifications", True, acting_user=None
        )
        PushDevice.objects.all().delete()
        msg_id = self.send_personal_message(self.iago, self.user_profile)
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                dm_email_notify=True,
                dm_push_notify=False,
                already_notified={"email_notified": True, "push_notified": False},
            )

    def test_topic_wildcard_mention(self) -> None:
        # By default, topic wildcard mentions should send notifications, just like regular mentions
        self.send_stream_message(self.user_profile, "Denmark")
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**topic** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called()
            args_dict = mock_enqueue.call_args_list[1][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                topic_wildcard_mention_email_notify=True,
                topic_wildcard_mention_push_notify=True,
                already_notified={"email_notified": True, "push_notified": True},
            )

    def test_topic_wildcard_mention_in_muted_stream(self) -> None:
        # Topic wildcard mentions in muted streams don't notify.
        self.change_subscription_properties({"is_muted": True})
        self.send_stream_message(self.user_profile, "Denmark")
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**topic** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called()
            args_dict = mock_enqueue.call_args_list[1][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                topic_wildcard_mention_email_notify=False,
                topic_wildcard_mention_push_notify=False,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_topic_wildcard_mention_in_muted_topic(self) -> None:
        # Topic wildcard mentions in muted topics don't notify.
        do_set_user_topic_visibility_policy(
            self.user_profile,
            get_stream("Denmark", self.user_profile.realm),
            "mutingtest",
            visibility_policy=UserTopic.VisibilityPolicy.MUTED,
        )
        self.send_stream_message(self.user_profile, "Denmark")
        msg_id = self.send_stream_message(
            self.iago, "Denmark", topic_name="mutingtest", content="@**topic** what's up?"
        )
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called()
            args_dict = mock_enqueue.call_args_list[1][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                topic_wildcard_mention_email_notify=False,
                topic_wildcard_mention_push_notify=False,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_stream_wildcard_mention(self) -> None:
        # By default, stream wildcard mentions should send notifications, just like regular mentions
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**all** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_wildcard_mention_email_notify=True,
                stream_wildcard_mention_push_notify=True,
                already_notified={"email_notified": True, "push_notified": True},
            )

    def test_stream_wildcard_mention_in_muted_stream(self) -> None:
        # stream wildcard mentions in muted streams don't notify.
        self.change_subscription_properties({"is_muted": True})
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**all** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                stream_wildcard_mention_email_notify=False,
                stream_wildcard_mention_push_notify=False,
                message_id=msg_id,
                user_id=self.user_profile.id,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_stream_wildcard_mention_in_muted_topic(self) -> None:
        # stream wildcard mentions in muted topics don't notify.
        do_set_user_topic_visibility_policy(
            self.user_profile,
            get_stream("Denmark", self.user_profile.realm),
            "mutingtest",
            visibility_policy=UserTopic.VisibilityPolicy.MUTED,
        )
        msg_id = self.send_stream_message(
            self.iago, "Denmark", topic_name="mutingtest", content="@**all** what's up?"
        )
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                stream_wildcard_mention_email_notify=False,
                stream_wildcard_mention_push_notify=False,
                message_id=msg_id,
                user_id=self.user_profile.id,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_wildcard_mentions_notify_global_setting(self) -> None:
        # With wildcard_mentions_notify=False for a user, we treat the user as not mentioned.
        do_change_user_setting(
            self.user_profile, "wildcard_mentions_notify", False, acting_user=None
        )
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**all** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_wildcard_mention_email_notify=False,
                stream_wildcard_mention_push_notify=False,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_wildcard_mentions_notify_stream_specific_setting(
        self,
    ) -> None:
        # If wildcard_mentions_notify=True for a stream and False for a user, we treat the user
        # as mentioned for that stream.
        do_change_user_setting(
            self.user_profile, "wildcard_mentions_notify", False, acting_user=None
        )
        self.change_subscription_properties({"wildcard_mentions_notify": True})
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**all** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_wildcard_mention_email_notify=True,
                stream_wildcard_mention_push_notify=True,
                already_notified={"email_notified": True, "push_notified": True},
            )

    def test_wildcard_mentions_notify_global_setting_is_a_wrapper(self) -> None:
        # If email notifications for direct messages and mentions themselves have been turned off,
        # even turning on `wildcard_mentions_notify` should not send email notifications
        do_change_user_setting(
            self.user_profile, "enable_offline_email_notifications", False, acting_user=None
        )
        do_change_user_setting(
            self.user_profile, "wildcard_mentions_notify", True, acting_user=None
        )
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**all** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            # We've turned off email notifications for personal mentions, but push notifications
            # for personal mentions are still on.
            # Because `wildcard_mentions_notify` is True, a message with `@all` should follow the
            # personal mention settings
            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_wildcard_mention_email_notify=False,
                stream_wildcard_mention_push_notify=True,
                already_notified={"email_notified": False, "push_notified": True},
            )

    def test_wildcard_mentions_notify_stream_specific_setting_is_a_wrapper(self) -> None:
        # Similar to the above test, but for the stream-specific version of the setting
        self.change_subscription_properties({"wildcard_mentions_notify": True})
        do_change_user_setting(
            self.user_profile, "enable_offline_email_notifications", False, acting_user=None
        )
        msg_id = self.send_stream_message(self.iago, "Denmark", content="@**all** what's up?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_wildcard_mention_email_notify=False,
                stream_wildcard_mention_push_notify=True,
                already_notified={"email_notified": False, "push_notified": True},
            )

    def test_user_group_mention(self) -> None:
        hamlet_and_cordelia = check_add_user_group(
            self.cordelia.realm,
            "hamlet_and_cordelia",
            [self.user_profile, self.cordelia],
            acting_user=self.user_profile,
        )
        msg_id = self.send_stream_message(
            self.iago, "Denmark", content="@*hamlet_and_cordelia* what's up?"
        )
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                mention_push_notify=True,
                mention_email_notify=True,
                mentioned_user_group_id=hamlet_and_cordelia.id,
                already_notified={"email_notified": True, "push_notified": True},
            )

    def test_stream_push_notify_global_setting(self) -> None:
        do_change_user_setting(
            self.user_profile, "enable_stream_push_notifications", True, acting_user=None
        )
        msg_id = self.send_stream_message(self.iago, "Denmark", content="what's up everyone?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_push_notify=True,
                stream_email_notify=False,
                already_notified={"email_notified": False, "push_notified": True},
            )

    def test_stream_email_notify_global_setting(self) -> None:
        do_change_user_setting(
            self.user_profile, "enable_stream_email_notifications", True, acting_user=None
        )
        msg_id = self.send_stream_message(self.iago, "Denmark", content="what's up everyone?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_push_notify=False,
                stream_email_notify=True,
                already_notified={"email_notified": True, "push_notified": False},
            )

    def test_stream_push_notify_global_setting_with_muted_stream(self) -> None:
        # Push notification should not be sent
        do_change_user_setting(
            self.user_profile, "enable_stream_push_notifications", True, acting_user=None
        )
        self.change_subscription_properties({"is_muted": True})
        msg_id = self.send_stream_message(self.iago, "Denmark", content="what's up everyone?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_stream_email_notify_global_setting_with_muted_topic(self) -> None:
        # Email notification should not be sent
        do_change_user_setting(
            self.user_profile, "enable_stream_email_notifications", True, acting_user=None
        )
        do_set_user_topic_visibility_policy(
            self.user_profile,
            get_stream("Denmark", self.user_profile.realm),
            "mutingtest",
            visibility_policy=UserTopic.VisibilityPolicy.MUTED,
        )
        msg_id = self.send_stream_message(
            self.iago, "Denmark", topic_name="mutingtest", content="what's up everyone?"
        )
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                already_notified={"email_notified": False, "push_notified": False},
            )

    def test_stream_push_notify_stream_specific_setting(self) -> None:
        self.change_subscription_properties({"push_notifications": True})
        msg_id = self.send_stream_message(self.iago, "Denmark", content="what's up everyone?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
                stream_push_notify=True,
                stream_email_notify=False,
                already_notified={"email_notified": False, "push_notified": True},
            )

    def test_stream_email_notify_stream_specific_setting(self) -> None:
        self.change_subscription_properties({"email_notifications": True})
        msg_id = self.send_stream_message(self.iago, "Denmark", content="what's up everyone?")
        with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
            missedmessage_hook(self.user_profile.id, self.client_descriptor, True)
            mock_enqueue.assert_called_once()
            args_dict = mock_enqueue.call_args_list[0][1]

            self.assert_maybe_enqueue_notifications_call_args(
                args_dict=args_dict,
                message_id=msg_id,
                user_id=self.user_profile.id,
