import time
from collections.abc import Callable
from typing import Any
from unittest import mock
from urllib.parse import urlsplit

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

from zerver.actions.channel_folders import check_add_channel_folder
from zerver.actions.custom_profile_fields import try_update_realm_custom_profile_field
from zerver.actions.message_send import check_send_message
from zerver.actions.presence import do_update_user_presence
from zerver.actions.streams import do_change_stream_folder
from zerver.actions.user_settings import do_change_user_setting
from zerver.lib.event_schema import check_web_reload_client_event
from zerver.lib.events import fetch_initial_state_data, post_process_state
from zerver.lib.exceptions import AccessDeniedError
from zerver.lib.request import RequestVariableMissingError
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    HostRequestMock,
    dummy_handler,
    reset_email_visibility_to_everyone_in_zulip_realm,
    stub_event_queue_user_events,
)
from zerver.lib.users import get_users_for_api
from zerver.models import CustomProfileField, UserMessage, UserPresence, UserProfile
from zerver.models.clients import get_client
from zerver.models.realms import get_realm
from zerver.models.streams import get_stream
from zerver.models.users import get_system_bot
from zerver.tornado.event_queue import (
    allocate_client_descriptor,
    clear_client_event_queues_for_testing,
    get_client_info_for_message_event,
    mark_clients_to_reload,
    process_message_event,
    send_web_reload_client_events,
)
from zerver.tornado.exceptions import BadEventQueueIdError
from zerver.tornado.views import get_events
from zerver.views.events_register import _default_all_public_streams, _default_narrow


