import json
import os
import random
import re
import string
from datetime import datetime, timedelta
from enum import Enum
from typing import Any
from unittest import mock, skipUnless

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

from confirmation.models import Confirmation, create_confirmation_link
from zerver.actions.create_realm import do_change_realm_subdomain, do_create_realm
from zerver.actions.create_user import do_create_user
from zerver.actions.message_send import (
    internal_send_group_direct_message,
    internal_send_private_message,
    internal_send_stream_message,
)
from zerver.actions.realm_settings import (
    clean_deactivated_realm_data,
    delete_expired_demo_organizations,
    do_add_deactivated_redirect,
    do_change_realm_max_invites,
    do_change_realm_org_type,
    do_change_realm_permission_group_setting,
    do_change_realm_plan_type,
    do_deactivate_realm,
    do_delete_all_realm_attachments,
    do_reactivate_realm,
    do_scrub_realm,
    do_send_realm_reactivation_email,
    do_set_realm_authentication_methods,
    do_set_realm_property,
    do_set_realm_user_default_setting,
)
from zerver.actions.streams import do_deactivate_stream, merge_streams
from zerver.actions.user_groups import check_add_user_group
from zerver.lib.realm_description import get_realm_rendered_description, get_realm_text_description
from zerver.lib.send_email import send_future_email
from zerver.lib.streams import create_stream_if_needed
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import activate_push_notification_service
from zerver.lib.upload import delete_message_attachments, upload_message_attachment
from zerver.models import (
    Attachment,
    CustomProfileField,
    Message,
    NamedUserGroup,
    Realm,
    RealmAuditLog,
    RealmReactivationStatus,
    RealmUserDefault,
    ScheduledEmail,
    Stream,
    UserGroupMembership,
    UserMessage,
    UserProfile,
)
from zerver.models.groups import SystemGroups
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import get_realm
from zerver.models.streams import get_stream
from zerver.models.users import get_system_bot, get_user_profile_by_id

if settings.ZILENCER_ENABLED:
    from corporate.lib.stripe import get_seat_count


