import re
import time
from collections.abc import Sequence
from datetime import timedelta
from typing import TYPE_CHECKING, Any, Union
from unittest.mock import MagicMock, patch
from urllib.parse import quote, urlencode, urlsplit

import orjson
from django.conf import settings
from django.contrib.auth.views import PasswordResetConfirmView
from django.contrib.contenttypes.models import ContentType
from django.db.utils import IntegrityError
from django.http import HttpResponse, HttpResponseBase
from django.template.response import TemplateResponse
from django.test import Client, override_settings
from django.utils import translation

from confirmation.models import Confirmation, one_click_unsubscribe_link
from zerver.actions.create_realm import do_change_realm_subdomain, do_create_realm
from zerver.actions.create_user import add_new_user_history, do_create_user
from zerver.actions.default_streams import do_add_default_stream, do_create_default_stream_group
from zerver.actions.invites import do_invite_users
from zerver.actions.message_send import internal_send_private_message
from zerver.actions.realm_settings import (
    do_deactivate_realm,
    do_scrub_realm,
    do_set_realm_authentication_methods,
    do_set_realm_property,
    do_set_realm_user_default_setting,
)
from zerver.actions.users import change_user_is_active, do_deactivate_user
from zerver.decorator import do_two_factor_login
from zerver.forms import HomepageForm
from zerver.lib.default_streams import get_slim_realm_default_streams
from zerver.lib.email_notifications import enqueue_welcome_emails
from zerver.lib.i18n import get_default_language_for_new_user
from zerver.lib.initial_password import initial_password
from zerver.lib.mobile_auth_otp import (
    ascii_to_hex,
    hex_to_ascii,
    is_valid_otp,
    otp_decrypt_api_key,
    otp_encrypt_api_key,
    xor_hex_strings,
)
from zerver.lib.name_restrictions import is_disposable_domain
from zerver.lib.send_email import EmailNotDeliveredError, FromAddress, send_future_email
from zerver.lib.stream_subscription import (
    get_stream_subscriptions_for_user,
    get_user_subscribed_streams,
)
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    HostRequestMock,
    avatar_disk_path,
    find_key_by_email,
    get_test_image_file,
    load_subdomain_token,
    message_stream_count,
    most_recent_message,
    most_recent_usermessage,
    ratelimit_rule,
    reset_email_visibility_to_everyone_in_zulip_realm,
)
from zerver.models import (
    CustomProfileField,
    CustomProfileFieldValue,
    DefaultStream,
    Message,
    OnboardingStep,
    OnboardingUserMessage,
    PreregistrationUser,
    Realm,
    RealmUserDefault,
    Recipient,
    ScheduledEmail,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
)
from zerver.models.realms import get_realm
from zerver.models.recipients import get_direct_message_group_user_ids
from zerver.models.streams import get_stream
from zerver.models.users import get_system_bot, get_user, get_user_by_delivery_email
from zerver.views.auth import redirect_and_log_into_subdomain, start_two_factor_auth
from zerver.views.development.registration import confirmation_key
from zproject.backends import ExternalAuthDataDict, ExternalAuthResult, email_auth_enabled

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


class RedirectAndLogIntoSubdomainTestCase(ZulipTestCase):
    def test_data(self) -> None:
        realm = get_realm("zulip")
        user_profile = self.example_user("hamlet")
        name = user_profile.full_name
        email = user_profile.delivery_email
        response = redirect_and_log_into_subdomain(ExternalAuthResult(user_profile=user_profile))
        data = load_subdomain_token(response)
        self.assertDictEqual(
            data,
            {"full_name": name, "email": email, "subdomain": realm.subdomain, "is_signup": False},
        )

        data_dict = ExternalAuthDataDict(is_signup=True, multiuse_object_key="key")
        response = redirect_and_log_into_subdomain(
            ExternalAuthResult(user_profile=user_profile, data_dict=data_dict)
        )
        data = load_subdomain_token(response)
        self.assertDictEqual(
            data,
            {
                "full_name": name,
                "email": email,
                "subdomain": realm.subdomain,
                # the email has an account at the subdomain,
                # so is_signup get overridden to False:
                "is_signup": False,
                "multiuse_object_key": "key",
            },
        )

        data_dict = ExternalAuthDataDict(
            email=self.nonreg_email("alice"),
            full_name="Alice",
            subdomain=realm.subdomain,
            is_signup=True,
            full_name_validated=True,
            multiuse_object_key="key",
        )
        response = redirect_and_log_into_subdomain(ExternalAuthResult(data_dict=data_dict))
        data = load_subdomain_token(response)
        self.assertDictEqual(
            data,
            {
                "full_name": "Alice",
                "email": self.nonreg_email("alice"),
                "full_name_validated": True,
                "subdomain": realm.subdomain,
                "is_signup": True,
                "multiuse_object_key": "key",
            },
        )


