import json
import os
import shutil
import uuid
from collections import defaultdict
from collections.abc import Callable, Iterable
from datetime import datetime, timedelta, timezone
from typing import Any
from unittest.mock import patch

import orjson
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.management.base import CommandError
from django.db.models import Q, QuerySet
from django.forms.models import model_to_dict
from django.test import override_settings
from django.utils.timezone import now as timezone_now
from typing_extensions import override

from analytics.models import UserCount
from version import ZULIP_VERSION
from zerver.actions.alert_words import do_add_alert_words
from zerver.actions.create_user import do_create_user
from zerver.actions.custom_profile_fields import (
    do_update_user_custom_profile_data_if_changed,
    try_add_realm_custom_profile_field,
)
from zerver.actions.muted_users import do_mute_user
from zerver.actions.navigation_views import do_add_navigation_view
from zerver.actions.presence import do_update_user_presence
from zerver.actions.reactions import check_add_reaction
from zerver.actions.realm_emoji import check_add_realm_emoji
from zerver.actions.realm_icon import do_change_icon_source
from zerver.actions.realm_logo import do_change_logo_source
from zerver.actions.realm_settings import (
    do_change_realm_plan_type,
    do_set_realm_authentication_methods,
)
from zerver.actions.saved_snippets import do_create_saved_snippet
from zerver.actions.scheduled_messages import check_schedule_message
from zerver.actions.streams import do_change_stream_description
from zerver.actions.user_activity import do_update_user_activity_interval
from zerver.actions.user_settings import do_change_user_delivery_email, do_change_user_setting
from zerver.actions.user_status import do_update_user_status
from zerver.actions.user_topics import do_set_user_topic_visibility_policy
from zerver.actions.users import do_deactivate_user
from zerver.lib import upload
from zerver.lib.avatar_hash import user_avatar_path
from zerver.lib.bot_config import set_bot_config
from zerver.lib.bot_lib import StateHandler
from zerver.lib.emoji import get_emoji_file_name, get_emoji_url
from zerver.lib.export import (
    PRESERVED_AUDIT_LOG_EVENT_TYPES,
    Record,
    do_export_realm,
    do_export_user,
    get_consented_user_ids,
)
from zerver.lib.import_realm import do_import_realm, get_incoming_message_ids
from zerver.lib.migration_status import STALE_MIGRATIONS, AppMigrations, MigrationStatusJson
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,
    create_s3_buckets,
    get_test_image_file,
    most_recent_message,
    most_recent_usermessage,
    read_test_image_file,
    use_s3_backend,
)
from zerver.lib.thumbnail import BadImageError
from zerver.lib.topic import DB_TOPIC_NAME
from zerver.lib.upload import claim_attachment, upload_avatar_image, upload_message_attachment
from zerver.lib.utils import assert_is_not_none, get_fk_field_name
from zerver.models import (
    AlertWord,
    Attachment,
    BotConfigData,
    BotStorageData,
    ChannelFolder,
    CustomProfileField,
    CustomProfileFieldValue,
    DirectMessageGroup,
    GroupGroupMembership,
    Message,
    MutedUser,
    NamedUserGroup,
    NavigationView,
    OnboardingStep,
    OnboardingUserMessage,
    Reaction,
    Realm,
    RealmAuditLog,
    RealmEmoji,
    RealmExport,
    RealmUserDefault,
    Recipient,
    ScheduledMessage,
    Stream,
    Subscription,
    UserActivity,
    UserGroup,
    UserGroupMembership,
    UserMessage,
    UserPresence,
    UserProfile,
    UserStatus,
    UserTopic,
)
from zerver.models.clients import Client, get_client
from zerver.models.groups import SystemGroups
from zerver.models.messages import ImageAttachment, SubMessage
from zerver.models.presence import PresenceSequence
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realm_emoji import get_all_custom_emoji_for_realm
from zerver.models.realms import get_realm
from zerver.models.recipients import (
    get_direct_message_group_hash,
    get_or_create_direct_message_group,
)
from zerver.models.streams import get_active_streams, get_stream
from zerver.models.users import ExternalAuthID, get_system_bot, get_user_by_delivery_email


def make_datetime(val: float) -> datetime:
    return datetime.fromtimestamp(val, tz=timezone.utc)


def get_output_dir() -> str:
    return os.path.join(settings.TEST_WORKER_DIR, "test-export")