class RealmTest(ZulipTestCase):
    def assert_user_profile_cache_gets_new_name(
        self, user_profile: UserProfile, new_realm_name: str
    ) -> None:
        self.assertEqual(user_profile.realm.name, new_realm_name)

    def test_realm_creation_ensures_internal_realms(self) -> None:
        with mock.patch("zerver.actions.create_realm.server_initialized", return_value=False):
            with (
                mock.patch(
                    "zerver.actions.create_realm.create_internal_realm"
                ) as mock_create_internal,
                self.assertLogs(level="INFO") as info_logs,
            ):
                do_create_realm("testrealm", "Test Realm")
                mock_create_internal.assert_called_once()
            self.assertEqual(
                info_logs.output,
                ["INFO:root:Server not yet initialized. Creating the internal realm first."],
            )

    def test_realm_creation_on_special_subdomains_disallowed(self) -> None:
        with self.settings(SOCIAL_AUTH_SUBDOMAIN="zulipauth"), self.assertRaises(AssertionError):
            do_create_realm("zulipauth", "Test Realm")

        with (
            self.settings(SELF_HOSTING_MANAGEMENT_SUBDOMAIN="zulipselfhosting"),
            self.assertRaises(AssertionError),
        ):
            do_create_realm("zulipselfhosting", "Test Realm")

    def test_permission_for_education_non_profit_organization(self) -> None:
        realm = do_create_realm(
            "test_education_non_profit",
            "education_org_name",
            org_type=Realm.ORG_TYPES["education_nonprofit"]["id"],
        )

        admins_group = NamedUserGroup.objects.get(
            name=SystemGroups.ADMINISTRATORS, realm_for_sharding=realm, is_system_group=True
        )
        self.assertEqual(realm.can_create_public_channel_group_id, admins_group.id)

        realm = get_realm("test_education_non_profit")
        moderators_group = NamedUserGroup.objects.get(
            name=SystemGroups.MODERATORS, realm_for_sharding=realm, is_system_group=True
        )
        self.assertEqual(realm.can_add_subscribers_group.id, moderators_group.id)
        self.assertEqual(realm.can_create_groups.id, moderators_group.id)
        self.assertEqual(realm.can_invite_users_group.id, admins_group.id)
        self.assertEqual(realm.can_move_messages_between_channels_group.id, moderators_group.id)

    def test_permission_for_education_for_profit_organization(self) -> None:
        realm = do_create_realm(
            "test_education_for_profit",
            "education_org_name",
            org_type=Realm.ORG_TYPES["education"]["id"],
        )

        admins_group = NamedUserGroup.objects.get(
            name=SystemGroups.ADMINISTRATORS, realm_for_sharding=realm, is_system_group=True
        )
        self.assertEqual(realm.can_create_public_channel_group_id, admins_group.id)

        realm = get_realm("test_education_for_profit")
        moderators_group = NamedUserGroup.objects.get(
            name=SystemGroups.MODERATORS, realm_for_sharding=realm, is_system_group=True
        )
        self.assertEqual(realm.can_add_subscribers_group.id, moderators_group.id)
        self.assertEqual(realm.can_create_groups.id, moderators_group.id)
        self.assertEqual(realm.can_invite_users_group.id, admins_group.id)
        self.assertEqual(realm.can_move_messages_between_channels_group.id, moderators_group.id)

    def test_realm_enable_spectator_access(self) -> None:
        realm = do_create_realm(
            "test_web_public_true",
            "Foo",
            plan_type=Realm.PLAN_TYPE_STANDARD,
            enable_spectator_access=True,
        )
        self.assertEqual(realm.enable_spectator_access, True)

        realm = do_create_realm("test_web_public_false", "Boo", enable_spectator_access=False)
        self.assertEqual(realm.enable_spectator_access, False)

        with self.assertRaises(AssertionError):
            realm = do_create_realm("test_web_public_false_1", "Foo", enable_spectator_access=True)

        with self.assertRaises(AssertionError):
            realm = do_create_realm(
                "test_web_public_false_2",
                "Foo",
                plan_type=Realm.PLAN_TYPE_LIMITED,
                enable_spectator_access=True,
            )

    def test_do_set_realm_name_caching(self) -> None:
        """The main complicated thing about setting realm names is fighting the
        cache, and we start by populating the cache for Hamlet, and we end
        by checking the cache to ensure that the new value is there."""
        realm = get_realm("zulip")
        new_name = "Zed You Elle Eye Pea"
        do_set_realm_property(realm, "name", new_name, acting_user=None)
        self.assertEqual(get_realm(realm.string_id).name, new_name)
        self.assert_user_profile_cache_gets_new_name(self.example_user("hamlet"), new_name)

    def test_update_realm_name_events(self) -> None:
        realm = get_realm("zulip")
        new_name = "Puliz"
        with self.capture_send_event_calls(expected_num_events=1) as events:
            do_set_realm_property(realm, "name", new_name, acting_user=None)
        event = events[0]["event"]
        self.assertEqual(
            event,
            dict(
                type="realm",
                op="update",
                property="name",
                value=new_name,
            ),
        )

    def test_update_realm_description_events(self) -> None:
        realm = get_realm("zulip")
        new_description = "zulip dev group"
        with self.capture_send_event_calls(expected_num_events=1) as events:
            do_set_realm_property(realm, "description", new_description, acting_user=None)
        event = events[0]["event"]
        self.assertEqual(
            event,
            dict(
                type="realm",
                op="update",
                property="description",
                value=new_description,
            ),
        )

    def test_update_realm_description(self) -> None:
        self.login("iago")
        new_description = "zulip dev group"
        data = dict(description=new_description)
        with self.capture_send_event_calls(expected_num_events=1) as events:
            result = self.client_patch("/json/realm", data)
            self.assert_json_success(result)
            realm = get_realm("zulip")
            self.assertEqual(realm.description, new_description)

        event = events[0]["event"]
        self.assertEqual(
            event,
            dict(
                type="realm",
                op="update",
                property="description",
                value=new_description,
            ),
        )

    def test_realm_description_length(self) -> None:
        new_description = "A" * 1001
        data = dict(description=new_description)

        # create an admin user
        self.login("iago")

        result = self.client_patch("/json/realm", data)
        self.assert_json_error(result, "description is too long (limit: 1000 characters)")
        realm = get_realm("zulip")
        self.assertNotEqual(realm.description, new_description)

    def test_demo_organization_invite_required(self) -> None:
        realm = get_realm("zulip")
        self.assertFalse(realm.invite_required)

        self.login("desdemona")
        data = dict(invite_required="true")
        result = self.client_patch("/json/realm", data)
        self.assert_json_success(result)
        realm.refresh_from_db()
        self.assertTrue(realm.invite_required)

        # Update realm to be a demo organization
        realm.demo_organization_scheduled_deletion_date = timezone_now() + timedelta(days=30)
        realm.save()

        # Demo organization owner's don't have an email address set initially
        desdemona = self.example_user("desdemona")
        desdemona.delivery_email = ""
        desdemona.save()

        data = dict(invite_required="false")
        result = self.client_patch("/json/realm", data)
        self.assert_json_error(result, "Configure owner account email address.")

        desdemona.delivery_email = "desdemona@zulip.com"
        desdemona.save()
        result = self.client_patch("/json/realm", data)
        self.assert_json_success(result)
        realm.refresh_from_db()
        self.assertFalse(realm.invite_required)

    def test_realm_convert_demo_realm(self) -> None:
        data = dict(string_id="coolrealm")

        self.login("iago")
        result = self.client_patch("/json/realm", data)
        self.assert_json_error(result, "Must be an organization owner")

        self.login("desdemona")
        result = self.client_patch("/json/realm", data)
        self.assert_json_error(result, "Must be a demo organization.")

        realm = get_realm("zulip")
        realm.demo_organization_scheduled_deletion_date = timezone_now() + timedelta(days=30)
        realm.save()

        # Demo organization owner must have added an email before converting.
        desdemona = self.example_user("desdemona")
        desdemona.delivery_email = ""
        desdemona.save()
        result = self.client_patch("/json/realm", data)
        self.assert_json_error(result, "Configure owner account email address.")

        desdemona.delivery_email = "desdemona@zulip.com"
        desdemona.save()

        # Subdomain must be available to convert demo organization.
        data = dict(string_id="lear")
        result = self.client_patch("/json/realm", data)
        self.assert_json_error(
            result, "Subdomain is already in use. Please choose a different one."
        )

        # Now try to change the string_id to something available.
        data = dict(string_id="coolrealm")
        result = self.client_patch("/json/realm", data)
        self.assert_json_success(result)
        json = orjson.loads(result.content)
        self.assertEqual(json["realm_uri"], "http://coolrealm.testserver")
        self.assertEqual(json["realm_url"], "http://coolrealm.testserver")
        realm = get_realm("coolrealm")
        self.assertIsNone(realm.demo_organization_scheduled_deletion_date)
        self.assertEqual(realm.string_id, data["string_id"])

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.REALM_SUBDOMAIN_CHANGED, acting_user=desdemona
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: "zulip",
            RealmAuditLog.NEW_VALUE: "coolrealm",
            "was_demo_organization": True,
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)
        self.assertEqual(realm_audit_log.acting_user, desdemona)

    def test_realm_name_length(self) -> None:
        new_name = "A" * (Realm.MAX_REALM_NAME_LENGTH + 1)
        data = dict(name=new_name)

        # create an admin user
        self.login("iago")

        result = self.client_patch("/json/realm", data)
        self.assert_json_error(result, "name is too long (limit: 40 characters)")
        realm = get_realm("zulip")
        self.assertNotEqual(realm.name, new_name)

    def test_admin_restrictions_for_changing_realm_name(self) -> None:
        new_name = "Mice will play while the cat is away"

        self.login("othello")

        req = dict(name=new_name)
        result = self.client_patch("/json/realm", req)
        self.assert_json_error(result, "Must be an organization administrator")

    def test_unauthorized_name_change(self) -> None:
        data = {"full_name": "Sir Hamlet"}
        user_profile = self.example_user("hamlet")
        self.login_user(user_profile)
        do_set_realm_property(user_profile.realm, "name_changes_disabled", True, acting_user=None)
        url = "/json/settings"
        result = self.client_patch(url, data)
        self.assertEqual(result.status_code, 200)
        # Since the setting fails silently, no message is returned
        self.assert_in_response("", result)
        # Realm admins can change their name even setting is disabled.
        data = {"full_name": "New Iago"}
        self.login("iago")
        url = "/json/settings"
        result = self.client_patch(url, data)
        self.assert_json_success(result)

    def test_do_deactivate_realm_clears_user_realm_cache(self) -> None:
        """The main complicated thing about deactivating realm names is
        updating the cache, and we start by populating the cache for
        Hamlet, and we end by checking the cache to ensure that his
        realm appears to be deactivated.  You can make this test fail
        by disabling cache.flush_realm()."""
        hamlet_id = self.example_user("hamlet").id
        get_user_profile_by_id(hamlet_id)
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        user = get_user_profile_by_id(hamlet_id)
        self.assertTrue(user.realm.deactivated)

    def test_do_change_realm_delete_clears_user_realm_cache(self) -> None:
        hamlet_id = self.example_user("hamlet").id
        get_user_profile_by_id(hamlet_id)
        realm = get_realm("zulip")
        realm.delete()
        with self.assertRaises(UserProfile.DoesNotExist):
            get_user_profile_by_id(hamlet_id)

    def test_do_change_realm_subdomain_clears_user_realm_cache(self) -> None:
        """The main complicated thing about changing realm subdomains is
        updating the cache, and we start by populating the cache for
        Hamlet, and we end by checking the cache to ensure that his
        realm appears to be deactivated.  You can make this test fail
        by disabling cache.flush_realm()."""
        hamlet_id = self.example_user("hamlet").id
        user = get_user_profile_by_id(hamlet_id)
        realm = get_realm("zulip")
        iago = self.example_user("iago")
        do_change_realm_subdomain(realm, "newzulip", acting_user=iago)
        user = get_user_profile_by_id(hamlet_id)
        self.assertEqual(user.realm.string_id, "newzulip")

        placeholder_realm = get_realm("zulip")
        self.assertTrue(placeholder_realm.deactivated)
        self.assertEqual(placeholder_realm.deactivated_redirect, user.realm.url)

        realm_audit_log = RealmAuditLog.objects.filter(
            event_type=AuditLogEventType.REALM_SUBDOMAIN_CHANGED, acting_user=iago
        ).last()
        assert realm_audit_log is not None
        expected_extra_data = {
            RealmAuditLog.OLD_VALUE: "zulip",
            RealmAuditLog.NEW_VALUE: "newzulip",
            "was_demo_organization": False,
        }
        self.assertEqual(realm_audit_log.extra_data, expected_extra_data)
        self.assertEqual(realm_audit_log.acting_user, iago)

    def test_do_deactivate_realm_clears_scheduled_jobs(self) -> None:
        user = self.example_user("hamlet")
        send_future_email(
            "zerver/emails/onboarding_zulip_topics",
            user.realm,
            to_user_ids=[user.id],
            delay=timedelta(hours=1),
        )
        self.assertEqual(ScheduledEmail.objects.count(), 1)
        do_deactivate_realm(
            user.realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertEqual(ScheduledEmail.objects.count(), 0)

    def test_do_change_realm_description_clears_cached_descriptions(self) -> None:
        realm = get_realm("zulip")
        rendered_description = get_realm_rendered_description(realm)
        text_description = get_realm_text_description(realm)

        realm.description = "New description"
        realm.save(update_fields=["description"])

        new_rendered_description = get_realm_rendered_description(realm)
        self.assertNotEqual(rendered_description, new_rendered_description)
        self.assertIn(realm.description, new_rendered_description)

        new_text_description = get_realm_text_description(realm)
        self.assertNotEqual(text_description, new_text_description)
        self.assertEqual(realm.description, new_text_description)

    def test_do_deactivate_realm_on_deactivated_realm(self) -> None:
        """Ensure early exit is working in realm deactivation"""
        realm = get_realm("zulip")
        self.assertFalse(realm.deactivated)

        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertTrue(realm.deactivated)

        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertTrue(realm.deactivated)

    def test_do_set_deactivated_redirect_on_deactivated_realm(self) -> None:
        """Ensure that the redirect url is working when deactivating realm"""
        realm = get_realm("zulip")

        redirect_url = "new_server.zulip.com"
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertTrue(realm.deactivated)
        do_add_deactivated_redirect(realm, redirect_url)
        self.assertEqual(realm.deactivated_redirect, redirect_url)

        new_redirect_url = "test.zulip.com"
        do_add_deactivated_redirect(realm, new_redirect_url)
        self.assertEqual(realm.deactivated_redirect, new_redirect_url)
        self.assertNotEqual(realm.deactivated_redirect, redirect_url)

    def test_do_reactivate_realm(self) -> None:
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm,
            acting_user=None,
            deactivation_reason="owner_request",
            email_owners=False,
            deletion_delay_days=15,
        )
        self.assertTrue(realm.deactivated)

        do_reactivate_realm(realm)
        self.assertFalse(realm.deactivated)
        self.assertEqual(realm.scheduled_deletion_date, None)

        log_entry = RealmAuditLog.objects.last()
        assert log_entry is not None

        self.assertEqual(log_entry.realm, realm)
        self.assertEqual(log_entry.event_type, AuditLogEventType.REALM_REACTIVATED)
        log_entry_id = log_entry.id

        with self.assertLogs(level="WARNING") as m:
            # do_reactivate_realm on a realm that's not deactivated should be a noop.
            do_reactivate_realm(realm)

        self.assertEqual(
            m.output,
            [f"WARNING:root:Realm {realm.id} cannot be reactivated because it is already active."],
        )

        self.assertFalse(realm.deactivated)

        latest_log_entry = RealmAuditLog.objects.last()
        assert latest_log_entry is not None
        self.assertEqual(latest_log_entry.id, log_entry_id)

    def test_realm_reactivation_link(self) -> None:
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertTrue(realm.deactivated)

        obj = RealmReactivationStatus.objects.create(realm=realm)
        confirmation_url = create_confirmation_link(obj, Confirmation.REALM_REACTIVATION)
        key = confirmation_url.split("/")[-1]
        response = self.client_get(confirmation_url)
        self.assert_in_success_response(["redirect-to-post-form"], response)

        response = self.client_post("/reactivate/", {"key": key})
        self.assert_in_success_response(
            ["Your organization has been successfully reactivated"], response
        )
        realm = get_realm("zulip")
        self.assertFalse(realm.deactivated)

        # Make sure the link can't be reused.
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        response = self.client_get(confirmation_url)
        self.assertEqual(response.status_code, 404)
        response = self.client_post("/reactivate/", {"key": key})
        self.assertEqual(response.status_code, 404)

    def test_realm_reactivation_confirmation_object(self) -> None:
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertTrue(realm.deactivated)
        obj = RealmReactivationStatus.objects.create(realm=realm)
        create_confirmation_link(obj, Confirmation.REALM_REACTIVATION)
        confirmation = Confirmation.objects.last()
        assert confirmation is not None
        self.assertEqual(confirmation.content_object, obj)
        self.assertEqual(confirmation.realm, realm)

    def test_realm_deactivation_demo_organization_owner_email_not_configured(self) -> None:
        demo_name = "demo deactivate no email"
        result = self.submit_demo_creation_form(demo_name)
        realm = Realm.objects.filter(name=demo_name).latest("date_created")
        self.assertEqual(result.status_code, 302)
        self.assertTrue(
            result["Location"].startswith(
                f"http://{realm.string_id}.testserver/accounts/login/subdomain"
            )
        )
        self.assertIsNotNone(realm.demo_organization_scheduled_deletion_date)

        result = self.client_get(result["Location"], subdomain=realm.string_id)
        self.assertEqual(result.status_code, 302)
        self.assertEqual(result["Location"], f"http://{realm.string_id}.testserver")

        demo_owner_account = realm.get_first_human_user()
        assert demo_owner_account is not None
        self.assert_logged_in_user_id(demo_owner_account.id)
        self.assertEqual(demo_owner_account.delivery_email, "")

        # There must be a value set for deletion_delay_days.
        result = self.client_post("/json/realm/deactivate", subdomain=realm.subdomain)
        self.assert_json_error(result, "Invalid data deletion time for demo organization.")

        # If the owner has not configured an email, then deletion_delay_days
        # must be set to 0, i.e., immediate data deletion as there is no way
        # to recover the demo organization for the user.
        result = self.client_post(
            "/json/realm/deactivate", {"deletion_delay_days": 14}, subdomain=realm.subdomain
        )
        self.assert_json_error(result, "Invalid data deletion time for demo organization.")

        with self.assertLogs(level="INFO"):
            result = self.client_post(
                "/json/realm/deactivate", {"deletion_delay_days": 0}, subdomain=realm.subdomain
            )
        self.assert_json_success(result)
        self.assert_length(mail.outbox, 0)
        self.assert_logged_in_user_id(None)

    def test_realm_deactivation_demo_organization_owner_email_configured(self) -> None:
        demo_name = "demo deactivate with email"
        result = self.submit_demo_creation_form(demo_name)
        realm = Realm.objects.filter(name=demo_name).latest("date_created")
        self.assertEqual(result.status_code, 302)
        self.assertTrue(
            result["Location"].startswith(
                f"http://{realm.string_id}.testserver/accounts/login/subdomain"
            )
        )
        self.assertIsNotNone(realm.demo_organization_scheduled_deletion_date)

        result = self.client_get(result["Location"], subdomain=realm.string_id)
        self.assertEqual(result.status_code, 302)
        self.assertEqual(result["Location"], f"http://{realm.string_id}.testserver")

        demo_owner_account = realm.get_first_human_user()
        assert demo_owner_account is not None
        self.assert_logged_in_user_id(demo_owner_account.id)
        self.assertEqual(demo_owner_account.delivery_email, "")

        # Set an email for the demo organization owner's account.
        demo_owner_account.delivery_email = "demo-owner@example.com"
        demo_owner_account.save()

        # There must be a value set for deletion_delay_days.
        result = self.client_post("/json/realm/deactivate", subdomain=realm.subdomain)
        self.assert_json_error(result, "Invalid data deletion time for demo organization.")

        # If the owner has configured an email, then deletion_delay_days
        # must be less than the automatic scheduled deletion date.
        result = self.client_post(
            "/json/realm/deactivate", {"deletion_delay_days": 90}, subdomain=realm.subdomain
        )
        self.assert_json_error(result, "Invalid data deletion time for demo organization.")

        result = self.client_post(
            "/json/realm/deactivate", {"deletion_delay_days": 10}, subdomain=realm.subdomain
        )
        self.assert_json_success(result)
        self.assert_length(mail.outbox, 1)
        self.assert_length(mail.outbox, 1)
        self.assertIn(
            f"Your Zulip organization {demo_name} has been deactivated", mail.outbox[0].subject
        )
        self.assertIn(
            f"You have deactivated your Zulip demo organization, {demo_name},", mail.outbox[0].body
        )
        self.assert_logged_in_user_id(None)

    def test_do_send_realm_deactivation_email_no_acting_user(self) -> None:
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=True
        )
        self.assertEqual(realm.deactivated, True)
        self.assert_length(mail.outbox, 1)
        self.assertIn(
            "Your Zulip organization Zulip Dev has been deactivated", mail.outbox[0].subject
        )
        self.assertIn("Your Zulip organization, Zulip Dev, was deactivated on", mail.outbox[0].body)

    def test_do_send_realm_deactivation_email_by_support(self) -> None:
        realm = get_realm("lear")
        king = self.lear_user("king")
        king.role = UserProfile.ROLE_REALM_OWNER
        king.save()
        iago = self.example_user("iago")
        do_deactivate_realm(
            realm, acting_user=iago, deactivation_reason="owner_request", email_owners=True
        )
        self.assertEqual(realm.deactivated, True)
        self.assert_length(mail.outbox, 1)
        self.assertIn(
            "Your Zulip organization Lear & Co. has been deactivated", mail.outbox[0].subject
        )
        self.assertIn(
            "Your Zulip organization, Lear & Co., was deactivated on",
            mail.outbox[0].body,
        )

    def test_do_send_realm_deactivation_email_by_owner(self) -> None:
        realm = get_realm("zulip")
        iago = self.example_user("iago")
        iago.role = UserProfile.ROLE_REALM_OWNER
        iago.save(update_fields=["role"])
        do_deactivate_realm(
            realm,
            acting_user=iago,
            deactivation_reason="owner_request",
            deletion_delay_days=14,
            email_owners=True,
        )
        self.assertEqual(realm.deactivated, True)
        self.assert_length(mail.outbox, 2)
        for email in mail.outbox:
            if email.to[0] == "iago@zulip.com":
                self.assertIn(
                    "Your Zulip organization Zulip Dev has been deactivated", email.subject
                )
                self.assertIn(
                    "You have deactivated your Zulip organization, Zulip Dev, on", email.body
                )
            else:
                self.assertIn(
                    "Your Zulip organization Zulip Dev has been deactivated", email.subject
                )
                self.assertIn(
                    "Your Zulip organization, Zulip Dev, was deactivated by Iago on", email.body
                )
            self.assertIn(
                "All data associated with this organization will be permanently deleted on",
                email.body,
            )

    def test_do_send_realm_deactivation_email_with_immediate_data_deletion(self) -> None:
        realm = get_realm("zulip")
        desdemona = self.example_user("desdemona")
        with (
            mock.patch("zerver.actions.realm_settings.do_scrub_realm") as mock_scrub_realm,
            self.assertLogs(level="INFO"),
        ):
            do_deactivate_realm(
                realm,
                acting_user=desdemona,
                deactivation_reason="owner_request",
                deletion_delay_days=0,
                email_owners=True,
            )
            self.assertEqual(realm.deactivated, True)
            mock_scrub_realm.assert_called_once_with(realm, acting_user=None)
            self.assert_length(mail.outbox, 1)
            email = mail.outbox[0]
            self.assertIn("Your Zulip organization Zulip Dev has been deactivated", email.subject)
            self.assertIn("You have deactivated your Zulip organization, Zulip Dev, on", email.body)
            self.assertIn(
                "All data associated with this organization has been permanently deleted.",
                email.body,
            )

    def test_do_send_realm_reactivation_email(self) -> None:
        realm = get_realm("zulip")
        do_deactivate_realm(
            realm, acting_user=None, deactivation_reason="owner_request", email_owners=False
        )
        self.assertEqual(realm.deactivated, True)
        iago = self.example_user("iago")
        do_send_realm_reactivation_email(realm, acting_user=iago)

        self.assert_length(mail.outbox, 1)
        self.assertEqual(self.email_envelope_from(mail.outbox[0]), settings.NOREPLY_EMAIL_ADDRESS)
        self.assertRegex(
            self.email_display_from(mail.outbox[0]),
            rf"^testserver account security <{self.TOKENIZED_NOREPLY_REGEX}>\Z",
        )
        self.assertIn("Reactivate your Zulip organization", mail.outbox[0].subject)
        self.assertIn("Dear former administrators", mail.outbox[0].body)
        admins = realm.get_human_admin_users()
        confirmation_url = self.get_confirmation_url_from_outbox(admins[0].delivery_email)
        key = confirmation_url.split("/")[-1]

        response = self.client_get(confirmation_url)
        self.assert_in_success_response(["redirect-to-post-form"], response)

        response = self.client_post("/reactivate/", {"key": key})
        self.assert_in_success_response(
            ["Your organization has been successfully reactivated"], response
        )
        realm = get_realm("zulip")
