from typing import TYPE_CHECKING, Any
from unittest import mock

import orjson
from django.db import connection
from django.test import override_settings
from typing_extensions import override

from zerver.actions.message_flags import do_update_message_flags
from zerver.actions.streams import do_change_stream_group_based_setting, do_change_stream_permission
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.fix_unreads import fix, fix_unsubscribed
from zerver.lib.message import (
    MessageDetailsDict,
    RawUnreadDirectMessageDict,
    RawUnreadMessagesResult,
    UnreadMessagesResult,
    add_message_to_unread_msgs,
    aggregate_unread_data,
    apply_unread_message_event,
    bulk_access_messages,
    bulk_access_stream_messages_query,
    format_unread_message_details,
    get_raw_unread_data,
)
from zerver.lib.message_cache import MessageDict
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import get_subscription
from zerver.lib.user_message import DEFAULT_HISTORICAL_FLAGS, create_historical_user_messages
from zerver.models import (
    Message,
    PushDevice,
    Recipient,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
    UserTopic,
)
from zerver.models.groups import NamedUserGroup
from zerver.models.realms import get_realm
from zerver.models.recipients import get_or_create_direct_message_group
from zerver.models.streams import get_stream

if TYPE_CHECKING:
    from django.test.client import _MonkeyPatchedWSGIResponse as TestHttpResponse


def check_flags(flags: list[str], expected: set[str]) -> None:
    """
    The has_alert_word flag can be ignored for most tests.
    """
    assert "has_alert_word" not in expected
    flag_set = set(flags)
    flag_set.discard("has_alert_word")
    if flag_set != expected:
        raise AssertionError(f"expected flags (ignoring has_alert_word) to be {expected}")


class FirstUnreadAnchorTests(ZulipTestCase):
    """
    HISTORICAL NOTE:

    The two tests in this class were originally written when
    we had the concept of a "pointer", and they may be a bit
    redundant in what they now check.
    """

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

        # Mark all existing messages as read
        result = self.client_post("/json/mark_all_as_read")
        result_dict = self.assert_json_success(result)
        self.assertTrue(result_dict["complete"])

        # Send a new message (this will be unread)
        new_message_id = self.send_stream_message(self.example_user("othello"), "Verona", "test")

        # If we call get_messages with use_first_unread_anchor=True, we
        # should get the message we just sent
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1
        )
        self.assertEqual(messages_response["messages"][0]["id"], new_message_id)
        self.assertEqual(messages_response["anchor"], new_message_id)

        # Test with the old way of expressing use_first_unread_anchor=True
        messages_response = self.get_messages_response(
            anchor=0, num_before=0, num_after=1, use_first_unread_anchor=True
        )
        self.assertEqual(messages_response["messages"][0]["id"], new_message_id)
        self.assertEqual(messages_response["anchor"], new_message_id)

        # We want to get the message_id of an arbitrary old message. We can
        # call get_messages with use_first_unread_anchor=False and simply
        # save the first message we're returned.
        messages = self.get_messages(
            anchor=0, num_before=0, num_after=2, use_first_unread_anchor=False
        )
        old_message_id = messages[0]["id"]

        # Verify the message is marked as read
        user_message = UserMessage.objects.get(
            message_id=old_message_id, user_profile=self.example_user("hamlet")
        )
        self.assertTrue(user_message.flags.read)

        # Let's set this old message to be unread
        result = self.client_post(
            "/json/messages/flags",
            {"messages": orjson.dumps([old_message_id]).decode(), "op": "remove", "flag": "read"},
        )

        # Verify it's now marked as unread
        user_message = UserMessage.objects.get(
            message_id=old_message_id, user_profile=self.example_user("hamlet")
        )
        self.assert_json_success(result)
        self.assertFalse(user_message.flags.read)

        # Now if we call get_messages with use_first_unread_anchor=True,
        # we should get the old message we just set to unread
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1
        )
        self.assertEqual(messages_response["messages"][0]["id"], old_message_id)
        self.assertEqual(messages_response["anchor"], old_message_id)

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

        result = self.client_post("/json/mark_all_as_read")
        result_dict = self.assert_json_success(result)
        self.assertTrue(result_dict["complete"])

        new_message_id = self.send_stream_message(self.example_user("othello"), "Verona", "test")

        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1
        )
        self.assertEqual(messages_response["messages"][0]["id"], new_message_id)
        self.assertEqual(messages_response["anchor"], new_message_id)

        with mock.patch(
            "zerver.lib.narrow.get_first_visible_message_id", return_value=new_message_id
        ):
            messages_response = self.get_messages_response(
                anchor="first_unread", num_before=0, num_after=1
            )
        self.assertEqual(messages_response["messages"][0]["id"], new_message_id)
        self.assertEqual(messages_response["anchor"], new_message_id)

        with mock.patch(
            "zerver.lib.narrow.get_first_visible_message_id",
            return_value=new_message_id + 1,
        ):
            messages_response = self.get_messages_response(
                anchor="first_unread", num_before=0, num_after=1
            )
        self.assert_length(messages_response["messages"], 0)
        self.assertIn("anchor", messages_response)

        with mock.patch(
            "zerver.lib.narrow.get_first_visible_message_id",
            return_value=new_message_id - 1,
        ):
            messages = self.get_messages(anchor="first_unread", num_before=0, num_after=1)
        self.assert_length(messages, 1)


