import hashlib
import random
from collections.abc import Sequence
from datetime import timedelta
from io import StringIO
from typing import TYPE_CHECKING, Any
from unittest import mock

import orjson
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import transaction
from django.http import HttpResponse
from django.utils.timezone import now as timezone_now
from typing_extensions import override

from zerver.actions.bots import do_change_bot_owner
from zerver.actions.channel_folders import check_add_channel_folder
from zerver.actions.create_realm import do_create_realm
from zerver.actions.default_streams import do_add_default_stream, do_create_default_stream_group
from zerver.actions.message_delete import do_delete_messages
from zerver.actions.realm_settings import (
    do_change_realm_permission_group_setting,
    do_change_realm_plan_type,
    do_set_realm_property,
)
from zerver.actions.streams import (
    bulk_add_subscriptions,
    bulk_remove_subscriptions,
    deactivated_streams_by_old_name,
    do_change_stream_group_based_setting,
    do_change_stream_permission,
    do_deactivate_stream,
    do_set_stream_property,
    do_unarchive_stream,
)
from zerver.actions.user_groups import bulk_add_members_to_user_groups, check_add_user_group
from zerver.actions.users import do_deactivate_user
from zerver.lib.attachments import (
    validate_attachment_request,
    validate_attachment_request_for_spectator_access,
)
from zerver.lib.exceptions import JsonableError
from zerver.lib.message import UnreadStreamInfo, aggregate_unread_data, get_raw_unread_data
from zerver.lib.response import json_success
from zerver.lib.stream_color import STREAM_ASSIGNMENT_COLORS, pick_colors
from zerver.lib.stream_subscription import (
    get_active_subscriptions_for_stream_id,
    num_subscribers_for_stream_id,
    subscriber_ids_with_stream_history_access,
)
from zerver.lib.stream_traffic import (
    get_average_weekly_stream_traffic,
    round_to_2_significant_digits,
)
from zerver.lib.streams import (
    StreamDict,
    StreamsCategorizedByPermissionsForAddingSubscribers,
    can_access_stream_metadata_user_ids,
    do_get_streams,
    ensure_stream,
    filter_stream_authorization_for_adding_subscribers,
    list_to_streams,
)
from zerver.lib.subscription_info import (
    bulk_get_subscriber_user_ids,
    gather_subscriptions,
    gather_subscriptions_helper,
    validate_user_access_to_subscribers_helper,
)
from zerver.lib.test_classes import ZulipTestCase, get_topic_messages
from zerver.lib.test_helpers import HostRequestMock, cache_tries_captured
from zerver.lib.types import UserGroupMembersData
from zerver.lib.user_groups import UserGroupMembershipDetails, get_group_setting_value_for_api
from zerver.models import (
    Attachment,
    DefaultStream,
    DefaultStreamGroup,
    Message,
    NamedUserGroup,
    Realm,
    RealmAuditLog,
    Recipient,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
)
from zerver.models.groups import SystemGroups
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import get_realm
from zerver.models.streams import StreamTopicsPolicyEnum, get_default_stream_groups, get_stream
from zerver.models.users import active_non_guest_user_ids, get_user, get_user_profile_by_id_in_realm
from zerver.views.streams import compose_views

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


def check_subscriptions_exists(user_profile: UserProfile, stream: Stream) -> bool:
    return (
        get_active_subscriptions_for_stream_id(stream.id, include_deactivated_users=True)
        .filter(
            user_profile=user_profile,
        )
        .exists()
    )