def make_export_output_dir() -> str:
    output_dir = get_output_dir()
    if os.path.exists(output_dir):
        shutil.rmtree(output_dir)
    os.makedirs(output_dir)
    return output_dir


def read_json(fn: str) -> Any:
    output_dir = get_output_dir()
    full_fn = os.path.join(output_dir, fn)
    with open(full_fn, "rb") as f:
        return orjson.loads(f.read())


def export_fn(fn: str) -> str:
    output_dir = get_output_dir()
    return os.path.join(output_dir, fn)


def get_user_id(r: Realm, full_name: str) -> int:
    return UserProfile.objects.get(realm=r, full_name=full_name).id


def get_direct_message_group_hashes(r: Realm) -> str:
    cordelia_full_name = "Cordelia, Lear's daughter"
    hamlet_full_name = "King Hamlet"
    othello_full_name = "Othello, the Moor of Venice"

    user_id_list = [
        get_user_id(r, cordelia_full_name),
        get_user_id(r, hamlet_full_name),
        get_user_id(r, othello_full_name),
    ]

    direct_message_group_hash = get_direct_message_group_hash(user_id_list)
    return direct_message_group_hash


class ExportFile(ZulipTestCase):
    """This class is a container for shared helper functions
    used for both the realm-level and user-level export tests."""

    @override
    def setUp(self) -> None:
        super().setUp()
        assert settings.LOCAL_UPLOADS_DIR is not None
        self.rm_tree(settings.LOCAL_UPLOADS_DIR)

        # Deleting LOCAL_UPLOADS_DIR results in the test database
        # having RealmEmoji records without associated files.
        #
        # Even if we didn't delete them, the way that the test runner
        # varies settings.LOCAL_UPLOADS_DIR for each test worker
        # process would likely result in this being necessary anyway.
        RealmEmoji.objects.all().delete()

    def upload_files_for_user(
        self, user_profile: UserProfile, *, emoji_name: str = "whatever"
    ) -> None:
        message = most_recent_message(user_profile)
        url = upload_message_attachment("dummy.txt", "text/plain", b"zulip!", user_profile)[0]
        attachment_path_id = url.replace("/user_uploads/", "")
        claim_attachment(
            path_id=attachment_path_id,
            message=message,
            is_message_realm_public=True,
        )

        with get_test_image_file("img.png") as img_file:
            upload_avatar_image(img_file, user_profile, future=False)

        user_profile.avatar_source = "U"
        user_profile.save()

        realm = user_profile.realm

        with get_test_image_file("img.png") as img_file:
            check_add_realm_emoji(realm, emoji_name, user_profile, img_file, "image/png")

    def upload_files_for_realm(self, user_profile: UserProfile) -> None:
        realm = user_profile.realm

        with get_test_image_file("img.png") as img_file:
            upload.upload_backend.upload_realm_icon_image(img_file, user_profile, "image/png")
            do_change_icon_source(realm, Realm.ICON_UPLOADED, acting_user=None)

        with get_test_image_file("img.png") as img_file:
            upload.upload_backend.upload_realm_logo_image(
                img_file, user_profile, night=False, content_type="image/png"
            )
            do_change_logo_source(realm, Realm.LOGO_UPLOADED, False, acting_user=user_profile)
        with get_test_image_file("img.png") as img_file:
            upload.upload_backend.upload_realm_logo_image(
                img_file, user_profile, night=True, content_type="image/png"
            )
            do_change_logo_source(realm, Realm.LOGO_UPLOADED, True, acting_user=user_profile)

    def verify_attachment_json(self, user: UserProfile) -> None:
        attachment = Attachment.objects.get(owner=user)
        (record,) = read_json("attachment.json")["zerver_attachment"]
        self.assertEqual(record["path_id"], attachment.path_id)
        self.assertEqual(record["owner"], attachment.owner_id)
        self.assertEqual(record["realm"], attachment.realm_id)

    def verify_uploads(self, user: UserProfile, is_s3: bool) -> None:
        realm = user.realm

        attachment = Attachment.objects.get(owner=user)
        path_id = attachment.path_id

        # Test uploads
        fn = export_fn(f"uploads/{path_id}")
        with open(fn) as f:
            self.assertEqual(f.read(), "zulip!")
        (record,) = read_json("uploads/records.json")
        self.assertEqual(record["path"], path_id)
        self.assertEqual(record["s3_path"], path_id)

        if is_s3:
            realm_str, random_hash, file_name = path_id.split("/")
            self.assertEqual(realm_str, str(realm.id))
            self.assert_length(random_hash, 24)
            self.assertEqual(file_name, "dummy.txt")

            self.assertEqual(record["realm_id"], realm.id)
            self.assertEqual(record["user_profile_id"], user.id)
        else:
            realm_str, slot, random_hash, file_name = path_id.split("/")
            self.assertEqual(realm_str, str(realm.id))
            # We randomly pick a number between 0 and 255 and turn it into
            # hex in order to avoid large directories.
            assert len(slot) <= 2
            self.assert_length(random_hash, 24)
            self.assertEqual(file_name, "dummy.txt")

    def verify_emojis(self, user: UserProfile, is_s3: bool) -> None:
        realm = user.realm

        realm_emoji = RealmEmoji.objects.get(author=user)
        file_name = realm_emoji.file_name
        assert file_name is not None
        assert file_name.endswith(".png")

        emoji_path = f"{realm.id}/emoji/images/{file_name}"
        emoji_dir = export_fn(f"emoji/{realm.id}/emoji/images")
        self.assertEqual(set(os.listdir(emoji_dir)), {file_name, file_name + ".original"})

        (record1, record2) = read_json("emoji/records.json")
        # The return order is not guaranteed, so sort it so that we can reliably
        # know which record is for the .original file and which for the actual emoji.
        record, record_original = sorted(
            (record1, record2), key=lambda r: r["path"].endswith(".original")
        )

        self.assertEqual(record["file_name"], file_name)
        self.assertEqual(record["path"], emoji_path)
        self.assertEqual(record["s3_path"], emoji_path)
        self.assertEqual(record_original["file_name"], file_name)
        self.assertEqual(record_original["path"], emoji_path + ".original")
        self.assertEqual(record_original["s3_path"], emoji_path + ".original")

        if is_s3:
            self.assertEqual(record["realm_id"], realm.id)
            self.assertEqual(record["user_profile_id"], user.id)
            self.assertEqual(record_original["realm_id"], realm.id)
            self.assertEqual(record_original["user_profile_id"], user.id)

    def verify_realm_logo_and_icon(self) -> None:
        records = read_json("realm_icons/records.json")
        image_files = set()

        for record in records:
            self.assertEqual(record["path"], record["s3_path"])
            image_path = export_fn(f"realm_icons/{record['path']}")
            if image_path.endswith(".original"):
                with open(image_path, "rb") as image_file:
                    image_data = image_file.read()
                self.assertEqual(image_data, read_test_image_file("img.png"))
            else:
                self.assertTrue(os.path.exists(image_path))

            image_files.add(os.path.basename(image_path))

        self.assertEqual(
            set(image_files),
            {
                "night_logo.png",
                "logo.original",
                "logo.png",
                "icon.png",
                "night_logo.original",
                "icon.original",
            },
        )

    def verify_avatars(self, user: UserProfile) -> None:
        records = read_json("avatars/records.json")
        exported_paths = set()

        # Make sure all files in records.json got written.
        for record in records:
            self.assertEqual(record["path"], record["s3_path"])
            path = record["path"]
            fn = export_fn(f"avatars/{path}")
            assert os.path.exists(fn)

            if path.endswith(".original"):
                exported_paths.add(path)

                # For now we know that all our tests use
                # emojis based on img.png.  This may change some
                # day.
                with open(fn, "rb") as fb:
                    fn_data = fb.read()

                self.assertEqual(fn_data, read_test_image_file("img.png"))

        assert exported_paths

        # Right now we expect only our user to have an uploaded avatar.
        db_paths = {user_avatar_path(user) + ".original"}
        self.assertEqual(exported_paths, db_paths)

    def get_applied_migrations_fixture(self, fixture_name: str) -> AppMigrations:
        fixture = orjson.loads(
            self.fixture_data(fixture_name, "import_fixtures/applied_migrations_fixtures")
        )
        return fixture

    def get_applied_migrations_error_message(self, fixture_name: str) -> str:
        fixture = self.fixture_data(fixture_name, "import_fixtures/check_migrations_errors")
        fixture = fixture.format(version_placeholder=ZULIP_VERSION)
        return fixture.strip()

    def verify_migration_status_json(self) -> None:
        # This function asserts that the generated migration_status.json
        # is structurally familiar for it to be used for assertion at
        # import_realm.py. Hence, it doesn't really matter if the individual
        # apps' migrations in migration_status.json fixture are outdated as
        # long as they have the same format.
        exported: MigrationStatusJson = read_json("migration_status.json")

        applied_migrations_fixtures = os.listdir(
            self.fixture_file_name("", "import_fixtures/applied_migrations_fixtures")
        )

        for fixture in applied_migrations_fixtures:
            migration_by_app: AppMigrations = self.get_applied_migrations_fixture(fixture)
            with self.subTest(migration_fixture=fixture):
                self.assertTrue(
                    set(migration_by_app).issubset(set(exported["migrations_by_app"])),
                    f"""
                    Please make sure the `{fixture}` fixture represents the actual
                    `migration_status.json` file. If the format for the same migration
                    status differs, this fixture is probably stale and needs
                    updating.

                    If this variation is needed for testing purposes feel free to
                    exempt the fixture from this test.""",
                )

        # Make sure export doesn't produce a migration_status.json with stale
        # migrations.
        stale_migrations = []
        for app, stale_migration in STALE_MIGRATIONS:
            installed_app = exported["migrations_by_app"].get(app)
            if installed_app:
                stale_migrations = [mig for mig in installed_app if mig.endswith(stale_migration)]
        self.assert_length(stale_migrations, 0)


