def test_multiple_upload(self):
     # type: () -> None
     email = self.example_email('iago')
     self.login(email)
     with get_test_image_file('img.png') as fp1, get_test_image_file('img.png') as fp2:
         result = self.client_post('/json/realm/emoji/my_emoji', {'f1': fp1, 'f2': fp2})
     self.assert_json_error(result, 'You must upload exactly one file.')

    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 _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 test_multiple_upload_failure(self) -> None:
     """
     Attempting to upload two files should fail.
     """
     self.login(self.example_email("hamlet"))
     with get_test_image_file('img.png') as fp1, \
             get_test_image_file('img.png') as fp2:
         result = self.client_post("/json/users/me/avatar", {'f1': fp1, 'f2': fp2})
     self.assert_json_error(result, "You must upload exactly one avatar.")

    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 test_add_bot_with_too_many_files(self) -> None:
     self.login(self.example_email('hamlet'))
     self.assert_num_bots_equal(0)
     with get_test_image_file('img.png') as fp1, \
             get_test_image_file('img.gif') as fp2:
         bot_info = dict(
             full_name='whatever',
             short_name='whatever',
             file1=fp1,
             file2=fp2,
         )
         result = self.client_post("/json/bots", bot_info)
     self.assert_json_error(result, 'You may only upload one file at a time')
     self.assert_num_bots_equal(0)

 def test_emoji_upload_file_size_error(self) -> None:
     email = self.example_email('iago')
     self.login(email)
     with get_test_image_file('img.png') as fp:
         with self.settings(MAX_EMOJI_FILE_SIZE=0):
             result = self.client_post('/json/realm/emoji/my_emoji', {'file': fp})
     self.assert_json_error(result, 'Uploaded file is larger than the allowed limit of 0 MB')

 def test_upload_already_existed_emoji(self) -> None:
     email = self.example_email('iago')
     self.login(email)
     with get_test_image_file('img.png') as fp1:
         emoji_data = {'f1': fp1}
         result = self.client_post('/json/realm/emoji/green_tick', info=emoji_data)
     self.assert_json_error(result, 'Realm emoji with this Realm and Name already exists.')

    def test_transfer_emoji_to_s3(self) -> None:
        bucket = create_s3_buckets(settings.S3_AVATAR_BUCKET)[0]
        othello = self.example_user('othello')
        RealmEmoji.objects.all().delete()

        emoji_name = "emoji.png"
        image_file = get_test_image_file("img.png")

        emoji = check_add_realm_emoji(othello.realm, emoji_name, othello, image_file)
        if not emoji:
            raise AssertionError("Unable to add emoji.")

        emoji_path = RealmEmoji.PATH_ID_TEMPLATE.format(
            realm_id=othello.realm_id,
            emoji_file_name=emoji.file_name,
        )

        transfer_emoji_to_s3(1)

        self.assertEqual(len(bucket.get_all_keys()), 2)
        original_key = bucket.get_key(emoji_path + ".original")
        resized_key = bucket.get_key(emoji_path)

        image_file.seek(0)
        image_data = image_file.read()
        resized_image_data = resize_emoji(image_data)

        self.assertEqual(image_data, original_key.get_contents_as_string())
        self.assertEqual(resized_image_data, resized_key.get_contents_as_string())

    def test_upload(self) -> None:
        email = self.example_email('iago')
        self.login(email)
        with get_test_image_file('img.png') as fp1:
            emoji_data = {'f1': fp1}
            result = self.client_post('/json/realm/emoji/my_emoji', info=emoji_data)
        self.assert_json_success(result)
        self.assertEqual(200, result.status_code)
        emoji = RealmEmoji.objects.get(name="my_emoji")
        self.assertEqual(emoji.author.email, email)

        result = self.client_get("/json/realm/emoji")
        content = result.json()
        self.assert_json_success(result)
        self.assertEqual(len(content["emoji"]), 2)
        self.assertIn('author', content["emoji"]['my_emoji'])
        self.assertEqual(
            content["emoji"]['my_emoji']['author']['email'], email)

        realm_emoji = RealmEmoji.objects.get(name='my_emoji')
        file_name = realm_emoji.name + '.png'
        self.assertEqual(
            str(realm_emoji),
            '<RealmEmoji(zulip): %s my_emoji False %s>' % (realm_emoji.id, file_name)
        )

 def test_realm_icon_upload_file_size_error(self):
     # type: () -> None
     self.login(self.example_email("iago"))
     with get_test_image_file(self.correct_files[0][0]) as fp:
         with self.settings(MAX_ICON_FILE_SIZE=0):
             result = self.client_post("/json/realm/icon", {'file': fp})
     self.assert_json_error(result, "Uploaded file is larger than the allowed limit of 0 MB")

 def test_avatar_upload_file_size_error(self):
     # type: () -> None
     self.login(self.example_email("hamlet"))
     with get_test_image_file(self.correct_files[0][0]) as fp:
         with self.settings(MAX_AVATAR_FILE_SIZE=0):
             result = self.client_post("/json/users/me/avatar", {'file': fp})
     self.assert_json_error(result, "Uploaded file is larger than the allowed limit of 0 MB")

 def test_upload_uppercase_exception(self) -> None:
     email = self.example_email('iago')
     self.login(email)
     with get_test_image_file('img.png') as fp1:
         emoji_data = {'f1': fp1}
         result = self.client_post('/json/realm/emoji/my_EMoji', info=emoji_data)
     self.assert_json_error(result, 'Invalid characters in emoji name')

 def test_emoji_upload_by_guest_user(self) -> None:
     email = self.example_email('polonius')
     self.login(email)
     with get_test_image_file('img.png') as fp1:
         emoji_data = {'f1': fp1}
         result = self.client_post('/json/realm/emoji/my_emoji', info=emoji_data)
     self.assert_json_error(result, 'Not allowed for guest users')

 def test_failed_file_upload(self) -> None:
     email = self.example_email('iago')
     self.login(email)
     with mock.patch('zerver.lib.upload.write_local_file', side_effect=Exception()):
         with get_test_image_file('img.png') as fp1:
             emoji_data = {'f1': fp1}
             result = self.client_post('/json/realm/emoji/my_emoji', info=emoji_data)
     self.assert_json_error(result, "Image file upload failed.")

    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 create_test_emoji(self, name: str, author: UserProfile) -> RealmEmoji:
     with get_test_image_file('img.png') as img_file:
         realm_emoji = check_add_realm_emoji(realm=author.realm,
                                             name=name,
                                             author=author,
                                             image_file=img_file)
         if realm_emoji is None:
             raise Exception("Error creating test emoji.")   # nocoverage
     return realm_emoji

 def test_realm_icon_version(self) -> None:
     self.login(self.example_email("iago"))
     realm = get_realm('zulip')
     icon_version = realm.icon_version
     self.assertEqual(icon_version, 1)
     with get_test_image_file(self.correct_files[0][0]) as fp:
         self.client_post("/json/realm/icon", {'file': fp})
     realm = get_realm('zulip')
     self.assertEqual(realm.icon_version, icon_version + 1)

 def test_upload_anyone(self) -> None:
     email = self.example_email('othello')
     self.login(email)
     realm = get_realm('zulip')
     realm.add_emoji_by_admins_only = False
     realm.save()
     with get_test_image_file('img.png') as fp1:
         emoji_data = {'f1': fp1}
         result = self.client_post('/json/realm/emoji/my_emoji', info=emoji_data)
     self.assert_json_success(result)

 def test_upload_admins_only(self) -> None:
     email = self.example_email('othello')
     self.login(email)
     realm = get_realm('zulip')
     realm.add_emoji_by_admins_only = True
     realm.save()
     with get_test_image_file('img.png') as fp1:
         emoji_data = {'f1': fp1}
         result = self.client_post('/json/realm/emoji/my_emoji', info=emoji_data)
     self.assert_json_error(result, 'Must be an organization administrator')