class TestMiscStuff(ZulipTestCase):
    def test_test_helper(self) -> None:
        cordelia = self.example_user("cordelia")
        s = self.subscribed_stream_name_list(cordelia)
        self.assertIn("* Verona", s)
        self.assertNotIn("* Denmark", s)

    def test_pick_colors(self) -> None:
        used_colors: set[str] = set()
        color_map: dict[int, str] = {}
        recipient_ids = list(range(30))
        user_color_map = pick_colors(used_colors, color_map, recipient_ids)
        self.assertEqual(
            user_color_map,
            {
                0: "#76ce90",
                1: "#fae589",
                2: "#a6c7e5",
                3: "#e79ab5",
                4: "#bfd56f",
                5: "#f4ae55",
                6: "#b0a5fd",
                7: "#addfe5",
                8: "#f5ce6e",
                9: "#c2726a",
                10: "#94c849",
                11: "#bd86e5",
                12: "#ee7e4a",
                13: "#a6dcbf",
                14: "#95a5fd",
                15: "#53a063",
                16: "#9987e1",
                17: "#e4523d",
                18: "#c2c2c2",
                19: "#4f8de4",
                20: "#c6a8ad",
                21: "#e7cc4d",
                22: "#c8bebf",
                23: "#a47462",
                # start repeating
                24: "#76ce90",
                25: "#fae589",
                26: "#a6c7e5",
                27: "#e79ab5",
                28: "#bfd56f",
                29: "#f4ae55",
            },
        )

        color_map = {98: "color98", 99: "color99"}
        used_colors = set(STREAM_ASSIGNMENT_COLORS) - {"#c6a8ad", "#9987e1"}
        recipient_ids = [99, 98, 1, 2, 3, 4]
        user_color_map = pick_colors(used_colors, color_map, recipient_ids)
        self.assertEqual(
            user_color_map,
            {98: "color98", 99: "color99", 1: "#9987e1", 2: "#c6a8ad", 3: "#e79ab5", 4: "#bfd56f"},
        )

        """
        If we are assigning colors to a user with 24+ streams, we have to start
        reusing old colors.  Our algorithm basically uses recipient_id % 24, so
        the following code reflects the worse case scenario that our new
        streams have recipient ids spaced out by exact multiples of 24.  We
        don't try to work around this edge case, since users who really depend
        on the stream colors can always just assign themselves custom colors
        for the streams that they really want to stand out.

        Even if recipient_ids were completely random, the odds of collisions
        are low, but it's often the case that bulk-adds are done for streams
        that either were or are being created at roughly the same time, so the
        recipient_ids tend to have even fewer collisions.
        """
        used_colors = set(STREAM_ASSIGNMENT_COLORS)
        color_map = {}
        recipient_ids = [2, 26, 50, 74]
        user_color_map = pick_colors(used_colors, color_map, recipient_ids)
        self.assertEqual(
            user_color_map,
            {2: "#a6c7e5", 26: "#a6c7e5", 50: "#a6c7e5", 74: "#a6c7e5"},
        )

    def test_empty_results(self) -> None:
        # These are essentially just tests to ensure line
        # coverage for codepaths that won't ever really be
        # called in practice.

        user_profile = self.example_user("cordelia")

        result = bulk_get_subscriber_user_ids(
            stream_dicts=[],
            user_profile=user_profile,
            subscribed_stream_ids=set(),
            streams_to_partially_fetch=[],
        )
        self.assertEqual(result, {})

        streams = do_get_streams(
            user_profile=user_profile,
            include_public=False,
            include_subscribed=False,
            include_all=False,
            include_default=False,
        )
        self.assertEqual(streams, [])


class RecipientTest(ZulipTestCase):
    def test_recipient(self) -> None:
        realm = get_realm("zulip")
        stream = get_stream("Verona", realm)
        recipient = Recipient.objects.get(
            type_id=stream.id,
            type=Recipient.STREAM,
        )
        self.assertEqual(repr(recipient), f"<Recipient: Verona ({stream.id}, {Recipient.STREAM})>")