class RealmImportExportTest(ExportFile):
    def create_user_and_login(self, email: str, realm: Realm) -> None:
        self.register(email, "test", subdomain=realm.subdomain)

    def export_realm(
        self,
        realm: Realm,
        export_type: int,
        exportable_user_ids: set[int] | None = None,
    ) -> None:
        output_dir = make_export_output_dir()
        if export_type == RealmExport.EXPORT_FULL_WITH_CONSENT:
            assert exportable_user_ids is not None

        with self.assertLogs(level="INFO"):
            do_export_realm(
                realm=realm,
                output_dir=output_dir,
                processes=1,
                export_type=export_type,
                exportable_user_ids=exportable_user_ids,
            )

            # This is a unique field and thus the cycle of export->import
            # within the same server (which is what happens in our tests)
            # will cause a conflict - so rotate it.
            realm.uuid = uuid.uuid4()
            realm.save()

    def export_realm_and_create_auditlog(
        self,
        original_realm: Realm,
        export_type: int = RealmExport.EXPORT_FULL_WITHOUT_CONSENT,
        exportable_user_ids: set[int] | None = None,
    ) -> None:
        RealmAuditLog.objects.create(
            realm=original_realm,
            event_type=AuditLogEventType.REALM_EXPORTED,
            event_time=timezone_now(),
        )
        self.export_realm(original_realm, export_type, exportable_user_ids)

    def test_export_files_from_local(self) -> None:
        user = self.example_user("hamlet")
        realm = user.realm
        self.upload_files_for_user(user)
        self.upload_files_for_realm(user)
        self.export_realm_and_create_auditlog(realm)

        self.verify_attachment_json(user)
        self.verify_uploads(user, is_s3=False)
        self.verify_avatars(user)
        self.verify_emojis(user, is_s3=False)
        self.verify_realm_logo_and_icon()
        self.verify_migration_status_json()

    def test_public_only_export_files_private_uploads_not_included(self) -> None:
        """
        This test verifies that when doing a public_only export, private uploads
        don't get included in the exported data.
        """

        user_profile = self.example_user("hamlet")
        realm = user_profile.realm

        # We create an attachment tied to a personal message. That means it shouldn't be
        # included in a public export, as it's private data.
        personal_message_id = self.send_personal_message(user_profile, self.example_user("othello"))
        url = upload_message_attachment("dummy.txt", "text/plain", b"zulip!", user_profile)[0]
        attachment_path_id = url.replace("/user_uploads/", "")
        attachment = claim_attachment(
            path_id=attachment_path_id,
            message=Message.objects.get(id=personal_message_id),
            is_message_realm_public=True,
        )

        self.export_realm_and_create_auditlog(realm, export_type=RealmExport.EXPORT_PUBLIC)

        # The attachment row shouldn't have been exported:
        self.assertEqual(read_json("attachment.json")["zerver_attachment"], [])

        # Aside of the attachment row, we also need to verify that the file itself
        # isn't included.
        fn = export_fn(f"uploads/{attachment.path_id}")
        self.assertFalse(os.path.exists(fn))

    @use_s3_backend
    def test_export_files_from_s3(self) -> None:
        create_s3_buckets(settings.S3_AUTH_UPLOADS_BUCKET, settings.S3_AVATAR_BUCKET)

        user = self.example_user("hamlet")
        realm = user.realm

        self.upload_files_for_user(user)
        self.upload_files_for_realm(user)
        self.export_realm_and_create_auditlog(realm)

        self.verify_attachment_json(user)
        self.verify_uploads(user, is_s3=True)
        self.verify_avatars(user)
        self.verify_emojis(user, is_s3=True)
        self.verify_realm_logo_and_icon()
        self.verify_migration_status_json()

    def test_zulip_realm(self) -> None:
        realm = Realm.objects.get(string_id="zulip")

        default_bot = self.example_user("default_bot")
        pm_a_msg_id = self.send_personal_message(self.example_user("AARON"), default_bot)
        pm_b_msg_id = self.send_personal_message(default_bot, self.example_user("iago"))
        pm_c_msg_id = self.send_personal_message(
            self.example_user("othello"), self.example_user("hamlet")
        )

        realm_user_default = RealmUserDefault.objects.get(realm=realm)
        realm_user_default.default_language = "de"
        realm_user_default.save()

        welcome_bot = get_system_bot(settings.WELCOME_BOT, realm.id)
        onboarding_message_id = self.send_stream_message(
            welcome_bot, str(Realm.ZULIP_SANDBOX_CHANNEL_NAME), recipient_realm=realm
        )
        OnboardingUserMessage.objects.create(
            realm=realm,
            message_id=onboarding_message_id,
            flags=OnboardingUserMessage.flags.starred,
        )

        self.export_realm_and_create_auditlog(realm)

        data = read_json("realm.json")
        self.assert_length(data["zerver_userprofile_crossrealm"], 3)
        self.assert_length(data["zerver_userprofile_mirrordummy"], 0)

        exported_user_emails = self.get_set(data["zerver_userprofile"], "delivery_email")
        self.assertIn(self.example_email("cordelia"), exported_user_emails)
        self.assertIn("default-bot@zulip.com", exported_user_emails)

        exported_streams = self.get_set(data["zerver_stream"], "name")
        self.assertEqual(
            exported_streams,
            {
                "Denmark",
                "Rome",
                "Scotland",
                "Venice",
                "Verona",
                "core team",
                "Zulip",
                "sandbox",
            },
        )

        exported_alert_words = data["zerver_alertword"]

        # We set up 4 alert words for Hamlet, Cordelia, etc.
        # when we populate the test database.
        num_zulip_users = 10
        self.assert_length(exported_alert_words, num_zulip_users * 4)

        self.assertIn("robotics", {r["word"] for r in exported_alert_words})

        exported_realm_user_default = data["zerver_realmuserdefault"]
        self.assert_length(exported_realm_user_default, 1)
        self.assertEqual(exported_realm_user_default[0]["default_language"], "de")

        exported_usergroups = data["zerver_usergroup"]
        self.assert_length(exported_usergroups, 14)
        self.assertFalse("direct_members" in exported_usergroups[2])
        self.assertFalse("direct_subgroups" in exported_usergroups[2])

        exported_namedusergroups = data["zerver_namedusergroup"]
        self.assert_length(exported_namedusergroups, 9)
        self.assertEqual(exported_namedusergroups[2]["name"], "role:administrators")
        self.assertTrue("usergroup_ptr" in exported_namedusergroups[2])
        self.assertTrue("realm_for_sharding" in exported_namedusergroups[2])
        self.assertFalse("realm" in exported_namedusergroups[2])
        self.assertFalse("direct_members" in exported_namedusergroups[2])
        self.assertFalse("direct_subgroups" in exported_namedusergroups[2])

        exported_onboarding_usermessages = data["zerver_onboardingusermessage"]
        self.assert_length(exported_onboarding_usermessages, 1)
        self.assertEqual(exported_onboarding_usermessages[0]["message"], onboarding_message_id)
        self.assertEqual(
            exported_onboarding_usermessages[0]["flags_mask"],
            OnboardingUserMessage.flags.starred.mask,
        )
        self.assertEqual(exported_onboarding_usermessages[0]["realm"], realm.id)

        data = read_json("messages-000001.json")
        um = UserMessage.objects.all()[0]
        exported_um = self.find_by_id(data["zerver_usermessage"], um.id)
        self.assertEqual(exported_um["message"], um.message_id)
        self.assertEqual(exported_um["user_profile"], um.user_profile_id)

        exported_message = self.find_by_id(data["zerver_message"], um.message_id)
        self.assertEqual(exported_message["content"], um.message.content)

        exported_message_ids = self.get_set(data["zerver_message"], "id")
        self.assertIn(pm_a_msg_id, exported_message_ids)
        self.assertIn(pm_b_msg_id, exported_message_ids)
        self.assertIn(pm_c_msg_id, exported_message_ids)

    def test_export_realm_with_exportable_user_ids(self) -> None:
        realm = Realm.objects.get(string_id="zulip")

        cordelia = self.example_user("cordelia")
        hamlet = self.example_user("hamlet")
        polonius = self.example_user("polonius")
        iago = self.example_user("iago")
        othello = self.example_user("othello")

        do_change_user_setting(cordelia, "allow_private_data_export", True, acting_user=cordelia)
        do_change_user_setting(hamlet, "allow_private_data_export", True, acting_user=hamlet)
        exportable_user_ids = {cordelia.id, hamlet.id}

        pm_a_msg_id = self.send_personal_message(polonius, othello)
        pm_b_msg_id = self.send_personal_message(cordelia, iago)
        pm_c_msg_id = self.send_personal_message(hamlet, othello)
        pm_d_msg_id = self.send_personal_message(iago, hamlet)

        self.export_realm_and_create_auditlog(
            realm,
            export_type=RealmExport.EXPORT_FULL_WITH_CONSENT,
            exportable_user_ids=exportable_user_ids,
        )

        realm_data = read_json("realm.json")

        exported_user_emails = self.get_set(realm_data["zerver_userprofile"], "delivery_email")
        self.assertIn(cordelia.delivery_email, exported_user_emails)
        self.assertIn(hamlet.delivery_email, exported_user_emails)
        self.assertNotIn("default-bot@zulip.com", exported_user_emails)
        self.assertNotIn(iago.delivery_email, exported_user_emails)
        self.assertNotIn(polonius.delivery_email, exported_user_emails)

        dummy_user_emails = self.get_set(
            realm_data["zerver_userprofile_mirrordummy"], "delivery_email"
        )
        self.assertIn(iago.delivery_email, dummy_user_emails)
        self.assertIn(othello.delivery_email, dummy_user_emails)
        self.assertIn(polonius.delivery_email, dummy_user_emails)
        self.assertIn("default-bot@zulip.com", dummy_user_emails)
        self.assertNotIn(cordelia.delivery_email, dummy_user_emails)
        self.assertNotIn(hamlet.delivery_email, dummy_user_emails)

        message_data = read_json("messages-000001.json")

        exported_message_ids = self.get_set(message_data["zerver_message"], "id")
        self.assertNotIn(pm_a_msg_id, exported_message_ids)
        self.assertIn(pm_b_msg_id, exported_message_ids)
        self.assertIn(pm_c_msg_id, exported_message_ids)
        self.assertIn(pm_d_msg_id, exported_message_ids)

        personal_recipient_type_ids = (
            r["type_id"] for r in realm_data["zerver_recipient"] if r["type"] == Recipient.PERSONAL
        )
        for user_profile_id in [cordelia.id, hamlet.id, iago.id, othello.id, polonius.id]:
            self.assertIn(user_profile_id, personal_recipient_type_ids)

    def test_get_consented_user_ids(self) -> None:
        realm = get_realm("zulip")
        consented_user = self.example_user("iago")
        do_change_user_setting(consented_user, "allow_private_data_export", True, acting_user=None)

        non_consented_user = self.example_user("hamlet")
        do_change_user_setting(
            non_consented_user, "allow_private_data_export", False, acting_user=None
        )

        bot_of_consented_user = self.create_test_bot(
            "bot-of-consented-user", consented_user, full_name="Bot of consented user"
        )
        # Bots don't really use allow_private_data_export setting (and it should be False by default)
        # but set explicitly just to be clear on the setup.
        # A bot of a consented user is considered consented no matter what.
        do_change_user_setting(
            bot_of_consented_user, "allow_private_data_export", False, acting_user=None
        )

        deactivated_bot_of_consented_user = self.create_test_bot(
            "deactivated-bot-of-consented-user",
            consented_user,
            full_name="Deactivated bot of consented user",
        )
        do_change_user_setting(
            deactivated_bot_of_consented_user, "allow_private_data_export", False, acting_user=None
        )
        do_deactivate_user(deactivated_bot_of_consented_user, acting_user=None)

        # A bot of a non-consented user is considered not consented.
        bot_of_non_consented_user = self.create_test_bot(
            "bot-of-non-consented-user", non_consented_user, full_name="Bot of non-consented user"
        )
        do_change_user_setting(
            bot_of_consented_user, "allow_private_data_export", False, acting_user=None
        )

        # Unless the bot has allow_private_data_export explicitly set to True. This is a rather
        # unlikely case to encounter, since bots don't have a UI for editing such settings; but it could
        # be flipped by a server admin.
        consented_bot_of_non_consented_user = self.create_test_bot(
            "consented-bot-of-non-consented-user",
            non_consented_user,
            full_name="Consented bot of non-consented user",
        )
        do_change_user_setting(
            consented_bot_of_non_consented_user, "allow_private_data_export", True, acting_user=None
        )

        consented_user_ids = get_consented_user_ids(realm)

        self.assertIn(consented_user.id, consented_user_ids)
        self.assertNotIn(non_consented_user.id, consented_user_ids)

        self.assertIn(bot_of_consented_user.id, consented_user_ids)
        self.assertIn(deactivated_bot_of_consented_user.id, consented_user_ids)

        self.assertNotIn(bot_of_non_consented_user.id, consented_user_ids)
        self.assertIn(consented_bot_of_non_consented_user.id, consented_user_ids)

        # A deactivated consented user is still considered consented.
        do_deactivate_user(consented_user, acting_user=None)
        self.assertIn(consented_user.id, get_consented_user_ids(realm))

        # A mirror dummy is always considered consented, no matter the setting.
        do_deactivate_user(non_consented_user, acting_user=None)
        non_consented_user.is_mirror_dummy = True
        non_consented_user.save(update_fields=["is_mirror_dummy"])
        self.assertIn(non_consented_user.id, get_consented_user_ids(realm))

    def test_client_objects_export(self) -> None:
        """
        Client objects require some special handling when exporting. They aren't
        scoped to a realm - e.g. a Client object "website" can be used by multiple
        realms. Any Message sent from the web app in any realm will have sending_client
        pointing to the "website" Client row.

        However, we cannot just export the whole server table when exporting a realm,
        as it'll leak information about Clients which might only be used by some other realm(s).
        Instead, the export system needs to carefully determine the set of Clients which are pointed
        to by other objects in the export and only export those.

        In an export with consent, this means only exporting Clients which were used by at least
        one consented user in the realm - omitting those that may have only been used by a non-consenting
        user, as those should be considered private.
        """

        realm = get_realm("zulip")
        lear_realm = get_realm("lear")

        # Set up a message in the lear realm for more ease of testing.
        cordelia_lear = self.lear_user("cordelia")
        king_lear = self.lear_user("king")
        self.send_personal_message(cordelia_lear, king_lear)

        # Consented users:
        hamlet = self.example_user("hamlet")
        othello = self.example_user("othello")
        # Iago will be non-consenting.
        iago = self.example_user("iago")
        cordelia = self.example_user("cordelia")
        do_change_user_setting(hamlet, "allow_private_data_export", True, acting_user=None)
        do_change_user_setting(othello, "allow_private_data_export", True, acting_user=None)
        do_change_user_setting(iago, "allow_private_data_export", False, acting_user=None)

        a_message_id = self.send_personal_message(hamlet, othello)

        # This Client won't be used by the realm.
        non_realm_client = Client.objects.create(name="Non-realm client")

        # This Client will be used by both of the realms.
        realm_shared_client = Client.objects.create(name="Client shared between realms")
        last_realm_message = Message.objects.get(id=a_message_id)
        last_second_realm_message = Message.objects.filter(realm=lear_realm).latest("id")
        last_realm_message.sending_client = realm_shared_client