class DeactivationNoticeTestCase(ZulipTestCase):
    def test_redirection_for_deactivated_realm(self) -> None:
        realm = get_realm("zulip")
        realm.deactivated = True
        realm.save(update_fields=["deactivated"])

        for url in ("/register/", "/login/"):
            result = self.client_get(url)
            self.assertRedirects(result, "/accounts/deactivated/", status_code=302)

    def test_redirection_for_active_realm(self) -> None:
        for url in ("/register/", "/login/"):
            result = self.client_get(url)
            self.assertEqual(result.status_code, 200)

    def test_deactivation_notice_when_realm_is_active(self) -> None:
        result = self.client_get("/accounts/deactivated/")
        self.assertRedirects(result, "/devlogin/", status_code=302)

    def test_deactivation_notice_when_deactivated(self) -> None:
        realm = get_realm("zulip")
        realm.deactivated = True
        realm.save(update_fields=["deactivated"])

        result = self.client_get("/login/", follow=True)
        self.assertEqual(result.redirect_chain[-1], ("/accounts/deactivated/", 302))
        self.assertIn("This organization has been deactivated.", result.content.decode())
        self.assertNotIn("and all organization data has been deleted", result.content.decode())

    def test_deactivation_notice_when_deactivated_and_deactivated_redirect_is_set_to_different_domain(
        self,
    ) -> None:
        realm = get_realm("zulip")
        realm.deactivated = True
        realm.deactivated_redirect = f"http://example.not_{settings.EXTERNAL_HOST}.com:9991"
        realm.save(update_fields=["deactivated", "deactivated_redirect"])

        result = self.client_get("/login/", follow=True)
        self.assert_in_success_response([f'href="{realm.deactivated_redirect}"'], result)

    def test_deactivation_notice_when_realm_subdomain_is_changed(self) -> None:
        realm = get_realm("zulip")
        do_change_realm_subdomain(realm, "new-subdomain-name", acting_user=None)

        result = self.client_get("/login/", follow=True)
        self.assert_in_success_response(
            [
                f'href="http://new-subdomain-name.{settings.EXTERNAL_HOST}/" id="deactivated-org-auto-redirect"'
            ],
            result,
        )

    def test_no_deactivation_notice_with_no_redirect(self) -> None:
        realm = get_realm("zulip")
        do_change_realm_subdomain(
            realm, "new-subdomain-name", acting_user=None, add_deactivated_redirect=False
        )

        result = self.client_get("/login/", follow=True)
        self.assertEqual(result.status_code, 404)
        self.assertNotIn(
            "new-subdomain-name",
            result.content.decode(),
        )

    def test_deactivated_redirect_field_of_placeholder_realms_are_modified_on_changing_subdomain_multiple_times(
        self,
    ) -> None:
        realm = get_realm("zulip")
        do_change_realm_subdomain(realm, "new-name-1", acting_user=None)

        result = self.client_get("/login/", follow=True)
        self.assert_in_success_response(
            ['href="http://new-name-1.testserver/" id="deactivated-org-auto-redirect"'], result
        )

        realm = get_realm("new-name-1")
        do_change_realm_subdomain(realm, "new-name-2", acting_user=None)
        result = self.client_get("/login/", follow=True)
        self.assert_in_success_response(
            ['href="http://new-name-2.testserver/" id="deactivated-org-auto-redirect"'], result
        )

    def test_deactivation_notice_when_deactivated_and_scrubbed(self) -> None:
        # We expect system bot messages when scrubbing a realm.
        internal_realm = get_realm(settings.SYSTEM_BOT_REALM)
        notification_bot = get_system_bot(settings.NOTIFICATION_BOT, internal_realm.id)
        hamlet = self.example_user("hamlet")
        internal_send_private_message(notification_bot, hamlet, "test")
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm,
            acting_user=None,
            deactivation_reason="owner_request",
            email_owners=False,
        )
        realm.refresh_from_db()
        assert realm.deactivated
        assert realm.deactivated_redirect is None
        with self.assertLogs(level="WARNING"):
            do_scrub_realm(realm, acting_user=None)

        result = self.client_get("/login/", follow=True)
        self.assertEqual(result.redirect_chain[-1], ("/accounts/deactivated/", 302))
        self.assertIn("This organization has been deactivated,", result.content.decode())
        self.assertIn("and all organization data has been deleted", result.content.decode())


