from datetime import timedelta
from typing import TYPE_CHECKING, Any

import orjson
from django.conf import settings
from django.test import override_settings
from django.utils.timezone import now as timezone_now
from typing_extensions import override

from analytics.models import StreamCount
from zerver.actions.streams import (
    do_change_stream_group_based_setting,
    do_change_stream_permission,
    do_deactivate_stream,
)
from zerver.lib.email_mirror_helpers import encode_email_address, get_channel_email_token
from zerver.lib.subscription_info import gather_subscriptions, gather_subscriptions_helper
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import most_recent_message
from zerver.lib.types import (
    APIStreamDict,
    APISubscriptionDict,
    NeverSubscribedStreamDict,
    SubscriptionInfo,
    UserGroupMembersData,
    UserGroupMembersDict,
)
from zerver.models import NamedUserGroup, Realm, Stream, Subscription, UserProfile
from zerver.models.groups import SystemGroups
from zerver.models.realms import get_realm
from zerver.models.streams import get_stream
from zerver.models.users import get_system_bot

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


def fix_expected_fields_for_stream_group_settings(expected_fields: set[str]) -> set[str]:
    for setting_name in Stream.stream_permission_group_settings:
        expected_fields -= {setting_name + "_id"}
        expected_fields |= {setting_name}

    return expected_fields