class EventsEndpointTest(ZulipTestCase):
    def test_events_register_without_user_agent(self) -> None:
        result = self.client_post("/json/register", skip_user_agent=True)
        self.assert_json_success(result)

    def test_narrows(self) -> None:
        user = self.example_user("hamlet")
        with mock.patch("zerver.lib.events.request_event_queue", return_value=None) as m:
            munge = lambda obj: orjson.dumps(obj).decode()
            narrow = [["stream", "devel"], ["is", "mentioned"]]
            payload = dict(narrow=munge(narrow))
            result = self.api_post(user, "/api/v1/register", payload)

        # We want the test to abort before actually fetching data.
        self.assert_json_error(result, "Could not allocate event queue")

        self.assertEqual(m.call_args.kwargs["narrow"], [["stream", "devel"], ["is", "mentioned"]])

    def test_events_register_endpoint(self) -> None:
        # This test is intended to get minimal coverage on the
        # events_register code paths
        user = self.example_user("hamlet")
        with mock.patch("zerver.views.events_register.do_events_register", return_value={}):
            result = self.api_post(user, "/api/v1/register")
        self.assert_json_success(result)

        with mock.patch("zerver.lib.events.request_event_queue", return_value=None):
            result = self.api_post(user, "/api/v1/register")
        self.assert_json_error(result, "Could not allocate event queue")

        return_event_queue = "15:11"
        return_user_events: list[dict[str, Any]] = []

        # We choose realm_emoji somewhat randomly--we want
        # a "boring" event type for the purpose of this test.
        event_type = "realm_emoji"
        empty_realm_emoji_dict: dict[str, Any] = {}
        test_event = dict(id=6, type=event_type, realm_emoji=empty_realm_emoji_dict)

        # Test that call is made to deal with a returning soft deactivated user.
        with (
            mock.patch("zerver.lib.events.reactivate_user_if_soft_deactivated") as fa,
            stub_event_queue_user_events(return_event_queue, return_user_events),
        ):
            result = self.api_post(
                user, "/api/v1/register", dict(event_types=orjson.dumps([event_type]).decode())
            )
            self.assertEqual(fa.call_count, 1)

        with stub_event_queue_user_events(return_event_queue, return_user_events):
            result = self.api_post(
                user, "/api/v1/register", dict(event_types=orjson.dumps([event_type]).decode())
            )

        result_dict = self.assert_json_success(result)
        self.assertEqual(result_dict["last_event_id"], -1)
        self.assertEqual(result_dict["queue_id"], "15:11")

        # Now start simulating returning actual data
        return_event_queue = "15:12"
        return_user_events = [test_event]

        with stub_event_queue_user_events(return_event_queue, return_user_events):
            result = self.api_post(
                user, "/api/v1/register", dict(event_types=orjson.dumps([event_type]).decode())
            )

        result_dict = self.assert_json_success(result)
        self.assertEqual(result_dict["last_event_id"], 6)
        self.assertEqual(result_dict["queue_id"], "15:12")

        # sanity check the data relevant to our event
        self.assertEqual(result_dict["realm_emoji"], {})

        # Now test with `fetch_event_types` not matching the event
        return_event_queue = "15:13"
        with stub_event_queue_user_events(return_event_queue, return_user_events):
            result = self.api_post(
                user,
                "/api/v1/register",
                dict(
                    event_types=orjson.dumps([event_type]).decode(),
                    fetch_event_types=orjson.dumps(["message"]).decode(),
                ),
            )
        result_dict = self.assert_json_success(result)
        self.assertEqual(result_dict["last_event_id"], 6)
        # Check that the message event types data is in there
        self.assertIn("max_message_id", result_dict)

        # Check that our original event type is not there.
        self.assertNotIn(event_type, result_dict)
        self.assertEqual(result_dict["queue_id"], "15:13")

        # Now test with `fetch_event_types` matching the event
        with stub_event_queue_user_events(return_event_queue, return_user_events):
            result = self.api_post(
                user,
                "/api/v1/register",
                dict(
                    fetch_event_types=orjson.dumps([event_type]).decode(),
                    event_types=orjson.dumps(["message"]).decode(),
                ),
            )
        result_dict = self.assert_json_success(result)
        self.assertEqual(result_dict["last_event_id"], 6)
        # Check that we didn't fetch the messages data
        self.assertNotIn("max_message_id", result_dict)

        # Check that the realm_emoji data is in there.
        self.assertIn("realm_emoji", result_dict)
        self.assertEqual(result_dict["realm_emoji"], {})
        self.assertEqual(result_dict["queue_id"], "15:13")

    def test_events_register_spectators(self) -> None:
        # Verify that POST /register works for spectators, but not for
        # normal users.
        with self.settings(WEB_PUBLIC_STREAMS_ENABLED=False), self.assert_database_query_count(2):
            result = self.client_post("/json/register")
            self.assert_json_error(
                result,
                "Not logged in: API authentication or user session required",
                status_code=401,
            )

        with self.assert_database_query_count(15):
            result = self.client_post("/json/register")
            result_dict = self.assert_json_success(result)
            self.assertEqual(result_dict["queue_id"], None)
            self.assertEqual(result_dict["realm_url"], "http://zulip.testserver")
            self.assertEqual(result_dict["realm_uri"], "http://zulip.testserver")

        result = self.client_post("/json/register")
        self.assertEqual(result.status_code, 200)

        result = self.client_post("/json/register", dict(client_gravatar="false"))
        self.assertEqual(result.status_code, 200)

        result = self.client_post("/json/register", dict(client_gravatar="true"))
        self.assert_json_error(
            result,
            "Invalid 'client_gravatar' parameter for anonymous request",
            status_code=400,
        )

        result = self.client_post("/json/register", dict(include_subscribers="true"))
        self.assert_json_error(
            result,
            "Invalid 'include_subscribers' parameter for anonymous request",
            status_code=400,
        )

    def test_channel_folders_for_spectators(self) -> None:
        realm = get_realm("zulip")
        iago = self.example_user("iago")

        frontend_folder = check_add_channel_folder(realm, "Frontend", "", acting_user=iago)
        backend_folder = check_add_channel_folder(realm, "Backend", "", acting_user=iago)

        result = self.client_post("/json/register")
        self.assertEqual(result.status_code, 200)

        channel_folders_data = orjson.loads(result.content)["channel_folders"]
        self.assert_length(channel_folders_data, 0)

        web_public_stream = get_stream("Rome", realm)
        do_change_stream_folder(web_public_stream, frontend_folder, acting_user=iago)

        public_stream = get_stream("Verona", realm)
        do_change_stream_folder(public_stream, backend_folder, acting_user=iago)

        result = self.client_post("/json/register")
        self.assertEqual(result.status_code, 200)

        channel_folders_data = orjson.loads(result.content)["channel_folders"]
        self.assert_length(channel_folders_data, 1)
        self.assertEqual(channel_folders_data[0]["name"], "Frontend")

    def test_events_register_endpoint_all_public_streams_access(self) -> None:
        guest_user = self.example_user("polonius")
        normal_user = self.example_user("hamlet")
        self.assertEqual(guest_user.role, UserProfile.ROLE_GUEST)
        self.assertEqual(normal_user.role, UserProfile.ROLE_MEMBER)

        with mock.patch("zerver.views.events_register.do_events_register", return_value={}):
            result = self.api_post(normal_user, "/api/v1/register", dict(all_public_streams="true"))
        self.assert_json_success(result)

        with mock.patch("zerver.views.events_register.do_events_register", return_value={}):
            result = self.api_post(guest_user, "/api/v1/register", dict(all_public_streams="true"))
        self.assert_json_error(result, "User not authorized for this query")

    def test_events_get_events_endpoint_guest_cant_use_all_public_streams_param(self) -> None:
        """
        This test is meant to execute the very beginning of the codepath
        to ensure guest users are immediately disallowed to use the
        all_public_streams param. Deeper testing is hard (and not necessary for this case)
        due to the codepath expecting AsyncDjangoHandler to be attached to the request,
        which doesn't happen in our test setup.
        """

        guest_user = self.example_user("polonius")
        self.assertEqual(guest_user.role, UserProfile.ROLE_GUEST)

        result = self.api_get(guest_user, "/api/v1/events", dict(all_public_streams="true"))
        self.assert_json_error(result, "User not authorized for this query")

    def test_tornado_endpoint(self) -> None:
        # This test is mostly intended to get minimal coverage on the
        # /api/internal/notify_tornado endpoint (only used in
        # puppeteer tests), so we can have 100% URL coverage, but it
        # does exercise a little bit of the codepath.
        post_data = dict(
            data=orjson.dumps(
                dict(
                    event=dict(
                        type="other",
                    ),
                    users=[self.example_user("hamlet").id],
                ),
            ).decode(),
        )
        req = HostRequestMock(post_data)
        req.META["REMOTE_ADDR"] = "127.0.0.1"
        with self.assertRaises(RequestVariableMissingError) as context:
            result = self.client_post_request("/api/internal/notify_tornado", req)
        self.assertEqual(str(context.exception), "Missing 'secret' argument")
        self.assertEqual(context.exception.http_status_code, 400)

        post_data["secret"] = "random"
        req = HostRequestMock(post_data, user_profile=None)
        req.META["REMOTE_ADDR"] = "127.0.0.1"
        with self.assertRaises(AccessDeniedError) as access_denied_error:
            result = self.client_post_request("/api/internal/notify_tornado", req)
        self.assertEqual(str(access_denied_error.exception), "Access denied")
        self.assertEqual(access_denied_error.exception.http_status_code, 403)

        post_data["secret"] = settings.SHARED_SECRET
        req = HostRequestMock(post_data, tornado_handler=dummy_handler)
        req.META["REMOTE_ADDR"] = "127.0.0.1"
        result = self.client_post_request("/api/internal/notify_tornado", req)
        self.assert_json_success(result)

        post_data = dict(secret=settings.SHARED_SECRET)
        req = HostRequestMock(post_data, tornado_handler=dummy_handler)
        req.META["REMOTE_ADDR"] = "127.0.0.1"
        with self.assertRaises(RequestVariableMissingError) as context:
            result = self.client_post_request("/api/internal/notify_tornado", req)
        self.assertEqual(str(context.exception), "Missing 'data' argument")
        self.assertEqual(context.exception.http_status_code, 400)

    def test_web_reload_clients(self) -> None:
        # Minimal testing of the /api/internal/web_reload_clients endpoint
        post_data = {
            "client_count": "1",
            "immediate": orjson.dumps(False).decode(),
            "secret": settings.SHARED_SECRET,
        }
        req = HostRequestMock(post_data, tornado_handler=dummy_handler)
        req.META["REMOTE_ADDR"] = "127.0.0.1"
        result = self.client_post_request("/api/internal/web_reload_clients", req)
        self.assert_json_success(result)
        self.assertEqual(orjson.loads(result.content)["sent_events"], 0)


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

    def test_get_events(self) -> None:
        user_profile = self.example_user("hamlet")
        email = user_profile.email
        recipient_user_profile = self.example_user("othello")
        recipient_email = recipient_user_profile.email
        self.login_user(user_profile)

        result = self.tornado_call(
            get_events,
            user_profile,
            {
                "apply_markdown": orjson.dumps(True).decode(),
                "client_gravatar": orjson.dumps(True).decode(),
                "event_types": orjson.dumps(["message"]).decode(),
                "user_client": "website",
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        self.assert_json_success(result)
        queue_id = orjson.loads(result.content)["queue_id"]

        recipient_result = self.tornado_call(
            get_events,
            recipient_user_profile,
            {
                "apply_markdown": orjson.dumps(True).decode(),
                "client_gravatar": orjson.dumps(True).decode(),
                "event_types": orjson.dumps(["message"]).decode(),
                "user_client": "website",
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        self.assert_json_success(recipient_result)
        recipient_queue_id = orjson.loads(recipient_result.content)["queue_id"]

        result = self.tornado_call(
            get_events,
            user_profile,
            {
                "queue_id": queue_id,
                "user_client": "website",
                "last_event_id": -1,
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        events = orjson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 0)

        local_id = "10.01"
        with self.captureOnCommitCallbacks(execute=True):
            check_send_message(
                sender=user_profile,
                client=get_client("whatever"),
                recipient_type_name="private",
                message_to=[recipient_email],
                topic_name=None,
                message_content="hello",
                local_id=local_id,
                sender_queue_id=queue_id,
            )

        result = self.tornado_call(
            get_events,
            user_profile,
            {
                "queue_id": queue_id,
                "user_client": "website",
                "last_event_id": -1,
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        events = orjson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 1)
        self.assertEqual(events[0]["type"], "message")
        self.assertEqual(events[0]["message"]["sender_email"], email)
        self.assertEqual(events[0]["local_message_id"], local_id)
        self.assertEqual(events[0]["message"]["display_recipient"][0]["is_mirror_dummy"], False)
        self.assertEqual(events[0]["message"]["display_recipient"][1]["is_mirror_dummy"], False)
        self.assertEqual(events[0]["message"]["recipient_id"], recipient_user_profile.recipient_id)

        last_event_id = events[0]["id"]
        local_id = "10.02"

        with self.captureOnCommitCallbacks(execute=True):
            check_send_message(
                sender=user_profile,
                client=get_client("whatever"),
                recipient_type_name="private",
                message_to=[recipient_email],
                topic_name=None,
                message_content="hello",
                local_id=local_id,
                sender_queue_id=queue_id,
            )

        result = self.tornado_call(
            get_events,
            user_profile,
            {
                "queue_id": queue_id,
                "user_client": "website",
                "last_event_id": last_event_id,
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        events = orjson.loads(result.content)["events"]
        self.assert_json_success(result)
        self.assert_length(events, 1)
        self.assertEqual(events[0]["type"], "message")
        self.assertEqual(events[0]["message"]["sender_email"], email)
        self.assertEqual(events[0]["local_message_id"], local_id)
        self.assertEqual(events[0]["message"]["recipient_id"], recipient_user_profile.recipient_id)

        # Test that the received message in the receiver's event queue
        # exists and does not contain a local id
        recipient_result = self.tornado_call(
            get_events,
            recipient_user_profile,
            {
                "queue_id": recipient_queue_id,
                "user_client": "website",
                "last_event_id": -1,
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        recipient_events = orjson.loads(recipient_result.content)["events"]
        self.assert_json_success(recipient_result)
        self.assert_length(recipient_events, 2)
        self.assertEqual(recipient_events[0]["type"], "message")
        self.assertEqual(recipient_events[0]["message"]["sender_email"], email)
        self.assertTrue("local_message_id" not in recipient_events[0])
        # Incoming DMs show the recipient_id that outgoing DMs would.
        self.assertEqual(recipient_events[0]["message"]["recipient_id"], user_profile.recipient_id)
        self.assertEqual(recipient_events[1]["type"], "message")
        self.assertEqual(recipient_events[1]["message"]["sender_email"], email)
        self.assertTrue("local_message_id" not in recipient_events[1])
        self.assertEqual(recipient_events[1]["message"]["recipient_id"], user_profile.recipient_id)

    def test_get_events_narrow(self) -> None:
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)

        def get_message(apply_markdown: bool, client_gravatar: bool) -> dict[str, Any]:
            result = self.tornado_call(
                get_events,
                user_profile,
                dict(
                    apply_markdown=orjson.dumps(apply_markdown).decode(),
                    client_gravatar=orjson.dumps(client_gravatar).decode(),
                    event_types=orjson.dumps(["message"]).decode(),
                    narrow=orjson.dumps([["stream", "denmark"]]).decode(),
                    user_client="website",
                    dont_block=orjson.dumps(True).decode(),
                ),
            )

            self.assert_json_success(result)
            queue_id = orjson.loads(result.content)["queue_id"]

            result = self.tornado_call(
                get_events,
                user_profile,
                {
                    "queue_id": queue_id,
                    "user_client": "website",
                    "last_event_id": -1,
                    "dont_block": orjson.dumps(True).decode(),
                },
            )
            events = orjson.loads(result.content)["events"]
            self.assert_json_success(result)
            self.assert_length(events, 0)

            self.send_personal_message(user_profile, self.example_user("othello"), "hello")
            self.send_stream_message(user_profile, "Denmark", "**hello**")

            result = self.tornado_call(
                get_events,
                user_profile,
                {
                    "queue_id": queue_id,
                    "user_client": "website",
                    "last_event_id": -1,
                    "dont_block": orjson.dumps(True).decode(),
                },
            )
            events = orjson.loads(result.content)["events"]
            self.assert_json_success(result)
            self.assert_length(events, 1)
            self.assertEqual(events[0]["type"], "message")
            return events[0]["message"]

        message = get_message(apply_markdown=False, client_gravatar=False)
        self.assertEqual(message["display_recipient"], "Denmark")
        self.assertEqual(message["content"], "**hello**")
        self.assertEqual(urlsplit(message["avatar_url"]).hostname, "secure.gravatar.com")

        message = get_message(apply_markdown=True, client_gravatar=False)
        self.assertEqual(message["display_recipient"], "Denmark")
        self.assertEqual(message["content"], "<p><strong>hello</strong></p>")
        self.assertIn("gravatar.com", message["avatar_url"])

        do_change_user_setting(
            user_profile,
            "email_address_visibility",
            UserProfile.EMAIL_ADDRESS_VISIBILITY_EVERYONE,
            acting_user=None,
        )

        message = get_message(apply_markdown=False, client_gravatar=True)
        self.assertEqual(message["display_recipient"], "Denmark")
        self.assertEqual(message["content"], "**hello**")
        self.assertEqual(message["avatar_url"], None)

        message = get_message(apply_markdown=True, client_gravatar=True)
        self.assertEqual(message["display_recipient"], "Denmark")
        self.assertEqual(message["content"], "<p><strong>hello</strong></p>")
        self.assertEqual(message["avatar_url"], None)

    def test_bogus_queue_id(self) -> None:
        user = self.example_user("hamlet")

        with self.assertRaises(BadEventQueueIdError):
            self.tornado_call(
                get_events,
                user,
                {
                    "queue_id": "hamster",
                    "user_client": "website",
                    "last_event_id": -1,
                    "dont_block": orjson.dumps(True).decode(),
                },
            )

    def test_wrong_user_queue_id(self) -> None:
        user = self.example_user("hamlet")
        wrong_user = self.example_user("othello")

        result = self.tornado_call(
            get_events,
            user,
            {
                "apply_markdown": orjson.dumps(True).decode(),
                "client_gravatar": orjson.dumps(True).decode(),
                "event_types": orjson.dumps(["message"]).decode(),
                "user_client": "website",
                "dont_block": orjson.dumps(True).decode(),
            },
        )
        self.assert_json_success(result)
        queue_id = orjson.loads(result.content)["queue_id"]

        with self.assertLogs(level="WARNING") as cm, self.assertRaises(BadEventQueueIdError):
            self.tornado_call(
                get_events,
                wrong_user,
                {
                    "queue_id": queue_id,
                    "user_client": "website",
                    "last_event_id": -1,
                    "dont_block": orjson.dumps(True).decode(),
                },
            )
        self.assertIn("not authorized for queue", cm.output[0])

    def test_get_events_custom_profile_fields(self) -> None:
        user_profile = self.example_user("iago")
        self.login_user(user_profile)
        profile_field = CustomProfileField.objects.get(realm=user_profile.realm, name="Pronouns")

        def check_pronouns_type_field_supported(
            pronouns_field_type_supported: bool, new_name: str
        ) -> None:
            clear_client_event_queues_for_testing()

            queue_data = dict(
                apply_markdown=True,
                all_public_streams=True,
                client_type_name="ZulipMobile",
                event_types=["custom_profile_fields"],
                last_connection_time=time.time(),
                queue_timeout=0,
                realm_id=user_profile.realm.id,
                user_profile_id=user_profile.id,
                pronouns_field_type_supported=pronouns_field_type_supported,
            )

            client = allocate_client_descriptor(queue_data)

            with self.captureOnCommitCallbacks(execute=True):
                try_update_realm_custom_profile_field(
                    realm=user_profile.realm, field=profile_field, name=new_name
                )

            result = self.tornado_call(
                get_events,
                user_profile,
                {
                    "queue_id": client.event_queue.id,
                    "user_client": "ZulipAndroid",
                    "last_event_id": -1,
                    "dont_block": orjson.dumps(True).decode(),
                },
            )
            events = orjson.loads(result.content)["events"]
            self.assert_json_success(result)
            self.assert_length(events, 1)

            [pronouns_field] = (
                field for field in events[0]["fields"] if field["id"] == profile_field.id
            )
            if pronouns_field_type_supported:
                expected_type = CustomProfileField.PRONOUNS
            else:
                expected_type = CustomProfileField.SHORT_TEXT
            self.assertEqual(pronouns_field["type"], expected_type)

        check_pronouns_type_field_supported(False, "Pronouns field")
        check_pronouns_type_field_supported(True, "Pronouns")


class FetchInitialStateDataTest(ZulipTestCase):
    # Non-admin users don't have access to all bots
    def test_realm_bots_non_admin(self) -> None:
        user_profile = self.example_user("cordelia")
        self.assertFalse(user_profile.is_realm_admin)
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)
        self.assert_length(result["realm_bots"], 0)

        # additionally the API key for a random bot is not present in the data
        api_key = self.notification_bot(user_profile.realm).api_key
        self.assertNotIn(api_key, str(result))

    # Admin users have access to all bots in the realm_bots field
    def test_realm_bots_admin(self) -> None:
        user_profile = self.example_user("hamlet")
        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.assertTrue(user_profile.is_realm_admin)
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)
        self.assertGreater(len(result["realm_bots"]), 2)

    def test_max_message_id_with_no_history(self) -> None:
        user_profile = self.example_user("aaron")
        # Delete all historical messages for this user
        UserMessage.objects.filter(user_profile=user_profile).delete()
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)
        self.assertEqual(result["max_message_id"], -1)

    def test_delivery_email_presence_for_non_admins(self) -> None:
        user_profile = self.example_user("aaron")
        hamlet = self.example_user("hamlet")
        self.assertFalse(user_profile.is_realm_admin)
        hamlet = self.example_user("hamlet")

        do_change_user_setting(
            hamlet,
            "email_address_visibility",
            UserProfile.EMAIL_ADDRESS_VISIBILITY_EVERYONE,
            acting_user=None,
        )
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)

        (hamlet_obj,) = (value for key, value in result["raw_users"].items() if key == hamlet.id)
        self.assertEqual(hamlet_obj["delivery_email"], hamlet.delivery_email)

        do_change_user_setting(
            hamlet,
            "email_address_visibility",
            UserProfile.EMAIL_ADDRESS_VISIBILITY_ADMINS,
            acting_user=None,
        )
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)

        (hamlet_obj,) = (value for key, value in result["raw_users"].items() if key == hamlet.id)
        self.assertIsNone(hamlet_obj["delivery_email"])

    def test_delivery_email_presence_for_admins(self) -> None:
        user_profile = self.example_user("iago")
        hamlet = self.example_user("hamlet")
        self.assertTrue(user_profile.is_realm_admin)
        hamlet = self.example_user("hamlet")

        do_change_user_setting(
            hamlet,
            "email_address_visibility",
            UserProfile.EMAIL_ADDRESS_VISIBILITY_EVERYONE,
            acting_user=None,
        )
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)

        (hamlet_obj,) = (value for key, value in result["raw_users"].items() if key == hamlet.id)
        self.assertEqual(hamlet_obj["delivery_email"], hamlet.delivery_email)

        do_change_user_setting(
            hamlet,
            "email_address_visibility",
            UserProfile.EMAIL_ADDRESS_VISIBILITY_ADMINS,
            acting_user=None,
        )
        result = fetch_initial_state_data(user_profile, realm=user_profile.realm)
        (hamlet_obj,) = (value for key, value in result["raw_users"].items() if key == hamlet.id)
        self.assertIn("delivery_email", hamlet_obj)

    def test_user_avatar_url_field_optional(self) -> None:
        hamlet = self.example_user("hamlet")
        users = [
            self.example_user("iago"),
            self.example_user("cordelia"),
            self.example_user("ZOE"),
            self.example_user("othello"),
        ]

        for user in users:
            user.long_term_idle = True
            user.save()

        long_term_idle_users_ids = [user.id for user in users]

        result = fetch_initial_state_data(
            user_profile=hamlet,
            realm=hamlet.realm,
            user_avatar_url_field_optional=True,
        )

        raw_users = result["raw_users"]

        for user_dict in raw_users.values():
            if user_dict["user_id"] in long_term_idle_users_ids:
                self.assertFalse("avatar_url" in user_dict)
            else:
                self.assertIsNotNone(user_dict["avatar_url"])

        gravatar_users_id = [
            user_dict["user_id"]
            for user_dict in raw_users.values()
            if "avatar_url" in user_dict
            and urlsplit(user_dict["avatar_url"]).hostname == "secure.gravatar.com"
        ]

        reset_email_visibility_to_everyone_in_zulip_realm()

        # Test again with client_gravatar = True
        result = fetch_initial_state_data(
            user_profile=hamlet,
            realm=hamlet.realm,
            client_gravatar=True,