class AddNewUserHistoryTest(ZulipTestCase):
    def test_add_new_user_history_race(self) -> None:
        """Sends a message during user creation"""
        # Create a user who hasn't had historical messages added
        realm = get_realm("zulip")
        stream = Stream.objects.get(realm=realm, name="Denmark")
        DefaultStream.objects.create(stream=stream, realm=realm)
        # Make sure at least 3 messages are sent to Denmark and it's a default stream.
        message_id = self.send_stream_message(self.example_user("hamlet"), stream.name, "test 1")
        self.send_stream_message(self.example_user("hamlet"), stream.name, "test 2")
        self.send_stream_message(self.example_user("hamlet"), stream.name, "test 3")

        with patch("zerver.actions.create_user.add_new_user_history"):
            self.register(self.nonreg_email("test"), "test")
        user_profile = self.nonreg_user("test")
        subs = Subscription.objects.select_related("recipient").filter(
            user_profile=user_profile, recipient__type=Recipient.STREAM
        )
        streams = Stream.objects.filter(id__in=[sub.recipient.type_id for sub in subs])

        # Sent a message afterwards to trigger a race between message
        # sending and `add_new_user_history`.
        race_message_id = self.send_stream_message(
            self.example_user("hamlet"), streams[0].name, "test"
        )

        # Overwrite MAX_NUM_RECENT_UNREAD_MESSAGES to 2
        MAX_NUM_RECENT_UNREAD_MESSAGES = 2
        with patch(
            "zerver.actions.create_user.MAX_NUM_RECENT_UNREAD_MESSAGES",
            MAX_NUM_RECENT_UNREAD_MESSAGES,
        ):
            add_new_user_history(user_profile, streams)

        # Our first message is in the user's history
        self.assertTrue(
            UserMessage.objects.filter(user_profile=user_profile, message_id=message_id).exists()
        )
        # The race message is in the user's history, marked unread & NOT historical.
        self.assertTrue(
            UserMessage.objects.filter(
                user_profile=user_profile, message_id=race_message_id
            ).exists()
        )
        self.assertFalse(
            UserMessage.objects.get(
                user_profile=user_profile, message_id=race_message_id
            ).flags.read.is_set
        )
        self.assertFalse(
            UserMessage.objects.get(
                user_profile=user_profile, message_id=race_message_id
            ).flags.historical.is_set
        )

        # Verify that the MAX_NUM_RECENT_UNREAD_MESSAGES latest messages
        # that weren't the race message are marked as unread & historical.
        latest_messages = (
            UserMessage.objects.filter(
                user_profile=user_profile,
                message__recipient__type=Recipient.STREAM,
            )
            .exclude(message_id=race_message_id)
            .order_by("-message_id")[0:MAX_NUM_RECENT_UNREAD_MESSAGES]
        )
        self.assert_length(latest_messages, 2)
        for msg in latest_messages:
            self.assertFalse(msg.flags.read.is_set)
            self.assertTrue(msg.flags.historical.is_set)

        # Verify that older messages are correctly marked as read & historical.
        older_messages = (
            UserMessage.objects.filter(
                user_profile=user_profile,
                message__recipient__type=Recipient.STREAM,
            )
            .exclude(message_id=race_message_id)
            .order_by("-message_id")[
                MAX_NUM_RECENT_UNREAD_MESSAGES : MAX_NUM_RECENT_UNREAD_MESSAGES + 1
            ]
        )
        self.assertGreater(len(older_messages), 0)
        for msg in older_messages:
            self.assertTrue(msg.flags.read.is_set)
            self.assertTrue(msg.flags.historical.is_set)

    def test_only_tracked_onboarding_messages_marked_unread(self) -> None:
        """
        Realms with tracked onboarding messages have only
        those messages marked as unread.
        """
        realm = do_create_realm("realm_string_id", "realm name")
        hamlet = do_create_user(
            "hamlet", "password", realm, "hamlet", realm_creation=True, acting_user=None
        )
        stream = Stream.objects.get(realm=realm, name=str(realm.ZULIP_SANDBOX_CHANNEL_NAME))

        # Onboarding messages sent during realm creation are tracked.
        self.assertTrue(OnboardingUserMessage.objects.filter(realm=realm).exists())
        onboarding_message_ids = OnboardingUserMessage.objects.filter(realm=realm).values_list(
            "message_id", flat=True
        )

        # Other messages sent before a new user joins.
        for i in range(3):
            self.send_stream_message(hamlet, stream.name, f"test {i}")

        new_user = do_create_user("new_user", "password", realm, "new_user", acting_user=None)

        # The onboarding messages are in the user's history and marked unread.
        onboarding_user_messages = UserMessage.objects.filter(
            user_profile=new_user, message_id__in=onboarding_message_ids
        )
        for user_message in onboarding_user_messages:
            self.assertFalse(user_message.flags.read.is_set)
            self.assertTrue(user_message.flags.historical.is_set)

        # Other messages are in user's history and marked as read.
        other_user_messages = UserMessage.objects.filter(
            user_profile=new_user, message__recipient__type=Recipient.STREAM
        ).exclude(message_id__in=onboarding_message_ids)
        self.assertTrue(other_user_messages.exists())
        for user_message in other_user_messages:
            self.assertTrue(user_message.flags.read.is_set)
            self.assertTrue(user_message.flags.historical.is_set)

        # Onboarding messages for hamlet (realm creator) are not
        # marked as historical.
        onboarding_user_messages = UserMessage.objects.filter(
            user_profile=hamlet, message_id__in=onboarding_message_ids
        )
        for user_message in onboarding_user_messages:
            self.assertFalse(user_message.flags.read.is_set)
            self.assertFalse(user_message.flags.historical.is_set)

    def test_tracked_onboarding_topics_first_messages_marked_starred(self) -> None:
        """
        Realms with tracked onboarding messages have only
        first message in each onboarding topic marked as starred.
        """
        realm = do_create_realm("realm_string_id", "realm name")
        hamlet = do_create_user(
            "hamlet", "password", realm, "hamlet", realm_creation=True, acting_user=None
        )

        # Onboarding messages sent during realm creation are tracked.
        self.assertTrue(OnboardingUserMessage.objects.filter(realm=realm).exists())

        seen_topics = set()
        onboarding_topics_first_message_ids = set()
        onboarding_messages = Message.objects.filter(
            realm=realm, recipient__type=Recipient.STREAM
        ).order_by("id")
        for message in onboarding_messages:
            topic_name = message.topic_name()
            if topic_name not in seen_topics:
                onboarding_topics_first_message_ids.add(message.id)
                seen_topics.add(topic_name)

        # The first onboarding message in each topic are in the
        # user's history and marked starred.
        onboarding_user_messages = UserMessage.objects.filter(
            user_profile=hamlet, message_id__in=onboarding_topics_first_message_ids
        )
        for user_message in onboarding_user_messages:
            self.assertTrue(user_message.flags.starred.is_set)
            self.assertFalse(user_message.flags.read.is_set)
            self.assertFalse(user_message.flags.historical.is_set)

        # Other messages are in user's history but not marked starred.
        other_user_messages = UserMessage.objects.filter(
            user_profile=hamlet, message__recipient__type=Recipient.STREAM
        ).exclude(message_id__in=onboarding_topics_first_message_ids)
        self.assertTrue(other_user_messages.exists())
        for user_message in other_user_messages:
            self.assertFalse(user_message.flags.starred.is_set)
            self.assertFalse(user_message.flags.read.is_set)
            self.assertFalse(user_message.flags.historical.is_set)

        # Initial DM sent by welcome bot is also starred.
        initial_direct_user_message = UserMessage.objects.get(
            user_profile=hamlet, message__recipient__type=Recipient.PERSONAL
        )
        self.assertTrue(initial_direct_user_message.flags.starred.is_set)

    def test_auto_subbed_to_personals(self) -> None:
        """
        Newly created users are auto-subbed to the ability to receive
        personals.
        """
        test_email = self.nonreg_email("test")
        self.register(test_email, "test")
        user_profile = self.nonreg_user("test")
        old_messages_count = message_stream_count(user_profile)
        self.send_personal_message(user_profile, user_profile)
        new_messages_count = message_stream_count(user_profile)
        self.assertEqual(new_messages_count, old_messages_count + 1)

        recipient = Recipient.objects.get(type_id=user_profile.id, type=Recipient.PERSONAL)
        message = most_recent_message(user_profile)
        self.assertEqual(message.recipient, recipient)

        with patch("zerver.models.Recipient.label", return_value="recip"):
            self.assertEqual(
                repr(message),
                f"<Message: recip /  / <UserProfile: {user_profile.email} {user_profile.realm!r}>>",
            )

            user_message = most_recent_usermessage(user_profile)
            self.assertEqual(
                repr(user_message),
                f"<UserMessage: recip / {user_profile.email} (['read'])>",
            )


