from collections.abc import Iterable
from datetime import timedelta
from email.headerregistry import Address
from typing import Any, TypeVar
from unittest import mock

import orjson
from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.contrib.sessions.models import Session
from django.core.exceptions import ValidationError
from django.test import override_settings
from django.utils.timezone import now as timezone_now

from confirmation.models import Confirmation
from corporate.lib.stripe import get_latest_seat_count
from zerver.actions.create_user import do_create_user, do_reactivate_user
from zerver.actions.invites import do_create_multiuse_invite_link, do_invite_users
from zerver.actions.message_send import RecipientInfoResult, get_recipient_info
from zerver.actions.muted_users import do_mute_user
from zerver.actions.realm_settings import (
    do_change_realm_permission_group_setting,
    do_set_realm_property,
)
from zerver.actions.user_settings import (
    bulk_regenerate_api_keys,
    do_change_full_name,
    do_change_user_setting,
)
from zerver.actions.user_topics import do_set_user_topic_visibility_policy
from zerver.actions.users import (
    change_user_is_active,
    do_change_can_change_user_emails,
    do_change_can_create_users,
    do_change_can_forge_sender,
    do_deactivate_user,
    do_delete_user,
    do_delete_user_preserving_messages,
)
from zerver.lib.avatar import avatar_url, get_avatar_field, get_gravatar_url
from zerver.lib.bulk_create import create_users
from zerver.lib.create_user import copy_default_settings
from zerver.lib.events import do_events_register
from zerver.lib.exceptions import JsonableError
from zerver.lib.send_email import clear_scheduled_emails, queue_scheduled_emails, send_future_email
from zerver.lib.stream_subscription import get_user_subscribed_streams
from zerver.lib.stream_topic import StreamTopicTarget
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    get_subscription,
    get_test_image_file,
    reset_email_visibility_to_everyone_in_zulip_realm,
    simulated_empty_cache,
)
from zerver.lib.upload import upload_avatar_image
from zerver.lib.user_groups import get_system_user_group_for_user
from zerver.lib.users import (
    Account,
    access_user_by_id,
    access_user_by_id_including_cross_realm,
    get_accounts_for_email,
    get_cross_realm_dicts,
    get_inaccessible_user_ids,
    user_ids_to_users,
)
from zerver.lib.utils import assert_is_not_none
from zerver.models import (
    CustomProfileField,
    Message,
    OnboardingStep,
    PreregistrationUser,
    RealmAuditLog,
    RealmDomain,
    RealmUserDefault,
    Recipient,
    ScheduledEmail,
    Stream,
    Subscription,
    UserGroupMembership,
    UserProfile,
    UserTopic,
)
from zerver.models.clients import get_client
from zerver.models.custom_profile_fields import check_valid_user_ids
from zerver.models.groups import NamedUserGroup, SystemGroups
from zerver.models.prereg_users import filter_to_valid_prereg_users
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import InvalidFakeEmailDomainError, get_fake_email_domain, get_realm
from zerver.models.streams import get_stream
from zerver.models.users import (
    get_source_profile,
    get_system_bot,
    get_user,
    get_user_by_delivery_email,
    get_user_by_id_in_realm_including_cross_realm,
)

K = TypeVar("K")
V = TypeVar("V")


def find_dict(lst: Iterable[dict[K, V]], k: K, v: V) -> dict[K, V]:
    for dct in lst:
        if dct[k] == v:
            return dct
    raise AssertionError(f"Cannot find element in list where key {k} == {v}")


