def copy_avatar(self, source_profile: UserProfile, target_profile: UserProfile) -> None:
        s3_source_file_name = user_avatar_path(source_profile)
        s3_target_file_name = user_avatar_path(target_profile)

        key = self.get_avatar_key(s3_source_file_name + ".original")
        image_data = key.get_contents_as_string()  # type: ignore # https://github.com/python/typeshed/issues/1552
        content_type = key.content_type

        self.write_avatar_images(s3_target_file_name, target_profile, image_data, content_type)  # type: ignore # image_data is `bytes`, boto subs are wrong

def move_avatars_to_be_uid_based(apps, schema_editor):
    # type: (StateApps, DatabaseSchemaEditor) -> None
    user_profile_model = apps.get_model('zerver', 'UserProfile')
    if settings.LOCAL_UPLOADS_DIR is not None:
        for user_profile in user_profile_model.objects.filter(avatar_source=u"U"):
            src_file_name = user_avatar_hash(user_profile.email)
            dst_file_name = user_avatar_path(user_profile)
            try:
                move_local_file('avatars', src_file_name + '.original', dst_file_name + '.original')
                move_local_file('avatars', src_file_name + '-medium.png', dst_file_name + '-medium.png')
                move_local_file('avatars', src_file_name + '.png', dst_file_name + '.png')
            except MissingAvatarException:
                # If the user's avatar is missing, it's probably
                # because they previously changed their email address.
                # So set them to have a gravatar instead.
                user_profile.avatar_source = u"G"
                user_profile.save(update_fields=["avatar_source"])
    else:
        conn = S3Connection(settings.S3_KEY, settings.S3_SECRET_KEY)
        bucket_name = settings.S3_AVATAR_BUCKET
        bucket = conn.get_bucket(bucket_name, validate=False)
        for user_profile in user_profile_model.objects.filter(avatar_source=u"U"):
            uid_hash_path = user_avatar_path(user_profile)
            email_hash_path = user_avatar_hash(user_profile.email)
            if bucket.get_key(uid_hash_path):
                continue
            if not bucket.get_key(email_hash_path):
                # This is likely caused by a user having previously changed their email
                # If the user's avatar is missing, it's probably
                # because they previously changed their email address.
                # So set them to have a gravatar instead.
                user_profile.avatar_source = u"G"
                user_profile.save(update_fields=["avatar_source"])
                continue

            bucket.copy_key(uid_hash_path + ".original",
                            bucket_name,
                            email_hash_path + ".original")
            bucket.copy_key(uid_hash_path + "-medium.png",
                            bucket_name,
                            email_hash_path + "-medium.png")
            bucket.copy_key(uid_hash_path,
                            bucket_name,
                            email_hash_path)

        # From an error handling sanity perspective, it's best to
        # start deleting after everything is copied, so that recovery
        # from failures is easy (just rerun one loop or the other).
        for user_profile in user_profile_model.objects.filter(avatar_source=u"U"):
            bucket.delete_key(user_avatar_hash(user_profile.email) + ".original")
            bucket.delete_key(user_avatar_hash(user_profile.email) + "-medium.png")
            bucket.delete_key(user_avatar_hash(user_profile.email))

    def delete_avatar_image(self, user: UserProfile) -> None:
        path_id = user_avatar_path(user)
        bucket_name = settings.S3_AVATAR_BUCKET

        self.delete_file_from_s3(path_id + ".original", bucket_name)
        self.delete_file_from_s3(path_id + "-medium.png", bucket_name)
        self.delete_file_from_s3(path_id, bucket_name)

    def upload_avatar_image(self, user_file, acting_user_profile, target_user_profile):
        # type: (File, UserProfile, UserProfile) -> None
        content_type = guess_type(user_file.name)[0]
        bucket_name = settings.S3_AVATAR_BUCKET
        s3_file_name = user_avatar_path(target_user_profile)

        image_data = user_file.read()
        upload_image_to_s3(
            bucket_name,
            s3_file_name + ".original",
            content_type,
            target_user_profile,
            image_data,
        )

        # custom 500px wide version
        resized_medium = resize_avatar(image_data, MEDIUM_AVATAR_SIZE)
        upload_image_to_s3(
            bucket_name,
            s3_file_name + "-medium.png",
            "image/png",
            target_user_profile,
            resized_medium
        )

        resized_data = resize_avatar(image_data)
        upload_image_to_s3(
            bucket_name,
            s3_file_name,
            'image/png',
            target_user_profile,
            resized_data,
        )

    def test_upload_avatar_image(self) -> None:
        conn = S3Connection(settings.S3_KEY, settings.S3_SECRET_KEY)
        bucket = conn.create_bucket(settings.S3_AVATAR_BUCKET)

        user_profile = self.example_user('hamlet')
        path_id = user_avatar_path(user_profile)
        original_image_path_id = path_id + ".original"
        medium_path_id = path_id + "-medium.png"

        with get_test_image_file('img.png') as image_file:
            zerver.lib.upload.upload_backend.upload_avatar_image(image_file, user_profile, user_profile)
        test_image_data = open(get_test_image_file('img.png').name, 'rb').read()
        test_medium_image_data = resize_avatar(test_image_data, MEDIUM_AVATAR_SIZE)

        original_image_key = bucket.get_key(original_image_path_id)
        self.assertEqual(original_image_key.key, original_image_path_id)
        image_data = original_image_key.get_contents_as_string()
        self.assertEqual(image_data, test_image_data)

        medium_image_key = bucket.get_key(medium_path_id)
        self.assertEqual(medium_image_key.key, medium_path_id)
        medium_image_data = medium_image_key.get_contents_as_string()
        self.assertEqual(medium_image_data, test_medium_image_data)
        bucket.delete_key(medium_image_key)

        zerver.lib.upload.upload_backend.ensure_medium_avatar_image(user_profile)
        medium_image_key = bucket.get_key(medium_path_id)
        self.assertEqual(medium_image_key.key, medium_path_id)

    def test_import_files_from_local(self) -> None:

        realm = Realm.objects.get(string_id='zulip')
        self._setup_export_files()
        self._export_realm(realm)

        with patch('logging.info'):
            do_import_realm('var/test-export', 'test-zulip')
        imported_realm = Realm.objects.get(string_id='test-zulip')

        # Test attachments
        uploaded_file = Attachment.objects.get(realm=imported_realm)
        self.assertEqual(len(b'zulip!'), uploaded_file.size)

        attachment_file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, 'files', uploaded_file.path_id)
        self.assertTrue(os.path.isfile(attachment_file_path))

        # Test emojis
        realm_emoji = RealmEmoji.objects.get(realm=imported_realm)
        emoji_path = RealmEmoji.PATH_ID_TEMPLATE.format(
            realm_id=imported_realm.id,
            emoji_file_name=realm_emoji.file_name,
        )
        emoji_file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", emoji_path)
        self.assertTrue(os.path.isfile(emoji_file_path))

        # Test avatars
        user_email = Message.objects.all()[0].sender.email
        user_profile = UserProfile.objects.get(email=user_email, realm=imported_realm)
        avatar_path_id = user_avatar_path(user_profile) + ".original"
        avatar_file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", avatar_path_id)
        self.assertTrue(os.path.isfile(avatar_file_path))

    def _setup_export_files(self) -> Tuple[str, str, str, bytes]:
        realm = Realm.objects.get(string_id='zulip')
        message = Message.objects.all()[0]
        user_profile = message.sender
        url = upload_message_file(u'dummy.txt', len(b'zulip!'), u'text/plain', b'zulip!', user_profile)
        attachment_path_id = url.replace('/user_uploads/', '')
        claim_attachment(
            user_profile=user_profile,
            path_id=attachment_path_id,
            message=message,
            is_message_realm_public=True
        )
        avatar_path_id = user_avatar_path(user_profile)
        original_avatar_path_id = avatar_path_id + ".original"

        emoji_path = RealmEmoji.PATH_ID_TEMPLATE.format(
            realm_id=realm.id,
            emoji_file_name='1.png',
        )

        with get_test_image_file('img.png') as img_file:
            upload_emoji_image(img_file, '1.png', user_profile)
        with get_test_image_file('img.png') as img_file:
            upload_avatar_image(img_file, user_profile, user_profile)
        test_image = open(get_test_image_file('img.png').name, 'rb').read()
        message.sender.avatar_source = 'U'
        message.sender.save()

        return attachment_path_id, emoji_path, original_avatar_path_id, test_image

    def upload_avatar_image(self, user_file: File,
                            acting_user_profile: UserProfile,
                            target_user_profile: UserProfile) -> None:
        file_path = user_avatar_path(target_user_profile)

        image_data = user_file.read()
        self.write_avatar_images(file_path, image_data)

 def _transfer_avatar_to_s3(user: UserProfile) -> int:
     avatar_path = user_avatar_path(user)
     file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", avatar_path) + ".original"
     try:
         with open(file_path, 'rb') as f:
             s3backend.upload_avatar_image(f, user, user)
             logging.info("Uploaded avatar for {} in realm {}".format(user.email, user.realm.name))
     except FileNotFoundError:
         pass
     return 0

    def ensure_medium_avatar_image(self, user_profile: UserProfile) -> None:
        file_path = user_avatar_path(user_profile)

        output_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", file_path + "-medium.png")
        if os.path.isfile(output_path):
            return

        image_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", file_path + ".original")
        image_data = open(image_path, "rb").read()
        resized_medium = resize_avatar(image_data, MEDIUM_AVATAR_SIZE)
        write_local_file('avatars', file_path + '-medium.png', resized_medium)

    def test_export_files_from_s3(self) -> None:
        conn = S3Connection(settings.S3_KEY, settings.S3_SECRET_KEY)
        conn.create_bucket(settings.S3_AUTH_UPLOADS_BUCKET)
        conn.create_bucket(settings.S3_AVATAR_BUCKET)

        realm = Realm.objects.get(string_id='zulip')
        message = Message.objects.all()[0]
        user_profile = message.sender

        url = upload_message_file(u'dummy.txt', len(b'zulip!'), u'text/plain', b'zulip!', user_profile)
        attachment_path_id = url.replace('/user_uploads/', '')
        claim_attachment(
            user_profile=user_profile,
            path_id=attachment_path_id,
            message=message,
            is_message_realm_public=True
        )

        avatar_path_id = user_avatar_path(user_profile)
        original_avatar_path_id = avatar_path_id + ".original"

        emoji_path = RealmEmoji.PATH_ID_TEMPLATE.format(
            realm_id=realm.id,
            emoji_file_name='1.png',
        )

        with get_test_image_file('img.png') as img_file:
            upload_emoji_image(img_file, '1.png', user_profile)
        with get_test_image_file('img.png') as img_file:
            upload_avatar_image(img_file, user_profile, user_profile)
        test_image = open(get_test_image_file('img.png').name, 'rb').read()

        full_data = self._export_realm(realm)

        data = full_data['attachment']
        self.assertEqual(len(data['zerver_attachment']), 1)
        record = data['zerver_attachment'][0]
        self.assertEqual(record['path_id'], attachment_path_id)

        # Test uploads
        fields = attachment_path_id.split('/')
        fn = os.path.join(full_data['uploads_dir'], os.path.join(fields[1], fields[2]))
        with open(fn) as f:
            self.assertEqual(f.read(), 'zulip!')

        # Test emojis
        fn = os.path.join(full_data['emoji_dir'], emoji_path)
        fn = fn.replace('1.png', '')
        self.assertIn('1.png', os.listdir(fn))

        # Test avatars
        fn = os.path.join(full_data['avatar_dir'], original_avatar_path_id)
        fn_data = open(fn, 'rb').read()
        self.assertEqual(fn_data, test_image)

    def ensure_basic_avatar_image(self, user_profile: UserProfile) -> None:  # nocoverage
        # TODO: Refactor this to share code with ensure_medium_avatar_image
        file_path = user_avatar_path(user_profile)

        output_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", file_path + ".png")
        if os.path.isfile(output_path):
            return

        image_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", file_path + ".original")
        image_data = open(image_path, "rb").read()
        resized_avatar = resize_avatar(image_data)
        write_local_file('avatars', file_path + '.png', resized_avatar)