class InitialPasswordTest(ZulipTestCase):
    def test_none_initial_password_salt(self) -> None:
        with self.settings(INITIAL_PASSWORD_SALT=None):
            self.assertIsNone(initial_password("test@test.com"))


class PasswordResetTest(ZulipTestCase):
    """
    Log in, reset password, log out, log in with new password.
    """

    def get_reset_mail_body(self, subdomain: str = "zulip") -> str:
        from django.core.mail import outbox

        [message] = outbox
        self.assertEqual(self.email_envelope_from(message), settings.NOREPLY_EMAIL_ADDRESS)
        self.assertRegex(
            self.email_display_from(message),
            # The email might be sent in different languages for i18n testing
            rf"^testserver .* <{self.TOKENIZED_NOREPLY_REGEX}>\Z",
        )
        self.assertIn(f"{subdomain}.testserver", message.extra_headers["List-Id"])

        return str(message.body)

    def test_password_reset(self) -> None:
        user = self.example_user("hamlet")
        email = user.delivery_email
        old_password = initial_password(email)
        assert old_password is not None

        self.login_user(user)

        # test password reset template
        result = self.client_get("/accounts/password/reset/")
        self.assert_in_response("Reset your password", result)

        # start the password reset process by supplying an email address
        result = self.client_post("/accounts/password/reset/", {"email": email})

        # check the redirect link telling you to check mail for password reset link
        self.assertEqual(result.status_code, 302)
        self.assertTrue(result["Location"].endswith("/accounts/password/reset/done/"))
        result = self.client_get(result["Location"])

        self.assert_in_response("Check your email in a few minutes to finish the process.", result)

        # Check that the password reset email is from a noreply address.
        body = self.get_reset_mail_body()
        self.assertIn("reset your password", body)

        # Visit the password reset link.
        password_reset_url = self.get_confirmation_url_from_outbox(
            email, url_pattern=settings.EXTERNAL_HOST + r"(\S\S+)"
        )
        result = self.client_get(password_reset_url)
        self.assertEqual(result.status_code, 302)
        self.assertTrue(
            result["Location"].endswith(f"/{PasswordResetConfirmView.reset_url_token}/")
        )

        final_reset_url = result["Location"]
        result = self.client_get(final_reset_url)
        self.assertEqual(result.status_code, 200)

        # Reset your password
        with self.settings(PASSWORD_MIN_LENGTH=3, PASSWORD_MIN_GUESSES=1000):
            # Verify weak passwords don't work.
            result = self.client_post(
                final_reset_url, {"new_password1": "easy", "new_password2": "easy"}
            )
            self.assert_in_response("The password is too weak.", result)

            result = self.client_post(
                final_reset_url, {"new_password1": "f657gdGGk9", "new_password2": "f657gdGGk9"}
            )
            # password reset succeeded
            self.assertEqual(result.status_code, 302)
            self.assertTrue(result["Location"].endswith("/password/done/"))

            # log back in with new password
            self.login_by_email(email, password="f657gdGGk9")
            user_profile = self.example_user("hamlet")
            self.assert_logged_in_user_id(user_profile.id)

            # make sure old password no longer works
            self.assert_login_failure(email, password=old_password)

    @patch("django.http.HttpRequest.get_host")
    def test_password_reset_page_redirects_for_root_alias_when_root_domain_landing_page_is_enabled(
        self, mock_get_host: MagicMock
    ) -> None:
        mock_get_host.return_value = "alias.testserver"
        with self.settings(ROOT_DOMAIN_LANDING_PAGE=True, ROOT_SUBDOMAIN_ALIASES=["alias"]):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 302)
            self.assertEqual(
                result["Location"], "/accounts/go/?next=%2Faccounts%2Fpassword%2Freset%2F"
            )

        mock_get_host.return_value = "www.testserver"
        with self.settings(
            ROOT_DOMAIN_LANDING_PAGE=True,
        ):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 302)
            self.assertEqual(
                result["Location"], "/accounts/go/?next=%2Faccounts%2Fpassword%2Freset%2F"
            )

    @patch("django.http.HttpRequest.get_host")
    def test_password_reset_page_redirects_for_root_domain_when_root_domain_landing_page_is_enabled(
        self, mock_get_host: MagicMock
    ) -> None:
        mock_get_host.return_value = "testserver"
        with self.settings(ROOT_DOMAIN_LANDING_PAGE=True):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 302)
            self.assertEqual(
                result["Location"], "/accounts/go/?next=%2Faccounts%2Fpassword%2Freset%2F"
            )

        mock_get_host.return_value = "www.zulip.example.com"
        with self.settings(
            ROOT_DOMAIN_LANDING_PAGE=True,
            EXTERNAL_HOST="www.zulip.example.com",
        ):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 302)
            self.assertEqual(
                result["Location"], "/accounts/go/?next=%2Faccounts%2Fpassword%2Freset%2F"
            )

    @patch("django.http.HttpRequest.get_host")
    def test_password_reset_page_works_for_root_alias_when_root_domain_landing_page_is_not_enabled(
        self, mock_get_host: MagicMock
    ) -> None:
        mock_get_host.return_value = "alias.testserver"
        with self.settings(ROOT_SUBDOMAIN_ALIASES=["alias"]):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 200)

        mock_get_host.return_value = "www.testserver"
        result = self.client_get("/accounts/password/reset/")
        self.assertEqual(result.status_code, 200)

    @patch("django.http.HttpRequest.get_host")
    def test_password_reset_page_works_for_root_domain_when_root_domain_landing_page_is_not_enabled(
        self, mock_get_host: MagicMock
    ) -> None:
        mock_get_host.return_value = "testserver"
        result = self.client_get("/accounts/password/reset/")
        self.assertEqual(result.status_code, 200)

        mock_get_host.return_value = "www.zulip.example.com"
        with self.settings(EXTERNAL_HOST="www.zulip.example.com", ROOT_SUBDOMAIN_ALIASES=[]):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 200)

    @patch("django.http.HttpRequest.get_host")
    def test_password_reset_page_works_always_for_subdomains(
        self, mock_get_host: MagicMock
    ) -> None:
        mock_get_host.return_value = "lear.testserver"
        with self.settings(ROOT_DOMAIN_LANDING_PAGE=True):
            result = self.client_get("/accounts/password/reset/")
            self.assertEqual(result.status_code, 200)

        result = self.client_get("/accounts/password/reset/")
        self.assertEqual(result.status_code, 200)

    def test_password_reset_for_non_existent_user(self) -> None:
        email = "nonexisting@mars.com"

        # start the password reset process by supplying an email address
        result = self.client_post("/accounts/password/reset/", {"email": email})

        # check the redirect link telling you to check mail for password reset link
        self.assertEqual(result.status_code, 302)
        self.assertTrue(result["Location"].endswith("/accounts/password/reset/done/"))
        result = self.client_get(result["Location"])

        self.assert_in_response("Check your email in a few minutes to finish the process.", result)

        # Check that the password reset email is from a noreply address.
        body = self.get_reset_mail_body()
        self.assertIn("Somebody (possibly you) requested a new password", body)
        self.assertIn("You do not have an account", body)
        self.assertIn("safely ignore", body)
        self.assertNotIn("reset your password", body)
        self.assertNotIn("deactivated", body)

    def test_password_reset_for_deactivated_user(self) -> None:
        user_profile = self.example_user("hamlet")
        email = user_profile.delivery_email
        do_deactivate_user(user_profile, acting_user=None)

        # start the password reset process by supplying an email address
        result = self.client_post("/accounts/password/reset/", {"email": email})

        # check the redirect link telling you to check mail for password reset link
        self.assertEqual(result.status_code, 302)
        self.assertTrue(result["Location"].endswith("/accounts/password/reset/done/"))
        result = self.client_get(result["Location"])

        self.assert_in_response("Check your email in a few minutes to finish the process.", result)

        # Check that the password reset email is from a noreply address.
        body = self.get_reset_mail_body()
        self.assertIn("Somebody (possibly you) requested a new password", body)
        self.assertIn("has been deactivated", body)
        self.assertIn("safely ignore", body)
        self.assertNotIn("reset your password", body)
        self.assertNotIn("not have an account", body)

    def test_password_reset_with_deactivated_realm(self) -> None:
        user_profile = self.example_user("hamlet")
        email = user_profile.delivery_email
        do_deactivate_realm(
            user_profile.realm,
            acting_user=None,
            deactivation_reason="owner_request",
            email_owners=False,
        )

        # start the password reset process by supplying an email address
        with self.assertLogs(level="INFO") as m:
            result = self.client_post("/accounts/password/reset/", {"email": email})
            self.assertEqual(m.output, ["INFO:root:Realm is deactivated"])

        # check the redirect link telling you to check mail for password reset link
        self.assertEqual(result.status_code, 302)
        self.assertTrue(result["Location"].endswith("/accounts/password/reset/done/"))
        result = self.client_get(result["Location"])

        self.assert_in_response("Check your email in a few minutes to finish the process.", result)

        # Check that the password reset email is from a noreply address.
        from django.core.mail import outbox

        self.assert_length(outbox, 0)

    @ratelimit_rule(10, 2, domain="password_reset_form_by_email")
    def test_rate_limiting(self) -> None:
        user_profile = self.example_user("hamlet")
        email = user_profile.delivery_email
        from django.core.mail import outbox

        start_time = time.time()
        with patch("time.time", return_value=start_time):
            self.client_post("/accounts/password/reset/", {"email": email})
            self.client_post("/accounts/password/reset/", {"email": email})
            self.assert_length(outbox, 2)

            # Too many password reset emails sent to the address, we won't send more.
            with self.assertLogs(level="INFO") as info_logs:
                self.client_post("/accounts/password/reset/", {"email": email})
            self.assertEqual(
                info_logs.output,
                [
                    "INFO:root:Too many password reset attempts for email hamlet@zulip.com from 127.0.0.1"
                ],
            )
            self.assert_length(outbox, 2)

            # Resetting for a different address works though.
            self.client_post("/accounts/password/reset/", {"email": self.example_email("othello")})
            self.assert_length(outbox, 3)
            self.client_post("/accounts/password/reset/", {"email": self.example_email("othello")})
            self.assert_length(outbox, 4)

        # After time, password reset emails can be sent again.
        with patch("time.time", return_value=start_time + 11):
            self.client_post("/accounts/password/reset/", {"email": email})
            self.client_post("/accounts/password/reset/", {"email": email})
            self.assert_length(outbox, 6)

    def test_wrong_subdomain(self) -> None:
        email = self.example_email("hamlet")

        # start the password reset process by supplying an email address
        result = self.client_post("/accounts/password/reset/", {"email": email}, subdomain="zephyr")

        # check the redirect link telling you to check mail for password reset link
        self.assertEqual(result.status_code, 302)