class UnreadCountTests(ZulipTestCase):
    @override
    def setUp(self) -> None:
        super().setUp()
        hamlet = self.example_user("hamlet")
        self.register_push_device(hamlet.id)
        with (
            mock.patch(
                "zerver.lib.push_notifications.send_push_notifications"
            ) as mock_send_push_notifications,
            mock.patch(
                "zerver.lib.push_notifications.push_notifications_configured", return_value=True
            ) as mock_push_notifications_configured,
        ):
            self.unread_msg_ids = [
                self.send_personal_message(self.example_user("iago"), hamlet, "hello"),
                self.send_personal_message(self.example_user("iago"), hamlet, "hello2"),
            ]
            mock_push_notifications_configured.assert_called()
            mock_send_push_notifications.assert_called()

    # Sending a new message results in unread UserMessages being created
    # for users other than sender.
    def test_new_message(self) -> None:
        self.login("hamlet")
        content = "Test message for unset read bit"
        last_msg = self.send_stream_message(self.example_user("hamlet"), "Verona", content)
        user_messages = list(UserMessage.objects.filter(message=last_msg))
        self.assertGreater(len(user_messages), 0)
        for um in user_messages:
            self.assertEqual(um.message.content, content)
            if um.user_profile.delivery_email != self.example_email("hamlet"):
                self.assertFalse(um.flags.read)
            else:
                self.assertTrue(um.flags.read)

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

        with self.captureOnCommitCallbacks(execute=True):
            result = self.client_post(
                "/json/messages/flags",
                {
                    "messages": orjson.dumps(self.unread_msg_ids).decode(),
                    "op": "add",
                    "flag": "read",
                },
            )
        self.assert_json_success(result)

        # Ensure we properly set the flags
        found = 0
        for msg in self.get_messages():
            if msg["id"] in self.unread_msg_ids:
                check_flags(msg["flags"], {"read"})
                found += 1
        self.assertEqual(found, 2)

        result = self.client_post(
            "/json/messages/flags",
            {
                "messages": orjson.dumps([self.unread_msg_ids[1]]).decode(),
                "op": "remove",
                "flag": "read",
            },
        )
        self.assert_json_success(result)

        # Ensure we properly remove just one flag
        for msg in self.get_messages():
            if msg["id"] == self.unread_msg_ids[0]:
                check_flags(msg["flags"], {"read"})
            elif msg["id"] == self.unread_msg_ids[1]:
                check_flags(msg["flags"], set())

    def test_update_flags_race(self) -> None:
        user = self.example_user("hamlet")
        self.login_user(user)
        self.unsubscribe(user, "Verona")

        first_message_id = self.send_stream_message(
            self.example_user("cordelia"),
            "Verona",
            topic_name="testing",
        )
        self.assertFalse(
            UserMessage.objects.filter(
                user_profile_id=user.id, message_id=first_message_id
            ).exists()
        )
        # When adjusting flags of messages that we did not receive, we
        # create UserMessage rows.
        with mock.patch(
            "zerver.actions.message_flags.create_historical_user_messages",
            wraps=create_historical_user_messages,
        ) as mock_backfill:
            result = self.client_post(
                "/json/messages/flags",
                {
                    "messages": orjson.dumps([first_message_id]).decode(),
                    "op": "add",
                    "flag": "starred",
                },
            )
            self.assert_json_success(result)

            mock_backfill.assert_called_once_with(
                user_id=user.id,
                message_ids=[first_message_id],
                flagattr=UserMessage.flags.starred,
                flag_target=UserMessage.flags.starred,
            )
            um_row = UserMessage.objects.get(user_profile_id=user.id, message_id=first_message_id)
            self.assertEqual(
                int(um_row.flags),
                UserMessage.flags.historical | UserMessage.flags.read | UserMessage.flags.starred,
            )

        # That creation may race with other things which also create
        # the UserMessage rows (e.g. reactions); ensure the end result
        # is correct still.
        def race_creation(
            *,
            user_id: int,
            message_ids: list[int],
            flagattr: int | None = None,
            flag_target: int | None = None,
        ) -> None:
            UserMessage.objects.create(
                user_profile_id=user_id, message_id=message_ids[0], flags=DEFAULT_HISTORICAL_FLAGS
            )
            create_historical_user_messages(
                user_id=user_id, message_ids=message_ids, flagattr=flagattr, flag_target=flag_target
            )

        second_message_id = self.send_stream_message(
            self.example_user("cordelia"),
            "Verona",
            topic_name="testing",
        )
        self.assertFalse(
            UserMessage.objects.filter(
                user_profile_id=user.id, message_id=second_message_id
            ).exists()
        )
        with mock.patch(
            "zerver.actions.message_flags.create_historical_user_messages", wraps=race_creation
        ) as mock_backfill:
            result = self.client_post(
                "/json/messages/flags",
                {
                    "messages": orjson.dumps([second_message_id]).decode(),
                    "op": "add",
                    "flag": "starred",
                },
            )
            self.assert_json_success(result)

            mock_backfill.assert_called_once_with(
                user_id=user.id,
                message_ids=[second_message_id],
                flagattr=UserMessage.flags.starred,
                flag_target=UserMessage.flags.starred,
            )

            um_row = UserMessage.objects.get(user_profile_id=user.id, message_id=second_message_id)
            self.assertEqual(
                int(um_row.flags),
                UserMessage.flags.historical | UserMessage.flags.read | UserMessage.flags.starred,
            )

    def test_update_flags_for_narrow(self) -> None:
        user = self.example_user("hamlet")
        self.login_user(user)
        message_ids = [
            self.send_stream_message(
                self.example_user("cordelia"), "Verona", topic_name=f"topic {i % 2}"
            )
            for i in range(10)
        ]

        response = self.assert_json_success(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": message_ids[5],
                    "num_before": 2,
                    "num_after": 2,
                    "narrow": "[]",
                    "op": "add",
                    "flag": "read",
                },
            )
        )
        self.assertEqual(response["processed_count"], 5)
        self.assertEqual(response["updated_count"], 5)
        self.assertEqual(response["first_processed_id"], message_ids[3])
        self.assertEqual(response["last_processed_id"], message_ids[7])
        self.assertEqual(response["found_oldest"], False)
        self.assertEqual(response["found_newest"], False)
        self.assertCountEqual(
            UserMessage.objects.filter(user_profile_id=user.id, message_id__in=message_ids)
            .extra(where=[UserMessage.where_read()])  # noqa: S610
            .values_list("message_id", flat=True),
            message_ids[3:8],
        )

        response = self.assert_json_success(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": message_ids[3],
                    "include_anchor": "false",
                    "num_before": 0,
                    "num_after": 5,
                    "narrow": orjson.dumps(
                        [
                            {"operator": "stream", "operand": "Verona"},
                            {"operator": "topic", "operand": "topic 1"},
                        ]
                    ).decode(),
                    "op": "add",
                    "flag": "starred",
                },
            )
        )
        # In this topic (1, 3, 5, 7, 9), processes everything after 3.
        self.assertEqual(response["processed_count"], 3)
        self.assertEqual(response["updated_count"], 3)
        self.assertEqual(response["first_processed_id"], message_ids[5])
        self.assertEqual(response["last_processed_id"], message_ids[9])
        self.assertEqual(response["found_oldest"], False)
        self.assertEqual(response["found_newest"], True)
        self.assertCountEqual(
            UserMessage.objects.filter(user_profile_id=user.id, message_id__in=message_ids)
            .extra(where=[UserMessage.where_starred()])  # noqa: S610
            .values_list("message_id", flat=True),
            message_ids[5::2],
        )
        response = self.assert_json_success(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": message_ids[3],
                    "include_anchor": "false",
                    "num_before": 0,
                    "num_after": 5,
                    "narrow": orjson.dumps([["stream", "Verona"], ["topic", "topic 1"]]).decode(),
                    "op": "add",
                    "flag": "starred",
                },
            )
        )
        cordelia = self.example_user("cordelia")
        response = self.assert_json_success(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": message_ids[3],
                    "include_anchor": "false",
                    "num_before": 0,
                    "num_after": 5,
                    "narrow": orjson.dumps([{"operator": "dm", "operand": [cordelia.id]}]).decode(),
                    "op": "add",
                    "flag": "starred",
                },
            )
        )

        # Testing the response when marking messages as unread in a
        # narrow that contains messages from unsubscribed streams
        stream_name = "Test Stream"
        stream = self.subscribe(user, stream_name)
        self.subscribe(self.example_user("cordelia"), stream_name)
        message_id = self.send_stream_message(self.example_user("cordelia"), stream_name, "hello")

        self.assert_json_success(
            self.client_post(
                "/json/mark_stream_as_read",
                {
                    "stream_id": stream.id,
                },
            )
        )
        um = UserMessage.objects.get(
            user_profile_id=user.id,
            message_id=message_id,
        )
        self.assertTrue(um.flags.read)

        # Unsubscribe the user from the stream
        self.unsubscribe(user, stream_name)

        # Marking recently added message and all other
        # messages added at the start of the test as unread
        # from an interleaved public narrow
        response = self.assert_json_success(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": message_id,
                    "num_before": 10,
                    "num_after": 0,
                    "narrow": orjson.dumps([{"operator": "streams", "operand": "public"}]).decode(),
                    "op": "remove",
                    "flag": "read",
                },
            )
        )

        self.assertEqual(response["processed_count"], 11)
        self.assertEqual(response["updated_count"], 5)
        self.assertEqual(response["first_processed_id"], message_ids[0])
        self.assertEqual(response["last_processed_id"], message_id)
        self.assertEqual(response["found_oldest"], False)
        self.assertEqual(response["found_newest"], False)
        self.assertEqual(response["ignored_because_not_subscribed_channels"], [stream.id])
        self.assertCountEqual(
            UserMessage.objects.filter(user_profile_id=user.id, message_id__in=message_ids)
            .extra(where=[UserMessage.where_unread()])  # noqa: S610
            .values_list("message_id", flat=True),
            message_ids,
        )

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

        self.assert_json_error(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": "1",
                    "include_anchor": "false",
                    "num_before": "1",
                    "num_after": "1",
                    "narrow": "[[],[]]",
                    "op": "add",
                    "flag": "read",
                },
            ),
            "Invalid narrow[0]: Value error, element is not a string pair",
        )

        self.assert_json_error(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": "1",
                    "include_anchor": "false",
                    "num_before": "1",
                    "num_after": "1",
                    "narrow": orjson.dumps(
                        [
                            {"operator": None, "operand": "Verona"},
                            {"operator": "topic", "operand": "topic 1"},
                        ]
                    ).decode(),
                    "op": "add",
                    "flag": "read",
                },
            ),
            "Invalid narrow[0]: Value error, operator is missing",
        )

        self.assert_json_error(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": "1",
                    "include_anchor": "false",
                    "num_before": "1",
                    "num_after": "1",
                    "narrow": orjson.dumps(
                        [
                            {"operator": "stream", "operand": None},
                            {"operator": "topic", "operand": "topic 1"},
                        ]
                    ).decode(),
                    "op": "add",
                    "flag": "read",
                },
            ),
            "Invalid narrow[0]: Value error, operand is missing",
        )

        self.assert_json_error(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": "1",
                    "include_anchor": "false",
                    "num_before": "1",
                    "num_after": "1",
                    "narrow": orjson.dumps(["asadasd"]).decode(),
                    "op": "add",
                    "flag": "read",
                },
            ),
            "Invalid narrow[0]: Value error, dict or list required",
        )

        self.assert_json_error(
            self.client_post(
                "/json/messages/flags/narrow",
                {
                    "anchor": "1",
                    "include_anchor": "false",
                    "num_before": "1",
                    "num_after": "1",
                    "narrow": orjson.dumps([{"operator": "search", "operand": ""}]).decode(),
                    "op": "add",
                    "flag": "read",
                },
            ),
            "operand cannot be blank.",
        )

        response = self.client_post(
            "/json/messages/flags/narrow",
            {
                "anchor": "0",
                "include_anchor": "false",
                "num_before": "1",
                "num_after": "1",
                "narrow": "[]",
                "op": "add",
                "flag": "read",
            },
        )
        self.assert_json_error(response, "The anchor can only be excluded at an end of the range")

    def test_mark_all_in_stream_read(self) -> None:
        self.login("hamlet")
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")
        iago = self.example_user("iago")
        for user in [hamlet, cordelia, iago]:
            self.subscribe(user, "test_stream")
            self.subscribe(user, "Denmark")
        stream = get_stream("test_stream", hamlet.realm)

        message_id = self.send_stream_message(cordelia, "test_stream", "hello")
        unrelated_message_id = self.send_stream_message(cordelia, "Denmark", "hello")

        with self.capture_send_event_calls(expected_num_events=1) as events:
            result = self.client_post(
                "/json/mark_stream_as_read",
                {
                    "stream_id": stream.id,
                },
            )

        self.assert_json_success(result)

        event = events[0]["event"]
        expected = dict(
            operation="add",
            messages=[message_id],
            flag="read",
            type="update_message_flags",
            all=False,
        )

        differences = [key for key in expected if expected[key] != event[key]]
        self.assert_length(differences, 0)

        # cordelia sent the message and hamlet marked it as read.
        um = list(UserMessage.objects.filter(message=message_id))
        for msg in um:
            if msg.user_profile.email in [hamlet.email, cordelia.email]:
                self.assertTrue(msg.flags.read)
            else:
                self.assertFalse(msg.flags.read)

        # cordelia sent the message, so marked as read only for him.
        unrelated_messages = list(UserMessage.objects.filter(message=unrelated_message_id))
        for msg in unrelated_messages:
            if msg.user_profile.email in [hamlet.email, iago.email]:
                self.assertFalse(msg.flags.read)

    def test_mark_all_in_invalid_stream_read(self) -> None:
        self.login("hamlet")
        invalid_stream_id = "12345678"
        result = self.client_post(
            "/json/mark_stream_as_read",
            {
                "stream_id": invalid_stream_id,
            },
        )
        self.assert_json_error(result, "Invalid channel ID")

    def test_mark_all_topics_unread_with_invalid_stream_name(self) -> None:
        self.login("hamlet")
        invalid_stream_id = "12345678"
        result = self.client_post(
            "/json/mark_topic_as_read",
            {
                "stream_id": invalid_stream_id,
                "topic_name": "whatever",
            },
        )
        self.assert_json_error(result, "Invalid channel ID")

    def test_mark_all_in_stream_topic_read(self) -> None:
        self.login("hamlet")
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")
        for user in [hamlet, cordelia]:
            self.subscribe(user, "test_stream")
            self.subscribe(user, "Denmark")

        message_id = self.send_stream_message(cordelia, "test_stream", "hello", "test_topic")
        unrelated_message_id = self.send_stream_message(cordelia, "Denmark", "hello", "Denmark2")
        with self.capture_send_event_calls(expected_num_events=1) as events:
            result = self.client_post(
                "/json/mark_topic_as_read",
                {
                    "stream_id": get_stream("test_stream", hamlet.realm).id,
                    "topic_name": "test_topic",
                },
            )

        self.assert_json_success(result)

        event = events[0]["event"]
        expected = dict(
            operation="add",
            messages=[message_id],
            flag="read",
            type="update_message_flags",
            all=False,
        )

        differences = [key for key in expected if expected[key] != event[key]]
        self.assert_length(differences, 0)

        um = list(UserMessage.objects.filter(message=message_id))
        for msg in um:
            if msg.user_profile_id == hamlet.id:
                self.assertTrue(msg.flags.read)

        unrelated_messages = list(UserMessage.objects.filter(message=unrelated_message_id))
        for msg in unrelated_messages:
            if msg.user_profile_id == hamlet.id:
                self.assertFalse(msg.flags.read)

    def test_mark_all_in_invalid_topic_read(self) -> None:
        self.login("hamlet")
        invalid_topic_name = "abc"
        result = self.client_post(
            "/json/mark_topic_as_read",
            {
                "stream_id": get_stream("Denmark", get_realm("zulip")).id,
                "topic_name": invalid_topic_name,
            },
        )
        self.assert_json_error(result, "No such topic 'abc'")