def _get_unversioned_avatar_url(avatar_source, email=None, realm_id=None,
                                user_profile_id=None, medium=False):
    # type: (Text, Text, Optional[int], Optional[int], bool) -> Text
    if avatar_source == u'U':
        if user_profile_id is not None and realm_id is not None:
            # If we can, avoid doing a database query to fetch user_profile
            hash_key = user_avatar_path_from_ids(user_profile_id, realm_id)
        else:
            user_profile = get_user_profile_by_email(email)
            hash_key = user_avatar_path(user_profile)
        return upload_backend.get_avatar_url(hash_key, medium=medium)
    elif settings.ENABLE_GRAVATAR:
        gravitar_query_suffix = "&s=%s" % (MEDIUM_AVATAR_SIZE,) if medium else ""
        hash_key = gravatar_hash(email)
        return u"https://secure.gravatar.com/avatar/%s?d=identicon%s" % (hash_key, gravitar_query_suffix)
    else:
        return settings.DEFAULT_AVATAR_URI+'?x=x'

    def test_export_files_from_local(self) -> None:
        message = Message.objects.all()[0]
        user_profile = message.sender
        url = upload_message_file(u'dummy.txt', len(b'zulip!'), u'text/plain', b'zulip!', user_profile)
        path_id = url.replace('/user_uploads/', '')
        claim_attachment(
            user_profile=user_profile,
            path_id=path_id,
            message=message,
            is_message_realm_public=True
        )
        avatar_path_id = user_avatar_path(user_profile)
        original_avatar_path_id = avatar_path_id + ".original"

        realm = Realm.objects.get(string_id='zulip')
        with get_test_image_file('img.png') as img_file:
            upload_emoji_image(img_file, '1.png', user_profile)
        with get_test_image_file('img.png') as img_file:
            upload_avatar_image(img_file, user_profile, user_profile)
        test_image = open(get_test_image_file('img.png').name, 'rb').read()
        message.sender.avatar_source = 'U'
        message.sender.save()

        full_data = self._export_realm(realm)

        data = full_data['attachment']
        self.assertEqual(len(data['zerver_attachment']), 1)
        record = data['zerver_attachment'][0]
        self.assertEqual(record['path_id'], path_id)

        # Test uploads
        fn = os.path.join(full_data['uploads_dir'], path_id)
        with open(fn) as f:
            self.assertEqual(f.read(), 'zulip!')

        # Test emojis
        fn = os.path.join(full_data['emoji_dir'],
                          RealmEmoji.PATH_ID_TEMPLATE.format(realm_id=realm.id, emoji_file_name='1.png'))
        fn = fn.replace('1.png', '')
        self.assertEqual('1.png', os.listdir(fn)[0])

        # Test avatars
        fn = os.path.join(full_data['avatar_dir'], original_avatar_path_id)
        fn_data = open(fn, 'rb').read()
        self.assertEqual(fn_data, test_image)

    def ensure_medium_avatar_image(self, user_profile):
        # type: (UserProfile) -> None
        file_path = user_avatar_path(user_profile)
        s3_file_name = file_path

        bucket_name = settings.S3_AVATAR_BUCKET
        conn = S3Connection(settings.S3_KEY, settings.S3_SECRET_KEY)
        bucket = get_bucket(conn, bucket_name)
        key = bucket.get_key(file_path)
        image_data = force_bytes(key.get_contents_as_string())

        resized_medium = resize_avatar(image_data, MEDIUM_AVATAR_SIZE)
        upload_image_to_s3(
            bucket_name,
            s3_file_name + "-medium.png",
            "image/png",
            user_profile,
            resized_medium
        )

    def test_transfer_avatars_to_s3(self) -> None:
        bucket = create_s3_buckets(settings.S3_AVATAR_BUCKET)[0]

        self.login(self.example_email("hamlet"))
        with get_test_image_file('img.png') as image_file:
            self.client_post("/json/users/me/avatar", {'file': image_file})

        user = self.example_user("hamlet")

        transfer_avatars_to_s3(1)

        path_id = user_avatar_path(user)
        image_key = bucket.get_key(path_id)
        original_image_key = bucket.get_key(path_id + ".original")
        medium_image_key = bucket.get_key(path_id + "-medium.png")

        self.assertEqual(len(bucket.get_all_keys()), 3)
        self.assertEqual(image_key.get_contents_as_string(), open(avatar_disk_path(user), "rb").read())
        self.assertEqual(original_image_key.get_contents_as_string(), open(avatar_disk_path(user, original=True), "rb").read())
        self.assertEqual(medium_image_key.get_contents_as_string(), open(avatar_disk_path(user, medium=True), "rb").read())

    def test_import_files_from_s3(self) -> None:
        conn = S3Connection(settings.S3_KEY, settings.S3_SECRET_KEY)
        uploads_bucket = conn.create_bucket(settings.S3_AUTH_UPLOADS_BUCKET)
        avatar_bucket = conn.create_bucket(settings.S3_AVATAR_BUCKET)

        realm = Realm.objects.get(string_id='zulip')
        self._setup_export_files()
        self._export_realm(realm)
        with patch('logging.info'):
            do_import_realm('var/test-export', 'test-zulip')
        imported_realm = Realm.objects.get(string_id='test-zulip')
        test_image_data = open(get_test_image_file('img.png').name, 'rb').read()

        # Test attachments
        uploaded_file = Attachment.objects.get(realm=imported_realm)
        self.assertEqual(len(b'zulip!'), uploaded_file.size)

        attachment_content = uploads_bucket.get_key(uploaded_file.path_id).get_contents_as_string()
        self.assertEqual(b"zulip!", attachment_content)

        # Test emojis
        realm_emoji = RealmEmoji.objects.get(realm=imported_realm)
        emoji_path = RealmEmoji.PATH_ID_TEMPLATE.format(
            realm_id=imported_realm.id,
            emoji_file_name=realm_emoji.file_name,
        )
        emoji_key = avatar_bucket.get_key(emoji_path)
        self.assertIsNotNone(emoji_key)
        self.assertEqual(emoji_key.key, emoji_path)

        # Test avatars
        user_email = Message.objects.all()[0].sender.email
        user_profile = UserProfile.objects.get(email=user_email, realm=imported_realm)
        avatar_path_id = user_avatar_path(user_profile) + ".original"
        original_image_key = avatar_bucket.get_key(avatar_path_id)
        self.assertEqual(original_image_key.key, avatar_path_id)
        image_data = original_image_key.get_contents_as_string()
        self.assertEqual(image_data, test_image_data)