class PermissionTest(ZulipTestCase):
    def test_role_setters(self) -> None:
        user_profile = self.example_user("hamlet")

        user_profile.is_realm_admin = True
        self.assertEqual(user_profile.is_realm_admin, True)
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)

        user_profile.is_guest = False
        self.assertEqual(user_profile.is_guest, False)
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)

        user_profile.is_realm_owner = False
        self.assertEqual(user_profile.is_realm_owner, False)
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)

        user_profile.is_moderator = False
        self.assertEqual(user_profile.is_moderator, True)
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_ADMINISTRATOR)

        user_profile.is_realm_admin = False
        self.assertEqual(user_profile.is_realm_admin, False)
        self.assertEqual(user_profile.role, UserProfile.ROLE_MEMBER)

        user_profile.is_guest = True
        self.assertEqual(user_profile.is_guest, True)
        self.assertEqual(user_profile.role, UserProfile.ROLE_GUEST)

        user_profile.is_realm_admin = False
        self.assertEqual(user_profile.is_guest, True)
        self.assertEqual(user_profile.role, UserProfile.ROLE_GUEST)

        user_profile.is_guest = False
        self.assertEqual(user_profile.is_guest, False)
        self.assertEqual(user_profile.role, UserProfile.ROLE_MEMBER)

        user_profile.is_realm_owner = True
        self.assertEqual(user_profile.is_realm_owner, True)
        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_OWNER)

        user_profile.is_realm_owner = False
        self.assertEqual(user_profile.is_realm_owner, False)
        self.assertEqual(user_profile.role, UserProfile.ROLE_MEMBER)

        user_profile.is_moderator = True
        self.assertEqual(user_profile.is_moderator, True)
        self.assertEqual(user_profile.role, UserProfile.ROLE_MODERATOR)

        user_profile.is_moderator = False
        self.assertEqual(user_profile.is_moderator, False)
        self.assertEqual(user_profile.role, UserProfile.ROLE_MEMBER)

    def test_get_admin_users(self) -> None:
        user_profile = self.example_user("hamlet")
        self.set_user_role(user_profile, UserProfile.ROLE_MEMBER)
        self.assertFalse(user_profile.is_realm_owner)
        admin_users = user_profile.realm.get_human_admin_users()
        self.assertFalse(user_profile in admin_users)
        admin_users = user_profile.realm.get_admin_users_and_bots()
        self.assertFalse(user_profile in admin_users)

        self.set_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.assertFalse(user_profile.is_realm_owner)
        admin_users = user_profile.realm.get_human_admin_users()
        self.assertTrue(user_profile in admin_users)
        admin_users = user_profile.realm.get_admin_users_and_bots()
        self.assertTrue(user_profile in admin_users)

        self.set_user_role(user_profile, UserProfile.ROLE_REALM_OWNER)
        self.assertTrue(user_profile.is_realm_owner)
        admin_users = user_profile.realm.get_human_admin_users()
        self.assertTrue(user_profile in admin_users)
        admin_users = user_profile.realm.get_human_admin_users(include_realm_owners=False)
        self.assertFalse(user_profile in admin_users)
        admin_users = user_profile.realm.get_admin_users_and_bots()
        self.assertTrue(user_profile in admin_users)
        admin_users = user_profile.realm.get_admin_users_and_bots(include_realm_owners=False)
        self.assertFalse(user_profile in admin_users)

    def test_get_first_human_user(self) -> None:
        realm = get_realm("zulip")
        UserProfile.objects.filter(realm=realm).delete()

        UserProfile.objects.create(
            realm=realm, email="bot1@zulip.com", delivery_email="bot1@zulip.com", is_bot=True
        )
        first_human_user = UserProfile.objects.create(
            realm=realm, email="user1@zulip.com", delivery_email="user1@zulip.com", is_bot=False
        )
        UserProfile.objects.create(
            realm=realm, email="user2@zulip.com", delivery_email="user2@zulip.com", is_bot=False
        )
        UserProfile.objects.create(
            realm=realm, email="bot2@zulip.com", delivery_email="bot2@zulip.com", is_bot=True
        )
        self.assertEqual(first_human_user, realm.get_first_human_user())

    def test_updating_non_existent_user(self) -> None:
        self.login("hamlet")
        admin = self.example_user("hamlet")
        self.set_user_role(admin, UserProfile.ROLE_REALM_ADMINISTRATOR)

        invalid_user_id = 1000
        result = self.client_patch(f"/json/users/{invalid_user_id}", {})
        self.assert_json_error(result, "No such user")

    def test_change_guest_user_role_with_manual_license_plan(self) -> None:
        desdemona = self.example_user("desdemona")
        polonius = self.example_user("polonius")
        self.login("desdemona")
        _, ledger = self.subscribe_realm_to_monthly_plan_on_manual_license_management(
            desdemona.realm, 5, 5
        )
        assert polonius.is_guest
        req = dict(role=UserProfile.ROLE_MEMBER)

        with self.settings(BILLING_ENABLED=True):
            result = self.client_patch(f"/json/users/{polonius.id}", req)
        self.assert_json_error(
            result,
            "Your organization does not have enough Zulip licenses to change a guest user's role.",
        )

        ledger.licenses = get_latest_seat_count(desdemona.realm) + 1
        ledger.save(update_fields=["licenses"])
        with self.settings(BILLING_ENABLED=True):
            result = self.client_patch(f"/json/users/{polonius.id}", req)
        self.assert_json_error(
            result,
            "Your organization does not have enough Zulip licenses to change a guest user's role.",
        )

        ledger.licenses_at_next_renewal = get_latest_seat_count(desdemona.realm) + 1
        ledger.save(update_fields=["licenses_at_next_renewal"])
        with self.settings(BILLING_ENABLED=True):
            result = self.client_patch(f"/json/users/{polonius.id}", req)
        self.assert_json_success(result)
        polonius.refresh_from_db()
        assert polonius.role == UserProfile.ROLE_MEMBER

    def test_owner_api(self) -> None:
        self.login("iago")

        desdemona = self.example_user("desdemona")
        othello = self.example_user("othello")
        iago = self.example_user("iago")
        realm = iago.realm

        self.set_user_role(iago, UserProfile.ROLE_REALM_OWNER)

        result = self.client_get("/json/users")
        members = self.assert_json_success(result)["members"]
        iago_dict = find_dict(members, "email", iago.email)
        self.assertTrue(iago_dict["is_owner"])
        othello_dict = find_dict(members, "email", othello.email)
        self.assertFalse(othello_dict["is_owner"])

        req = dict(role=UserProfile.ROLE_REALM_OWNER)
        with self.capture_send_event_calls(expected_num_events=6) as events:
            result = self.client_patch(f"/json/users/{othello.id}", req)
        self.assert_json_success(result)
        owner_users = realm.get_human_owner_users()
        self.assertTrue(othello in owner_users)
        person = events[0]["event"]["person"]
        self.assertEqual(person["user_id"], othello.id)
        self.assertEqual(person["role"], UserProfile.ROLE_REALM_OWNER)

        req = dict(role=UserProfile.ROLE_MEMBER)
        with self.capture_send_event_calls(expected_num_events=5) as events:
            result = self.client_patch(f"/json/users/{othello.id}", req)
        self.assert_json_success(result)
        owner_users = realm.get_human_owner_users()
        self.assertFalse(othello in owner_users)
        person = events[0]["event"]["person"]
        self.assertEqual(person["user_id"], othello.id)
        self.assertEqual(person["role"], UserProfile.ROLE_MEMBER)

        # Cannot take away from last owner
        self.login("desdemona")
        req = dict(role=UserProfile.ROLE_MEMBER)
        with self.capture_send_event_calls(expected_num_events=4) as events:
            result = self.client_patch(f"/json/users/{iago.id}", req)
        self.assert_json_success(result)
        owner_users = realm.get_human_owner_users()
        self.assertFalse(iago in owner_users)
        person = events[0]["event"]["person"]
        self.assertEqual(person["user_id"], iago.id)
        self.assertEqual(person["role"], UserProfile.ROLE_MEMBER)
        with self.capture_send_event_calls(expected_num_events=0):
            result = self.client_patch(f"/json/users/{desdemona.id}", req)
        self.assert_json_error(
            result, "The owner permission cannot be removed from the only organization owner."
        )

        self.set_user_role(iago, UserProfile.ROLE_REALM_ADMINISTRATOR)
        self.login("iago")
        with self.capture_send_event_calls(expected_num_events=0):
            result = self.client_patch(f"/json/users/{desdemona.id}", req)
        self.assert_json_error(result, "Must be an organization owner")

    def test_admin_api(self) -> None:
        self.login("desdemona")

        hamlet = self.example_user("hamlet")
        othello = self.example_user("othello")
        desdemona = self.example_user("desdemona")
        realm = hamlet.realm

        # Make sure we see is_admin flag in /json/users
        result = self.client_get("/json/users")
        members = self.assert_json_success(result)["members"]
        desdemona_dict = find_dict(members, "email", desdemona.email)
        self.assertTrue(desdemona_dict["is_admin"])
        othello_dict = find_dict(members, "email", othello.email)
        self.assertFalse(othello_dict["is_admin"])

        # Giveth
        req = dict(role=orjson.dumps(UserProfile.ROLE_REALM_ADMINISTRATOR).decode())

        with self.capture_send_event_calls(expected_num_events=6) as events:
            result = self.client_patch(f"/json/users/{othello.id}", req)
        self.assert_json_success(result)
        admin_users = realm.get_human_admin_users()
        self.assertTrue(othello in admin_users)
        person = events[0]["event"]["person"]
        self.assertEqual(person["user_id"], othello.id)
        self.assertEqual(person["role"], UserProfile.ROLE_REALM_ADMINISTRATOR)

        # Taketh away
        req = dict(role=orjson.dumps(UserProfile.ROLE_MEMBER).decode())
        with self.capture_send_event_calls(expected_num_events=5) as events:
            result = self.client_patch(f"/json/users/{othello.id}", req)
        self.assert_json_success(result)
        admin_users = realm.get_human_admin_users()
        self.assertFalse(othello in admin_users)
        person = events[0]["event"]["person"]
        self.assertEqual(person["user_id"], othello.id)
        self.assertEqual(person["role"], UserProfile.ROLE_MEMBER)

        # Make sure only admins can patch other user's info.
        self.login("othello")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Insufficient permission")

    def test_admin_api_hide_emails(self) -> None:
        reset_email_visibility_to_everyone_in_zulip_realm()

        user = self.example_user("hamlet")
        admin = self.example_user("iago")
        self.login_user(user)

        # First, verify client_gravatar works normally
        result = self.client_get("/json/users", {"client_gravatar": "true"})
        members = self.assert_json_success(result)["members"]
        hamlet = find_dict(members, "user_id", user.id)
        self.assertEqual(hamlet["email"], user.email)
        self.assertIsNone(hamlet["avatar_url"])
        self.assertEqual(hamlet["delivery_email"], user.delivery_email)

        # Also verify the /events code path.  This is a bit hacky, but
        # we need to verify client_gravatar is not being overridden.
        with mock.patch(
            "zerver.lib.events.request_event_queue", return_value=None
        ) as mock_request_event_queue:
            with self.assertRaises(JsonableError):
                do_events_register(user, user.realm, get_client("website"), client_gravatar=True)
            self.assertEqual(mock_request_event_queue.call_args_list[0][0][3], True)

        #############################################################
        # Now, switch email address visibility, check client_gravatar
        # is automatically disabled for the user.
        with self.captureOnCommitCallbacks(execute=True):
            do_change_user_setting(
                user,
                "email_address_visibility",
                UserProfile.EMAIL_ADDRESS_VISIBILITY_ADMINS,
                acting_user=None,
            )
        result = self.client_get("/json/users", {"client_gravatar": "true"})
        members = self.assert_json_success(result)["members"]
        hamlet = find_dict(members, "user_id", user.id)
        self.assertEqual(hamlet["email"], f"user{user.id}@zulip.testserver")
        self.assertEqual(
            hamlet["avatar_url"], get_gravatar_url(user.delivery_email, 1, get_realm("zulip").id)
        )

        # client_gravatar is still turned off for admins.  In theory,
        # it doesn't need to be, but client-side changes would be
        # required in apps like the mobile apps.
        # delivery_email is sent for admins.
        admin.refresh_from_db()
        user.refresh_from_db()
        self.login_user(admin)
        result = self.client_get("/json/users", {"client_gravatar": "true"})
        members = self.assert_json_success(result)["members"]
        hamlet = find_dict(members, "user_id", user.id)
        self.assertEqual(hamlet["email"], f"user{user.id}@zulip.testserver")
        self.assertEqual(
            hamlet["avatar_url"], get_gravatar_url(user.delivery_email, 1, get_realm("zulip").id)
        )
        self.assertEqual(hamlet["delivery_email"], self.example_email("hamlet"))

    def test_user_cannot_promote_to_admin(self) -> None:
        self.login("hamlet")
        req = dict(role=orjson.dumps(UserProfile.ROLE_REALM_ADMINISTRATOR).decode())
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_error(result, "Insufficient permission")

    def test_admin_user_can_change_full_name(self) -> None:
        new_name = "new name"
        self.login("iago")
        hamlet = self.example_user("hamlet")
        req = dict(full_name=new_name)
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_success(result)
        hamlet = self.example_user("hamlet")
        self.assertEqual(hamlet.full_name, new_name)

    def test_non_admin_cannot_change_full_name(self) -> None:
        self.login("hamlet")
        req = dict(full_name="new name")
        result = self.client_patch("/json/users/{}".format(self.example_user("othello").id), req)
        self.assert_json_error(result, "Insufficient permission")

    def test_admin_cannot_set_long_full_name(self) -> None:
        new_name = "a" * (UserProfile.MAX_NAME_LENGTH + 1)
        self.login("iago")
        req = dict(full_name=new_name)
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_error(result, "Name too long!")

    def test_admin_cannot_set_empty_full_name(self) -> None:
        self.login("iago")
        req = dict(full_name="")
        # Empty name is treated as a no-op
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_success(result)
        iago = self.example_user("iago")
        self.assertEqual(iago.full_name, "Iago")

    def test_not_allowed_format(self) -> None:
        # Name of format "Alice|999" breaks in Markdown
        new_name = "iago|72"
        self.login("iago")
        req = dict(full_name=new_name)
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_error(result, "Invalid format!")

    def test_allowed_format_complex(self) -> None:
        # Adding characters after r'|d+' doesn't break Markdown
        new_name = "Hello- 12iago|72k"
        self.login("iago")
        req = dict(full_name=new_name)
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_success(result)

    def test_require_unique_names(self) -> None:
        self.login("desdemona")
        iago = self.example_user("iago")
        hamlet = self.example_user("hamlet")

        do_set_realm_property(hamlet.realm, "require_unique_names", True, acting_user=None)
        req = dict(full_name="IaGo")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Unique names required in this organization.")

        req = dict(full_name="𝕚𝕒𝕘𝕠")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Unique names required in this organization.")

        req = dict(full_name="ｉａｇｏ")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Unique names required in this organization.")

        req = dict(full_name="𝒾𝒶𝑔𝑜")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Unique names required in this organization.")

        # check for uniqueness including imported users
        iago.is_mirror_dummy = True
        req = dict(full_name="iago")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Unique names required in this organization.")

        # check for uniqueness including deactivated users
        do_deactivate_user(iago, acting_user=None)
        req = dict(full_name="iago")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_error(result, "Unique names required in this organization.")

        do_set_realm_property(hamlet.realm, "require_unique_names", False, acting_user=None)
        req = dict(full_name="iago")
        result = self.client_patch(f"/json/users/{hamlet.id}", req)
        self.assert_json_success(result)

    def test_not_allowed_format_complex(self) -> None:
        new_name = "Hello- 12iago|72"
        self.login("iago")
        req = dict(full_name=new_name)
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_error(result, "Invalid format!")

    def test_invalid_role(self) -> None:
        self.login("iago")
        req = dict(role=1000)
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_error(
            result, "Invalid role: Value error, Not in the list of possible values"
        )

    def test_admin_cannot_set_full_name_with_invalid_characters(self) -> None:
        new_name = "Opheli*"
        self.login("iago")
        req = dict(full_name=new_name)
        result = self.client_patch("/json/users/{}".format(self.example_user("hamlet").id), req)
        self.assert_json_error(result, "Invalid characters in name!")

    def test_access_user_by_id(self) -> None:
        iago = self.example_user("iago")
        internal_realm = get_realm(settings.SYSTEM_BOT_REALM)

        # Must be a valid user ID in the realm
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, 1234, for_admin=False)
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(iago, 1234, for_admin=False)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, self.mit_user("sipbtest").id, for_admin=False)
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(
                iago, self.mit_user("sipbtest").id, for_admin=False
            )

        # Can only access bot users if allow_bots is passed
        bot = self.example_user("default_bot")
        access_user_by_id(iago, bot.id, allow_bots=True, for_admin=True)
        access_user_by_id_including_cross_realm(iago, bot.id, allow_bots=True, for_admin=True)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, bot.id, for_admin=True)
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(iago, bot.id, for_admin=True)

        # Only the including_cross_realm variant works for system bots.
        system_bot = get_system_bot(settings.WELCOME_BOT, internal_realm.id)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, system_bot.id, allow_bots=True, for_admin=False)
        access_user_by_id_including_cross_realm(
            iago, system_bot.id, allow_bots=True, for_admin=False
        )
        # And even then, only if `allow_bots` was passed.
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, system_bot.id, for_admin=False)
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(iago, system_bot.id, for_admin=False)

        # Can only access deactivated users if allow_deactivated is passed
        hamlet = self.example_user("hamlet")
        do_deactivate_user(hamlet, acting_user=None)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, hamlet.id, for_admin=False)
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(iago, hamlet.id, for_admin=False)

        with self.assertRaises(JsonableError):
            access_user_by_id(iago, hamlet.id, for_admin=True)
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(iago, hamlet.id, for_admin=True)
        access_user_by_id(iago, hamlet.id, allow_deactivated=True, for_admin=True)
        access_user_by_id_including_cross_realm(
            iago, hamlet.id, allow_deactivated=True, for_admin=True
        )

        # Non-admin user can't admin another user
        with self.assertRaises(JsonableError):
            access_user_by_id(
                self.example_user("cordelia"), self.example_user("aaron").id, for_admin=True
            )
        with self.assertRaises(JsonableError):
            access_user_by_id_including_cross_realm(
                self.example_user("cordelia"), self.example_user("aaron").id, for_admin=True
            )

        # But does have read-only access to it.
        access_user_by_id(
            self.example_user("cordelia"), self.example_user("aaron").id, for_admin=False
        )
        access_user_by_id_including_cross_realm(
            self.example_user("cordelia"), self.example_user("aaron").id, for_admin=False
        )

    def test_access_user_by_id_when_personal_recipient_is_none(self) -> None:
        self.set_up_db_for_testing_user_access()
        polonius = self.example_user("polonius")

        # Removing the personal recipient to ensure a new direct message group is used for 1:1 dms.
        polonius.recipient = None
        polonius.save()

        # Restricting the "Members" system group to not allow access to all users.
        realm = get_realm("zulip")
        members_system_group = NamedUserGroup.objects.get(name=SystemGroups.MEMBERS, realm=realm)
        do_change_realm_permission_group_setting(
            realm, "can_access_all_users_group", members_system_group, acting_user=None
        )

        aaron = self.example_user("aaron")
        target_user = access_user_by_id(polonius, aaron.id, for_admin=False)
        self.assertEqual(target_user, aaron)

        othello = self.example_user("othello")
        with self.assertRaises(JsonableError):
            access_user_by_id(polonius, othello.id, for_admin=False)

    def check_property_for_role(self, user_profile: UserProfile, role: int) -> bool:
        if role == UserProfile.ROLE_REALM_ADMINISTRATOR:
            return (
                user_profile.is_realm_admin
                and not user_profile.is_guest
                and not user_profile.is_realm_owner
            )
        elif role == UserProfile.ROLE_REALM_OWNER:
            return (
                user_profile.is_realm_owner
                and user_profile.is_realm_admin
                and not user_profile.is_guest
            )
        elif role == UserProfile.ROLE_MODERATOR:
            return user_profile.is_moderator or (
                user_profile.is_realm_admin and not user_profile.is_guest
            )

        if role == UserProfile.ROLE_MEMBER:
            return (
                not user_profile.is_guest
                and not user_profile.is_moderator
                and not user_profile.is_realm_admin
                and not user_profile.is_realm_owner
            )

        assert role == UserProfile.ROLE_GUEST
        return (
            user_profile.is_guest
            and not user_profile.is_moderator
            and not user_profile.is_realm_admin
            and not user_profile.is_realm_owner
        )

    def check_user_role_change(
        self,
        user_email: str,
        new_role: int,
    ) -> None:
        self.login("desdemona")

        user_profile = self.example_user(user_email)
        old_role = user_profile.role
        old_system_group = get_system_user_group_for_user(user_profile)

        self.assertTrue(self.check_property_for_role(user_profile, old_role))
        self.assertTrue(
            UserGroupMembership.objects.filter(
                user_profile=user_profile, user_group=old_system_group
            ).exists()
        )

        req = dict(role=orjson.dumps(new_role).decode())

        # The basic events sent in all cases on changing role are - one event
        # for changing role and one event each for adding and removing user
        # from system user group.
        num_events = 3

        if UserProfile.ROLE_MEMBER in [old_role, new_role]:
            # There is one additional event for adding/removing user from
            # the "Full members" group as well.
            num_events += 1

        if new_role == UserProfile.ROLE_GUEST:
            # There is one additional event deleting the unsubscribed public
            # streams that the user will not able to access after becoming guest.
            num_events += 1

        if old_role == UserProfile.ROLE_GUEST:
            # User will receive one event for creation of unsubscribed public
            # (and private, if the new role is owner or admin) streams that
            # they did not have access to previously and 3 more peer_add
            # events for each of the public stream.
            if new_role in [UserProfile.ROLE_REALM_ADMINISTRATOR, UserProfile.ROLE_REALM_OWNER]:
                # If the new role is owner or admin, the peer_add event will be
                # sent for one private stream as well.
                num_events += 7
            else:
                num_events += 6
        elif new_role in [
            UserProfile.ROLE_REALM_ADMINISTRATOR,
            UserProfile.ROLE_REALM_OWNER,
        ] and old_role not in [UserProfile.ROLE_REALM_ADMINISTRATOR, UserProfile.ROLE_REALM_OWNER]:
            # If old_role is not guest and user's role is changed to admin or owner from moderator
            # or member, then the user gains access to unsubscribed private streams and thus
            # receives one event for stream creation and one peer_add event for it.
            num_events += 2

        with self.capture_send_event_calls(expected_num_events=num_events) as events:
            result = self.client_patch(f"/json/users/{user_profile.id}", req)
        self.assert_json_success(result)

        user_profile = self.example_user(user_email)
        self.assertTrue(self.check_property_for_role(user_profile, new_role))
        system_group = get_system_user_group_for_user(user_profile)
        self.assertTrue(
            UserGroupMembership.objects.filter(
                user_profile=user_profile, user_group=system_group
            ).exists()
        )

        person = events[0]["event"]["person"]
        self.assertEqual(person["user_id"], user_profile.id)
        self.assertTrue(person["role"], new_role)

    def test_change_regular_member_to_guest(self) -> None:
        self.check_user_role_change("hamlet", UserProfile.ROLE_GUEST)

    def test_change_guest_to_regular_member(self) -> None:
        self.check_user_role_change("polonius", UserProfile.ROLE_MEMBER)

    def test_change_admin_to_guest(self) -> None:
        self.check_user_role_change("iago", UserProfile.ROLE_GUEST)

    def test_change_guest_to_admin(self) -> None:
        self.check_user_role_change("polonius", UserProfile.ROLE_REALM_ADMINISTRATOR)