class FixUnreadTests(ZulipTestCase):
    def test_fix_unreads(self) -> None:
        user = self.example_user("hamlet")
        othello = self.example_user("othello")
        realm = get_realm("zulip")

        def send_message(stream_name: str, topic_name: str) -> int:
            self.subscribe(othello, stream_name)
            msg_id = self.send_stream_message(othello, stream_name, topic_name=topic_name)
            um = UserMessage.objects.get(user_profile=user, message_id=msg_id)
            return um.id

        def assert_read(user_message_id: int) -> None:
            um = UserMessage.objects.get(id=user_message_id)
            self.assertTrue(um.flags.read)

        def assert_unread(user_message_id: int) -> None:
            um = UserMessage.objects.get(id=user_message_id)
            self.assertFalse(um.flags.read)

        def mute_stream(stream_name: str) -> None:
            stream = get_stream(stream_name, realm)
            recipient = stream.recipient
            subscription = Subscription.objects.get(
                user_profile=user,
                recipient=recipient,
            )
            subscription.is_muted = True
            subscription.save()

        def mute_topic(stream_name: str, topic_name: str) -> None:
            stream = get_stream(stream_name, realm)

            do_set_user_topic_visibility_policy(
                user,
                stream,
                topic_name,
                visibility_policy=UserTopic.VisibilityPolicy.MUTED,
            )

        def force_unsubscribe(stream_name: str) -> None:
            """
            We don't want side effects here, since the eventual
            unsubscribe path may mark messages as read, defeating
            the test setup here.
            """
            sub = get_subscription(stream_name, user)
            sub.active = False
            sub.save()

        # The data setup here is kind of funny, because some of these
        # conditions should not actually happen in practice going forward,
        # but we may have had bad data from the past.

        mute_stream("Denmark")
        mute_topic("Verona", "muted_topic")

        um_normal_id = send_message("Verona", "normal")
        um_muted_topic_id = send_message("Verona", "muted_topic")
        um_muted_stream_id = send_message("Denmark", "whatever")

        self.subscribe(user, "temporary")
        um_unsubscribed_id = send_message("temporary", "whatever")
        force_unsubscribe("temporary")

        # Verify the setup
        assert_unread(um_normal_id)
        assert_unread(um_muted_topic_id)
        assert_unread(um_muted_stream_id)
        assert_unread(um_unsubscribed_id)

        # fix unsubscribed
        with (
            connection.cursor() as cursor,
            self.assertLogs("zulip.fix_unreads", "INFO") as info_logs,
        ):
            fix_unsubscribed(cursor, user)

        self.assertEqual(info_logs.output[0], "INFO:zulip.fix_unreads:get recipients")
        self.assertTrue("INFO:zulip.fix_unreads:[" in info_logs.output[1])
        self.assertTrue("INFO:zulip.fix_unreads:elapsed time:" in info_logs.output[2])
        self.assertEqual(
            info_logs.output[3],
            "INFO:zulip.fix_unreads:finding unread messages for non-active streams",
        )
        self.assertEqual(info_logs.output[4], "INFO:zulip.fix_unreads:rows found: 1")
        self.assertTrue("INFO:zulip.fix_unreads:elapsed time:" in info_logs.output[5])
        self.assertEqual(
            info_logs.output[6],
            "INFO:zulip.fix_unreads:fixing unread messages for non-active streams",
        )
        self.assertTrue("INFO:zulip.fix_unreads:elapsed time:" in info_logs.output[7])

        # Muted messages don't change.
        assert_unread(um_muted_topic_id)
        assert_unread(um_muted_stream_id)
        assert_unread(um_normal_id)

        # The unsubscribed entry should change.
        assert_read(um_unsubscribed_id)

        with self.assertLogs("zulip.fix_unreads", "INFO") as info_logs:
            # test idempotency
            fix(user)

        self.assertEqual(info_logs.output[0], f"INFO:zulip.fix_unreads:\n---\nFixing {user.id}:")
        self.assertEqual(info_logs.output[1], "INFO:zulip.fix_unreads:get recipients")
        self.assertTrue("INFO:zulip.fix_unreads:[" in info_logs.output[2])
        self.assertTrue("INFO:zulip.fix_unreads:elapsed time:" in info_logs.output[3])
        self.assertEqual(
            info_logs.output[4],
            "INFO:zulip.fix_unreads:finding unread messages for non-active streams",
        )
        self.assertEqual(info_logs.output[5], "INFO:zulip.fix_unreads:rows found: 0")
        self.assertTrue("INFO:zulip.fix_unreads:elapsed time:" in info_logs.output[6])

        assert_unread(um_normal_id)
        assert_unread(um_muted_topic_id)
        assert_unread(um_muted_stream_id)
        assert_read(um_unsubscribed_id)