class StreamAdminTest(ZulipTestCase):
    def test_make_stream_public(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        self.make_stream("private_stream_1", invite_only=True)
        self.make_stream("private_stream_2", invite_only=True)

        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        params = {
            "is_private": orjson.dumps(False).decode(),
        }
        stream_id = get_stream("private_stream_1", user_profile.realm).id
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_error(result, "Channel content access is required.")

        stream = self.subscribe(user_profile, "private_stream_1")

        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        params = {
            "is_private": orjson.dumps(False).decode(),
        }
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)

        realm = user_profile.realm
        stream = get_stream("private_stream_1", realm)
        self.assertFalse(stream.invite_only)
        self.assertTrue(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 1)
        expected_notification = (
            f"@_**King Hamlet|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Private, protected history** to **Public**."
        )
        self.assertEqual(messages[0].content, expected_notification)

        history_public_to_subscribers_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert history_public_to_subscribers_log is not None

        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: False,
            RealmAuditLog.NEW_VALUE: True,
            "property": "history_public_to_subscribers",
        }
        self.assertEqual(history_public_to_subscribers_log.extra_data, expected_extra_data)

        invite_only_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).order_by("-id")[1]
        assert invite_only_log is not None

        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: True,
            RealmAuditLog.NEW_VALUE: False,
            "property": "invite_only",
        }
        self.assertEqual(invite_only_log.extra_data, expected_extra_data)

        private_stream = self.make_stream("private_stream", realm=realm, invite_only=True)
        realm.moderation_request_channel = private_stream
        realm.save()
        params = {
            "is_private": orjson.dumps(False).decode(),
        }
        result = self.client_patch(f"/json/streams/{private_stream.id}", params)
        self.assert_json_error(result, "Moderation request channel must be private.")
        self.assertTrue(private_stream.invite_only)

        stream = self.subscribe(user_profile, "private_stream_3", invite_only=True)
        self.set_user_role(user_profile, UserProfile.ROLE_REALM_OWNER)
        nobody_group = NamedUserGroup.objects.get(
            name=SystemGroups.NOBODY, realm_for_sharding=realm, is_system_group=True
        )
        do_change_realm_permission_group_setting(
            realm,
            "can_create_public_channel_group",
            nobody_group,
            acting_user=None,
        )
        result = self.client_patch(f"/json/streams/{stream.id}", params)
        self.assert_json_error(result, "Insufficient permission")

        user_profile_group = check_add_user_group(
            realm, "user_profile_group", [user_profile], acting_user=user_profile
        )
        do_change_realm_permission_group_setting(
            realm,
            "can_create_public_channel_group",
            user_profile_group,
            acting_user=None,
        )
        result = self.client_patch(f"/json/streams/{stream.id}", params)
        self.assert_json_success(result)
        stream = get_stream("private_stream_3", realm)
        self.assertFalse(stream.invite_only)

    def test_make_stream_private(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        realm = user_profile.realm
        self.make_stream("public_stream_1", realm=realm)
        self.make_stream("public_stream_2")

        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        params = {
            "is_private": orjson.dumps(True).decode(),
        }
        stream_id = self.subscribe(user_profile, "public_stream_1").id
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)
        stream = get_stream("public_stream_1", realm)
        self.assertTrue(stream.invite_only)
        self.assertFalse(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 1)
        expected_notification = (
            f"@_**King Hamlet|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Public** to **Private, protected history**."
        )
        self.assertEqual(messages[0].content, expected_notification)

        history_public_to_subscribers_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert history_public_to_subscribers_log is not None

        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: True,
            RealmAuditLog.NEW_VALUE: False,
            "property": "history_public_to_subscribers",
        }
        self.assertEqual(history_public_to_subscribers_log.extra_data, expected_extra_data)

        invite_only_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).order_by("-id")[1]
        assert invite_only_log is not None

        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: False,
            RealmAuditLog.NEW_VALUE: True,
            "property": "invite_only",
        }
        self.assertEqual(invite_only_log.extra_data, expected_extra_data)

        default_stream = self.make_stream("default_stream", realm=realm)
        do_add_default_stream(default_stream)
        params = {
            "is_private": orjson.dumps(True).decode(),
        }
        result = self.client_patch(f"/json/streams/{default_stream.id}", params)
        self.assert_json_error(result, "A default channel cannot be private.")
        self.assertFalse(default_stream.invite_only)

        stream = self.subscribe(user_profile, "public_stream_3")
        self.set_user_role(user_profile, UserProfile.ROLE_REALM_OWNER)
        nobody_group = NamedUserGroup.objects.get(
            name=SystemGroups.NOBODY, realm_for_sharding=realm, is_system_group=True
        )
        do_change_realm_permission_group_setting(
            realm,
            "can_create_private_channel_group",
            nobody_group,
            acting_user=None,
        )
        result = self.client_patch(f"/json/streams/{stream.id}", params)
        self.assert_json_error(result, "Insufficient permission")

        user_profile_group = check_add_user_group(
            realm, "user_profile_group", [user_profile], acting_user=user_profile
        )
        do_change_realm_permission_group_setting(
            realm,
            "can_create_private_channel_group",
            user_profile_group,
            acting_user=None,
        )
        result = self.client_patch(f"/json/streams/{stream.id}", params)
        self.assert_json_success(result)
        stream = get_stream("public_stream_3", realm)
        self.assertTrue(stream.invite_only)

        # Test notification is not sent in `channel events` if `send_channel_events_messages` is `False`.
        do_set_realm_property(stream.realm, "send_channel_events_messages", False, acting_user=None)
        stream = self.make_stream("stream_without_notification")
        result = self.client_patch(
            f"/json/streams/{stream.id}",
            params,
        )
        self.assert_json_success(result)
        with self.assertRaises(Message.DoesNotExist):
            Message.objects.get(recipient__type_id=stream.id)

    def test_create_web_public_stream(self) -> None:
        user_profile = self.example_user("hamlet")
        owner = self.example_user("desdemona")

        stream_names = ["new1", "new2", "new3"]
        stream_descriptions = ["des1", "des2", "des3"]
        streams_raw: list[StreamDict] = [
            {"name": stream_name, "description": stream_description, "is_web_public": True}
            for (stream_name, stream_description) in zip(
                stream_names, stream_descriptions, strict=False
            )
        ]

        request_settings_dict = dict.fromkeys(Stream.stream_permission_group_settings)

        self.assertFalse(user_profile.can_create_web_public_streams())
        self.assertTrue(owner.can_create_web_public_streams())
        # As per can_create_web_public_channel_group, only owners
        # can create web-public streams by default.
        with self.assertRaisesRegex(JsonableError, "Insufficient permission"):
            list_to_streams(
                streams_raw,
                user_profile,
                autocreate=True,
                request_settings_dict=request_settings_dict,
            )

        with self.settings(WEB_PUBLIC_STREAMS_ENABLED=False):
            self.assertFalse(user_profile.can_create_web_public_streams())
            self.assertFalse(owner.can_create_web_public_streams())
            with self.assertRaisesRegex(JsonableError, "Web-public channels are not enabled."):
                list_to_streams(
                    streams_raw,
                    owner,
                    autocreate=True,
                    request_settings_dict=request_settings_dict,
                )

        existing_streams, new_streams = list_to_streams(
            streams_raw,
            owner,
            autocreate=True,
            request_settings_dict=request_settings_dict,
        )

        self.assert_length(new_streams, 3)
        self.assert_length(existing_streams, 0)

        actual_stream_names = {stream.name for stream in new_streams}
        self.assertEqual(actual_stream_names, set(stream_names))
        actual_stream_descriptions = {stream.description for stream in new_streams}
        self.assertEqual(actual_stream_descriptions, set(stream_descriptions))
        for stream in new_streams:
            self.assertTrue(stream.is_web_public)

    def test_make_stream_private_with_public_history(self) -> None:
        # Convert a public stream to a private stream with shared history
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        realm = user_profile.realm
        self.make_stream("public_history_stream", realm=realm)

        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        params = {
            "is_private": orjson.dumps(True).decode(),
            "history_public_to_subscribers": orjson.dumps(True).decode(),
        }
        stream_id = self.subscribe(user_profile, "public_history_stream").id
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)
        stream = get_stream("public_history_stream", realm)
        self.assertTrue(stream.invite_only)
        self.assertTrue(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 1)
        expected_notification = (
            f"@_**King Hamlet|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Public** to **Private, shared history**."
        )
        self.assertEqual(messages[0].content, expected_notification)

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: False,
            RealmAuditLog.NEW_VALUE: True,
            "property": "invite_only",
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)

        # Convert a private stream with protected history to a private stream
        # with shared history.
        self.make_stream(
            "private_protected_stream",
            realm=realm,
            invite_only=True,
            history_public_to_subscribers=False,
        )
        params = {
            "is_private": orjson.dumps(True).decode(),
            "history_public_to_subscribers": orjson.dumps(True).decode(),
        }
        stream_id = self.subscribe(user_profile, "private_protected_stream").id
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)
        stream = get_stream("private_protected_stream", realm)
        self.assertTrue(stream.invite_only)
        self.assertTrue(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 1)
        expected_notification = (
            f"@_**King Hamlet|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Private, protected history** to **Private, shared history**."
        )
        self.assertEqual(messages[0].content, expected_notification)

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: False,
            RealmAuditLog.NEW_VALUE: True,
            "property": "history_public_to_subscribers",
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)

        # Test notification is not sent in `channel events` if `send_channel_events_messages` is `False`.
        do_set_realm_property(stream.realm, "send_channel_events_messages", False, acting_user=None)
        stream = self.make_stream("stream_without_notification")
        result = self.client_patch(
            f"/json/streams/{stream.id}",
            params,
        )
        self.assert_json_success(result)
        with self.assertRaises(Message.DoesNotExist):
            Message.objects.get(recipient__type_id=stream.id)

    def test_make_stream_web_public(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        realm = user_profile.realm
        self.make_stream("test_stream", realm=realm)
        stream = self.subscribe(user_profile, "test_stream")
        stream_id = stream.id

        user_profile_group = check_add_user_group(
            realm, "user_profile_group", [user_profile], acting_user=user_profile
        )
        do_change_stream_group_based_setting(
            stream,
            "can_administer_channel_group",
            user_profile_group,
            acting_user=user_profile,
        )

        params = {
            "is_web_public": orjson.dumps(True).decode(),
            "history_public_to_subscribers": orjson.dumps(True).decode(),
        }
        owners_group = NamedUserGroup.objects.get(
            name=SystemGroups.OWNERS, realm_for_sharding=realm, is_system_group=True
        )
        do_change_realm_permission_group_setting(
            realm,
            "can_create_web_public_channel_group",
            owners_group,
            acting_user=None,
        )
        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_error(result, "Insufficient permission")

        nobody_group = NamedUserGroup.objects.get(
            name=SystemGroups.NOBODY, realm_for_sharding=realm, is_system_group=True
        )
        do_change_realm_permission_group_setting(
            realm,
            "can_create_web_public_channel_group",
            nobody_group,
            acting_user=None,
        )
        self.set_user_role(user_profile, UserProfile.ROLE_REALM_OWNER)
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_error(result, "Insufficient permission")

        do_change_realm_permission_group_setting(
            realm,
            "can_create_web_public_channel_group",
            owners_group,
            acting_user=None,
        )
        self.set_user_role(user_profile, UserProfile.ROLE_REALM_OWNER)
        with self.settings(WEB_PUBLIC_STREAMS_ENABLED=False):
            result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_error(result, "Web-public channels are not enabled.")

        bad_params = {
            "is_web_public": orjson.dumps(True).decode(),
            "is_private": orjson.dumps(True).decode(),
            "history_public_to_subscribers": orjson.dumps(True).decode(),
        }
        result = self.client_patch(f"/json/streams/{stream_id}", bad_params)
        self.assert_json_error(result, "Invalid parameters")

        bad_params = {
            "is_web_public": orjson.dumps(True).decode(),
            "is_private": orjson.dumps(False).decode(),
            "history_public_to_subscribers": orjson.dumps(False).decode(),
        }
        result = self.client_patch(f"/json/streams/{stream_id}", bad_params)
        self.assert_json_error(result, "Invalid parameters")

        stream = get_stream("test_stream", realm)
        self.assertFalse(stream.is_web_public)

        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)

        stream = get_stream("test_stream", realm)
        self.assertTrue(stream.is_web_public)
        self.assertFalse(stream.invite_only)
        self.assertTrue(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 1)
        expected_notification = (
            f"@_**King Hamlet|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Public** to **Web-public**."
        )
        self.assertEqual(messages[0].content, expected_notification)

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: False,
            RealmAuditLog.NEW_VALUE: True,
            "property": "is_web_public",
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)

    def test_change_history_access_for_private_streams(self) -> None:
        user_profile = self.example_user("iago")
        self.login_user(user_profile)
        realm = user_profile.realm
        self.make_stream("private_stream", realm=realm, invite_only=True)
        stream_id = self.subscribe(user_profile, "private_stream").id

        params = {
            "history_public_to_subscribers": orjson.dumps(True).decode(),
        }
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)

        stream = get_stream("private_stream", realm)
        self.assertTrue(stream.invite_only)
        self.assertTrue(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 1)
        expected_notification = (
            f"@_**Iago|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Private, protected history** to **Private, shared history**."
        )
        self.assertEqual(messages[0].content, expected_notification)

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: False,
            RealmAuditLog.NEW_VALUE: True,
            "property": "history_public_to_subscribers",
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)

        params = {
            "history_public_to_subscribers": orjson.dumps(False).decode(),
        }
        result = self.client_patch(f"/json/streams/{stream_id}", params)
        self.assert_json_success(result)

        stream = get_stream("private_stream", realm)
        self.assertTrue(stream.invite_only)
        self.assertFalse(stream.history_public_to_subscribers)

        messages = get_topic_messages(user_profile, stream, "channel events")
        self.assert_length(messages, 2)
        expected_notification = (
            f"@_**Iago|{user_profile.id}** changed the [access permissions](/help/channel-permissions) "
            "for this channel from **Private, shared history** to **Private, protected history**."
        )
        self.assertEqual(messages[1].content, expected_notification)

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.CHANNEL_PROPERTY_CHANGED,
            modified_stream=stream,
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: True,
            RealmAuditLog.NEW_VALUE: False,
            "property": "history_public_to_subscribers",
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)

    def test_updating_protected_history_and_can_create_topic_group_for_streams(self) -> None:
        user_profile = self.example_user("iago")
        self.login_user(user_profile)
        realm = user_profile.realm
        stream = self.make_stream(
            "stream", realm, invite_only=True, history_public_to_subscribers=True
        )
        everyone_system_group = NamedUserGroup.objects.get(
            name="role:everyone", realm=realm, is_system_group=True
        )
        moderators_system_group = NamedUserGroup.objects.get(
            name="role:moderators", realm=realm, is_system_group=True
        )
        hamletcharacters_group = NamedUserGroup.objects.get(name="hamletcharacters", realm=realm)
        anonymous_group = UserGroupMembersData(
            direct_members=[user_profile.id],
            direct_subgroups=[moderators_system_group.id],
        )

        error_msg = "Unsupported parameter combination: history_public_to_subscribers, can_create_topic_group"

        def check_stream_property_update(
            property_name: str,
            setting_value: bool | int | UserGroupMembersData,
            expect_fail: bool = False,
        ) -> None:
            params = {property_name: orjson.dumps(setting_value).decode()}
            if property_name == "can_create_topic_group":
                params = {property_name: orjson.dumps({"new": setting_value}).decode()}

            result = self.client_patch(f"/json/streams/{stream.id}", params)
            stream.refresh_from_db()

            if expect_fail:
                self.assert_json_error(result, error_msg)
                return

            self.assert_json_success(result)
            if property_name == "can_create_topic_group":
                expected_value = setting_value
                current_value = get_group_setting_value_for_api(stream.can_create_topic_group)
                self.assertEqual(expected_value, current_value)
            else:
                self.assertEqual(getattr(stream, property_name), setting_value)

        # Test changing can_create_topic_group setting for a private stream
        # with protected history.
        do_change_stream_permission(
            stream,
            invite_only=True,
            history_public_to_subscribers=False,
            is_web_public=False,