class GetStreamsTest(ZulipTestCase):
    def test_streams_api_for_bot_owners(self) -> None:
        hamlet = self.example_user("hamlet")
        test_bot = self.create_test_bot("foo", hamlet)
        assert test_bot is not None
        realm = get_realm("zulip")
        self.login_user(hamlet)

        # Check it correctly lists the bot owner's subs with
        # include_owner_subscribed=true
        filters = dict(
            include_owner_subscribed="true",
            include_public="false",
            include_subscribed="false",
        )
        with self.assert_database_query_count(7):
            result = self.api_get(test_bot, "/api/v1/streams", filters)
        owner_subs = self.api_get(hamlet, "/api/v1/users/me/subscriptions")

        json = self.assert_json_success(result)
        self.assertIn("streams", json)
        self.assertIsInstance(json["streams"], list)

        self.assert_json_success(owner_subs)
        owner_subs_json = orjson.loads(owner_subs.content)

        self.assertEqual(
            sorted(s["name"] for s in json["streams"]),
            sorted(s["name"] for s in owner_subs_json["subscriptions"]),
        )

        # Check it correctly lists the bot owner's subs and the
        # bot's subs
        self.subscribe(test_bot, "Scotland")
        filters = dict(
            include_owner_subscribed="true",
            include_public="false",
            include_subscribed="true",
        )
        with self.assert_database_query_count(8):
            result = self.api_get(test_bot, "/api/v1/streams", filters)

        json = self.assert_json_success(result)
        self.assertIn("streams", json)
        self.assertIsInstance(json["streams"], list)

        actual = sorted(s["name"] for s in json["streams"])
        expected = [s["name"] for s in owner_subs_json["subscriptions"]]
        expected.append("Scotland")
        expected.sort()

        self.assertEqual(actual, expected)

        # Check it correctly lists the bot owner's subs + all public streams
        self.make_stream("private_stream", realm=realm, invite_only=True)
        self.subscribe(test_bot, "private_stream")
        with self.assert_database_query_count(7):
            result = self.api_get(
                test_bot,
                "/api/v1/streams",
                {
                    "include_owner_subscribed": "true",
                    "include_public": "true",
                    "include_subscribed": "false",
                },
            )

        json = self.assert_json_success(result)
        self.assertIn("streams", json)
        self.assertIsInstance(json["streams"], list)

        actual = sorted(s["name"] for s in json["streams"])
        expected = [s["name"] for s in owner_subs_json["subscriptions"]]
        expected.extend(["Rome", "Venice", "Scotland"])
        expected.sort()

        self.assertEqual(actual, expected)

        # Check it correctly lists the bot owner's subs + all public streams +
        # the bot's subs
        with self.assert_database_query_count(8):
            result = self.api_get(
                test_bot,
                "/api/v1/streams",
                {
                    "include_owner_subscribed": "true",
                    "include_public": "true",
                    "include_subscribed": "true",
                },
            )

        json = self.assert_json_success(result)
        self.assertIn("streams", json)
        self.assertIsInstance(json["streams"], list)

        actual = sorted(s["name"] for s in json["streams"])
        expected = [s["name"] for s in owner_subs_json["subscriptions"]]
        expected.extend(["Rome", "Venice", "Scotland", "private_stream"])
        expected.sort()

        self.assertEqual(actual, expected)

        private_stream_2 = self.make_stream("private_stream_2", realm=realm, invite_only=True)
        private_stream_3 = self.make_stream("private_stream_3", realm=realm, invite_only=True)
        self.make_stream("private_stream_4", realm=realm, invite_only=True)
        test_bot_group_member_dict = UserGroupMembersData(
            direct_members=[test_bot.id], direct_subgroups=[]
        )
        do_change_stream_group_based_setting(
            private_stream_2,
            "can_add_subscribers_group",
            test_bot_group_member_dict,
            acting_user=hamlet,
        )
        do_change_stream_group_based_setting(
            private_stream_3,
            "can_administer_channel_group",
            test_bot_group_member_dict,
            acting_user=hamlet,
        )
        # Check it correctly lists the bot owner's subs + the channels
        # bot has content access to.
        with self.assert_database_query_count(10):
            result = self.api_get(
                test_bot,
                "/api/v1/streams",
                {
                    "include_owner_subscribed": "true",
                    "include_can_access_content": "true",
                },
            )

        json = self.assert_json_success(result)
        self.assertIn("streams", json)
        self.assertIsInstance(json["streams"], list)

        actual = sorted(s["name"] for s in json["streams"])
        expected = [s["name"] for s in owner_subs_json["subscriptions"]]
        expected.extend(["Rome", "Venice", "Scotland", "private_stream", "private_stream_2"])
        expected.sort()

        self.assertEqual(actual, expected)

    def test_all_streams_api(self) -> None:
        url = "/api/v1/streams"
        data = {"include_all": "true"}
        backward_compatible_data = {"include_all_active": "true"}

        # Normal user should be able to make this request and get all
        # the streams they have metadata access to.
        normal_user = self.example_user("cordelia")
        realm = normal_user.realm
        normal_user_group_members_dict = UserGroupMembersData(
            direct_members=[normal_user.id], direct_subgroups=[]
        )

        private_stream_1 = self.make_stream("private_stream_1", realm=realm, invite_only=True)
        private_stream_2 = self.make_stream("private_stream_2", realm=realm, invite_only=True)
        private_stream_3 = self.make_stream("private_stream_3", realm=realm, invite_only=True)
        self.make_stream("private_stream_4", realm=realm, invite_only=True)
        deactivated_public_stream = self.make_stream(
            "deactivated_public_stream", realm=realm, invite_only=False
        )
        do_deactivate_stream(deactivated_public_stream, acting_user=normal_user)

        self.subscribe(normal_user, private_stream_1.name)
        do_change_stream_group_based_setting(
            private_stream_2,
            "can_add_subscribers_group",
            normal_user_group_members_dict,
            acting_user=normal_user,
        )
        do_change_stream_group_based_setting(
            private_stream_3,
            "can_administer_channel_group",
            normal_user_group_members_dict,
            acting_user=normal_user,
        )

        result_stream_names: list[str] = [
            stream.name
            for stream in Stream.objects.filter(realm=realm, invite_only=False, deactivated=False)
        ]
        result_stream_names.extend(
            [private_stream_1.name, private_stream_2.name, private_stream_3.name]
        )
        with self.assert_database_query_count(8):
            result = self.api_get(normal_user, url, data)
        json = self.assert_json_success(result)
        self.assertEqual(sorted(s["name"] for s in json["streams"]), sorted(result_stream_names))

        # Normal user should be able to make this request and get all
        # the streams they have metadata access to.
        guest_user = self.example_user("polonius")
        guest_user_group_member_dict = UserGroupMembersData(
            direct_members=[guest_user.id], direct_subgroups=[]
        )

        self.subscribe(guest_user, private_stream_1.name)
        self.subscribe(guest_user, "design")
        do_change_stream_group_based_setting(
            private_stream_2,
            "can_add_subscribers_group",
            guest_user_group_member_dict,
            acting_user=normal_user,
        )
        do_change_stream_group_based_setting(
            get_stream("Rome", realm),
            "can_add_subscribers_group",
            guest_user_group_member_dict,
            acting_user=normal_user,
        )
        do_change_stream_group_based_setting(
            private_stream_3,
            "can_administer_channel_group",
            guest_user_group_member_dict,
            acting_user=normal_user,
        )
        do_change_stream_group_based_setting(
            get_stream("Denmark", realm),
            "can_administer_channel_group",
            guest_user_group_member_dict,
            acting_user=normal_user,
        )

        # Guest user should not gain metadata access to a channel via
        # `can_add_subscribers_group` or `can_administer_channel_group`
        # since `allow_everyone_group` if false for both of those groups.
        result_stream_names = ["Verona", "private_stream_1", "design", "Rome"]
        with self.assert_database_query_count(7):
            result = self.api_get(guest_user, url, data)
        json = self.assert_json_success(result)
        self.assertEqual(sorted(s["name"] for s in json["streams"]), sorted(result_stream_names))

        # Realm admin users can see all active streams if
        # `exclude_archived` is not set.
        admin_user = self.example_user("iago")
        self.assertTrue(admin_user.is_realm_admin)

        with self.assert_database_query_count(7):
            result = self.api_get(admin_user, url, data)
        json = self.assert_json_success(result)

        backward_compatible_result = self.api_get(admin_user, url, backward_compatible_data)
        json_for_backward_compatible_request = self.assert_json_success(backward_compatible_result)

        self.assertEqual(json, json_for_backward_compatible_request)

        self.assertIn("streams", json)
        self.assertIsInstance(json["streams"], list)

        stream_names = {s["name"] for s in json["streams"]}
        result_stream_names = [
            stream.name for stream in Stream.objects.filter(realm=realm, deactivated=False)
        ]
        self.assertEqual(
            sorted(stream_names),
            sorted(result_stream_names),
        )

        # Realm admin users can see all streams if `exclude_archived`
        # is set to false.
        data = {"include_all": "true", "exclude_archived": "false"}
        with self.assert_database_query_count(7):
            result = self.api_get(admin_user, url, data)
        json = self.assert_json_success(result)
        stream_names = {s["name"] for s in json["streams"]}
        result_stream_names = [stream.name for stream in Stream.objects.filter(realm=realm)]
        self.assertEqual(
            sorted(stream_names),
            sorted(result_stream_names),
        )

        # This case will not happen in practice, we are adding this
        # test block to add coverage for the case where
        # `get_metadata_access_streams` returns an empty list without
        # query if an empty list of streams is passed to it.
        all_active_streams = Stream.objects.filter(realm=realm, deactivated=False)
        for stream in all_active_streams:
            do_deactivate_stream(stream, acting_user=None)

        data = {"include_all": "true"}
        with self.assert_database_query_count(3):
            result = self.api_get(admin_user, url, data)
        json = self.assert_json_success(result)
        stream_names = {s["name"] for s in json["streams"]}
        self.assertEqual(stream_names, set())

    def test_public_streams_api(self) -> None:
        """
        Ensure that the query we use to get public streams successfully returns
        a list of streams
        """
        user = self.example_user("hamlet")
        realm = get_realm("zulip")
        self.login_user(user)

        # Check it correctly lists the user's subs with include_public=false
        result = self.api_get(user, "/api/v1/streams", {"include_public": "false"})
        result2 = self.api_get(user, "/api/v1/users/me/subscriptions")

        json = self.assert_json_success(result)

        self.assertIn("streams", json)

        self.assertIsInstance(json["streams"], list)

        self.assert_json_success(result2)
        json2 = orjson.loads(result2.content)

        self.assertEqual(
            sorted(s["name"] for s in json["streams"]),
            sorted(s["name"] for s in json2["subscriptions"]),
        )

        # Check it correctly lists all public streams with include_subscribed=false
        filters = dict(include_public="true", include_subscribed="false")
        result = self.api_get(user, "/api/v1/streams", filters)
        json = self.assert_json_success(result)
        all_streams = [
            stream.name for stream in Stream.objects.filter(realm=realm, invite_only=False)
        ]
        self.assertEqual(sorted(s["name"] for s in json["streams"]), sorted(all_streams))

    def test_include_can_access_content_streams_api(self) -> None:
        """
        Ensure that the query we use to get public streams successfully returns
        a list of streams
        """
        # Cordelia is not subscribed to private stream `core team`.
        user = self.example_user("cordelia")
        realm = get_realm("zulip")
        self.login_user(user)
        user_group_members_dict = UserGroupMembersData(
            direct_members=[user.id], direct_subgroups=[]
        )

        private_stream_1 = self.make_stream("private_stream_1", realm=realm, invite_only=True)
        private_stream_2 = self.make_stream("private_stream_2", realm=realm, invite_only=True)
        private_stream_3 = self.make_stream("private_stream_3", realm=realm, invite_only=True)
        self.make_stream("private_stream_4", realm=realm, invite_only=True)

        self.subscribe(user, private_stream_1.name)
        do_change_stream_group_based_setting(
            private_stream_2, "can_add_subscribers_group", user_group_members_dict, acting_user=user
        )
        do_change_stream_group_based_setting(
            private_stream_3,
            "can_administer_channel_group",
            user_group_members_dict,
            acting_user=user,
        )

        # Check it correctly lists all content access streams with
        # include_can_access_content=false
        filters = dict(include_can_access_content="true")
        with self.assert_database_query_count(8):
            result = self.api_get(user, "/api/v1/streams", filters)
        json = self.assert_json_success(result)
        result_streams = [
            stream.name for stream in Stream.objects.filter(realm=realm, invite_only=False)
        ]
        result_streams.extend([private_stream_1.name, private_stream_2.name])
        self.assertEqual(sorted(s["name"] for s in json["streams"]), sorted(result_streams))

    def test_get_single_stream_api(self) -> None:
        self.login("hamlet")
        realm = get_realm("zulip")
        denmark_stream = get_stream("Denmark", realm)
        result = self.client_get(f"/json/streams/{denmark_stream.id}")
        json = self.assert_json_success(result)
        self.assertEqual(json["stream"]["name"], "Denmark")
        self.assertEqual(json["stream"]["stream_id"], denmark_stream.id)

        result = self.client_get("/json/streams/9999")
        self.assert_json_error(result, "Invalid channel ID")

        private_stream = self.make_stream("private_stream", invite_only=True)
        self.subscribe(self.example_user("cordelia"), "private_stream")

        # Non-admins cannot access unsubscribed private streams.
        result = self.client_get(f"/json/streams/{private_stream.id}")
        self.assert_json_error(result, "Invalid channel ID")

        self.login("iago")
        result = self.client_get(f"/json/streams/{private_stream.id}")
        json = self.assert_json_success(result)
        self.assertEqual(json["stream"]["name"], "private_stream")
        self.assertEqual(json["stream"]["stream_id"], private_stream.id)

        self.login("cordelia")
        result = self.client_get(f"/json/streams/{private_stream.id}")
        json = self.assert_json_success(result)
        self.assertEqual(json["stream"]["name"], "private_stream")
        self.assertEqual(json["stream"]["stream_id"], private_stream.id)

    def test_get_stream_email_address(self) -> None:
        self.login("hamlet")
        hamlet = self.example_user("hamlet")
        iago = self.example_user("iago")
        polonius = self.example_user("polonius")
        realm = get_realm("zulip")
        email_gateway_bot = get_system_bot(settings.EMAIL_GATEWAY_BOT, realm.id)
        denmark_stream = get_stream("Denmark", realm)
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        json = self.assert_json_success(result)
        email_token = get_channel_email_token(
            denmark_stream, creator=hamlet, sender=email_gateway_bot
        )
        hamlet_denmark_email = encode_email_address(
            denmark_stream.name, email_token, show_sender=True
        )
        self.assertEqual(json["email"], hamlet_denmark_email)

        self.login("polonius")
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        self.assert_json_error(result, "Invalid channel ID")

        self.subscribe(polonius, "Denmark")
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        json = self.assert_json_success(result)
        email_token = get_channel_email_token(
            denmark_stream, creator=polonius, sender=email_gateway_bot
        )
        polonius_denmark_email = encode_email_address(
            denmark_stream.name, email_token, show_sender=True
        )
        self.assertEqual(json["email"], polonius_denmark_email)

        do_change_stream_permission(
            denmark_stream,
            invite_only=True,
            history_public_to_subscribers=True,
            is_web_public=False,
            acting_user=iago,
        )
        self.login("hamlet")
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        json = self.assert_json_success(result)
        self.assertEqual(json["email"], hamlet_denmark_email)

        self.unsubscribe(hamlet, "Denmark")
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        self.assert_json_error(result, "Invalid channel ID")

        self.login("iago")
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        json = self.assert_json_success(result)
        email_token = get_channel_email_token(
            denmark_stream, creator=iago, sender=email_gateway_bot
        )
        iago_denmark_email = encode_email_address(
            denmark_stream.name, email_token, show_sender=True
        )
        self.assertEqual(json["email"], iago_denmark_email)

        self.unsubscribe(iago, "Denmark")
        result = self.client_get(f"/json/streams/{denmark_stream.id}/email_address")
        self.assert_json_error(result, "Invalid channel ID")

    def test_guest_user_access_to_streams(self) -> None:
        user_profile = self.example_user("polonius")
        self.login_user(user_profile)
        self.assertEqual(user_profile.role, UserProfile.ROLE_GUEST)

        # Get all the streams that Polonius has access to (subscribed + web-public streams)
        result = self.client_get("/json/streams", {"include_web_public": "true"})
        streams = self.assert_json_success(result)["streams"]
        sub_info = gather_subscriptions_helper(user_profile)

        subscribed = sub_info.subscriptions
        unsubscribed = sub_info.unsubscribed
        never_subscribed = sub_info.never_subscribed

        self.assert_length(streams, len(subscribed) + len(unsubscribed) + len(never_subscribed))
        stream_names = [stream["name"] for stream in streams]
        expected_stream_names = [stream["name"] for stream in subscribed + unsubscribed]
        expected_stream_names += [stream["name"] for stream in never_subscribed]
        self.assertEqual(set(stream_names), set(expected_stream_names))


class StreamIdTest(ZulipTestCase):
    def test_get_stream_id(self) -> None:
        user = self.example_user("hamlet")
        self.login_user(user)
        stream = gather_subscriptions(user)[0][0]
        result = self.client_get("/json/get_stream_id", {"stream": stream["name"]})
        response_dict = self.assert_json_success(result)
        self.assertEqual(response_dict["stream_id"], stream["stream_id"])

    def test_get_stream_id_wrong_name(self) -> None:
        user = self.example_user("hamlet")
        self.login_user(user)
        result = self.client_get("/json/get_stream_id", {"stream": "wrongname"})
        self.assert_json_error(result, "Invalid channel name 'wrongname'")


class GetSubscribersTest(ZulipTestCase):
    @override
    def setUp(self) -> None:
        super().setUp()
        self.user_profile = self.example_user("hamlet")
        self.login_user(self.user_profile)

    def test_api_fields(self) -> None:
        """Verify that all the fields from `Stream.API_FIELDS` and `Subscription.API_FIELDS` present
        in `APIStreamDict` and `APISubscriptionDict`, respectively.
        """
        expected_fields = set(Stream.API_FIELDS) | {"stream_id", "is_archived"}
        expected_fields -= {"id", "deactivated"}
        expected_fields = fix_expected_fields_for_stream_group_settings(expected_fields)

        stream_dict_fields = set(APIStreamDict.__annotations__.keys())
        computed_fields = {
            "is_announcement_only",
            "is_default",
            "stream_post_policy",
            "stream_weekly_traffic",
        }

        self.assertEqual(stream_dict_fields - computed_fields, expected_fields)

        expected_fields = set(Subscription.API_FIELDS)

        subscription_dict_fields = set(APISubscriptionDict.__annotations__.keys())
        computed_fields = {"in_home_view", "email_address", "stream_weekly_traffic", "subscribers"}
        # `APISubscriptionDict` is a subclass of `APIStreamDict`, therefore having all the
        # fields in addition to the computed fields and `Subscription.API_FIELDS` that
        # need to be excluded here.
        self.assertEqual(
            subscription_dict_fields - computed_fields - stream_dict_fields,
            expected_fields,
        )

    def verify_sub_fields(self, sub_data: SubscriptionInfo) -> None:
        other_fields = {
            "is_archived",
            "is_announcement_only",
            "in_home_view",
            "stream_id",
            "stream_post_policy",
            "stream_weekly_traffic",
            "subscribers",
        }

        expected_fields = set(Stream.API_FIELDS) | set(Subscription.API_FIELDS) | other_fields
        expected_fields -= {"id", "deactivated"}
        expected_fields = fix_expected_fields_for_stream_group_settings(expected_fields)

        for lst in [sub_data.subscriptions, sub_data.unsubscribed]:
            for sub in lst:
                self.assertEqual(set(sub), expected_fields)

        other_fields = {
            "is_archived",
            "is_announcement_only",
            "stream_id",
            "stream_post_policy",
            "stream_weekly_traffic",
            "subscribers",
        }

        expected_fields = set(Stream.API_FIELDS) | other_fields
        expected_fields -= {"id", "deactivated"}
        expected_fields = fix_expected_fields_for_stream_group_settings(expected_fields)

        for never_sub in sub_data.never_subscribed:
            self.assertEqual(set(never_sub), expected_fields)

    def assert_user_got_subscription_notification(
        self, user: UserProfile, expected_msg: str
    ) -> None:
        # verify that the user was sent a message informing them about the subscription
        realm = user.realm
        msg = most_recent_message(user)
        self.assertEqual(msg.recipient.type, msg.recipient.PERSONAL)
        self.assertEqual(msg.sender_id, self.notification_bot(realm).id)

        def non_ws(s: str) -> str:
            return s.replace("\n", "").replace(" ", "")

        assert msg.rendered_content is not None
        self.assertEqual(non_ws(msg.rendered_content), non_ws(expected_msg))

    def check_well_formed_result(
        self, result: dict[str, Any], stream_name: str, realm: Realm
    ) -> None:
        """
        A successful call to get_subscribers returns the list of subscribers in
        the form:

        {"msg": "",
         "result": "success",
         "subscribers": [hamlet_user.id, prospero_user.id]}
        """
        self.assertIn("subscribers", result)
        self.assertIsInstance(result["subscribers"], list)
        true_subscribers = [
            user_profile.id for user_profile in self.users_subscribed_to_stream(stream_name, realm)
        ]
        self.assertEqual(sorted(result["subscribers"]), sorted(true_subscribers))

    def make_subscriber_request(
        self, stream_id: int, user: UserProfile | None = None
    ) -> "TestHttpResponse":
        if user is None:
            user = self.user_profile
        return self.api_get(user, f"/api/v1/streams/{stream_id}/members")

    def make_successful_subscriber_request(self, stream_name: str) -> None:
        stream_id = get_stream(stream_name, self.user_profile.realm).id
        result = self.make_subscriber_request(stream_id)
        response_dict = self.assert_json_success(result)
        self.check_well_formed_result(response_dict, stream_name, self.user_profile.realm)

    def test_subscriber(self) -> None:
        """
        get_subscribers returns the list of subscribers.
        """
        stream_name = gather_subscriptions(self.user_profile)[0][0]["name"]
        self.make_successful_subscriber_request(stream_name)

    @override_settings(MIN_PARTIAL_SUBSCRIBERS_CHANNEL_SIZE=5)
    def test_gather_partial_subscriptions(self) -> None:
        othello = self.example_user("othello")
        user_names = ["iago", "cordelia", "polonius", "shiva", "prospero"]
        idle_users = [self.example_user(name) for name in user_names]
        for user in idle_users:
            user.long_term_idle = True
            user.save()
        bot = self.create_test_bot("bot", othello, "Foo Bot")

        stream_names = [
            "never_subscribed_only_bots",
            "never_subscribed_many_more_than_bots",
            "unsubscribed_only_bots",
            "subscribed_more_than_bots_including_idle",
            "subscribed_many_more_than_bots",
        ]
        for stream_name in stream_names:
            self.make_stream(stream_name)

        self.subscribe_via_post(
            self.user_profile,
            ["never_subscribed_only_bots"],
            dict(principals=orjson.dumps([bot.id]).decode()),
        )
        self.subscribe_via_post(
            self.user_profile,
            ["never_subscribed_many_more_than_bots"],
            dict(
                principals=orjson.dumps(
                    [bot.id, othello.id] + [user.id for user in idle_users]
                ).decode()
            ),
        )
        self.subscribe_via_post(
            self.user_profile,
            ["unsubscribed_only_bots"],
            dict(principals=orjson.dumps([bot.id, self.user_profile.id]).decode()),
        )
        self.unsubscribe(
            self.user_profile,
            "unsubscribed_only_bots",
        )
        self.subscribe_via_post(
            self.user_profile,
            ["subscribed_more_than_bots_including_idle"],
            dict(
                principals=orjson.dumps(
                    [bot.id, othello.id, self.user_profile.id, idle_users[0].id]
                ).decode()
            ),
        )
        self.subscribe_via_post(
            self.user_profile,
            ["subscribed_many_more_than_bots"],
            dict(
                principals=orjson.dumps(
                    [bot.id, othello.id, self.user_profile.id] + [user.id for user in idle_users]
                ).decode()
            ),
        )

        with self.assert_database_query_count(9):
            sub_data = gather_subscriptions_helper(self.user_profile, include_subscribers="partial")
            never_subscribed_streams = sub_data.never_subscribed
            unsubscribed_streams = sub_data.unsubscribed
            subscribed_streams = sub_data.subscriptions
        self.assertGreaterEqual(len(never_subscribed_streams), 2)
        self.assertGreaterEqual(len(unsubscribed_streams), 1)
        self.assertGreaterEqual(len(subscribed_streams), 1)

        # Streams with only bots have sent all of their subscribers,
        # since we always send bots. We tell the client it doesn't
        # need to fetch more, by filling "subscribers" instead
        # of "partial_subscribers". If there are non-bot subscribers,
        # a partial fetch will return only partial subscribers.

        for sub in never_subscribed_streams:
            if sub["name"] == "never_subscribed_only_bots":
                self.assert_length(sub["subscribers"], 1)
                self.assertIsNone(sub.get("partial_subscribers"))
                continue
            if sub["name"] == "never_subscribed_many_more_than_bots":
                # the bot and Othello (who is not long_term_idle)
                self.assert_length(sub["partial_subscribers"], 2)
                self.assertIsNone(sub.get("subscribers"))

        for sub in unsubscribed_streams:
            if sub["name"] == "unsubscribed_only_bots":
                self.assert_length(sub["subscribers"], 1)
                self.assertIsNone(sub.get("partial_subscribers"))
                break

        for sub in subscribed_streams:
            # fewer than MIN_PARTIAL_SUBSCRIBERS_CHANNEL_SIZE subscribers,
            # so we get all of them
            if sub["name"] == "subscribed_more_than_bots_including_idle":
                self.assertNotIn("partial_subscribers", sub)
                self.assert_length(sub["subscribers"], 4)
            if sub["name"] == "subscribed_many_more_than_bots":
                # the bot, Othello (who is not long_term_idle), and current user
                self.assert_length(sub["partial_subscribers"], 3)
                self.assertNotIn("subscribers", sub)

    @override_settings(MIN_PARTIAL_SUBSCRIBERS_CHANNEL_SIZE=5)
    def test_gather_partial_subscriptions_api(self) -> None:
        othello = self.example_user("othello")
        user_names = ["iago", "cordelia", "polonius", "shiva", "prospero"]
        idle_users = [self.example_user(name) for name in user_names]
        for user in idle_users:
            user.long_term_idle = True
            user.save()
        bot = self.create_test_bot("bot", othello, "Foo Bot")

        stream_names = [
            "subscribed_more_than_bots_including_idle",
            "subscribed_many_more_than_bots",
        ]
        for stream_name in stream_names:
            self.make_stream(stream_name)

