Загрузка тестового файла с помощью тестового клиента Flask

Я тестирую два представления, причем оба они используют объект FileStorage для сохранения загруженных файлов из запроса на публикацию.

Во время тестов я не хочу сохранять загруженное хранилище, поэтому я исправляю FileStorage.save метод с использованием unittest"s patch декоратор.

Тест № 1:

@patch.object(FileStorage, 'save')
def test_create_data_file(self, fs_save_mock):
    """Test creation/upload of a data data file."""
    file_storage = FileStorage(
        name='Test Storage', filename='teststorage.ext')
    data = {'file': file_storage}

    rv = self.client.post('/api/data/upload', data=data)
    json_data = rv.get_json()
    datafile = DataFile.query.get(json_data['id'])

    fs_save_mock.assert_called_once_with(
        generate_upload_filepath(file_storage.filename))

Тест № 2:

@patch.object(FileStorage, 'save')
def test_create_data_file(self, fs_save_mock):
    """Test creation/upload of a data data file."""
    file_storage = FileStorage(
        name='Test Storage', filename='teststorage.ext')
    data = {'file': file_storage}

    rv = self.client.post('/api/data/upload', data=data)
    json_data = rv.get_json()
    datafile = DataFile.query.get(json_data['id'])

    fs_save_mock.assert_called_once_with(
        generate_upload_filepath(file_storage.filename))

Одно из представлений, которые я тестирую, выглядит так:

@blueprint.route('/upload', methods=['POST'])
@use_kwargs({'file_storage': fields.Field(location='files', load_from='file')})
def upload_datafile(file_storage):
    """Upload a datafile."""
    if file_storage is missing:
        abort(400, 'No file found in the request.')

    filepath = generate_upload_filepath(file_storage.filename)
    file_storage.save(filepath)

    data_file = DataFile.create(filename=file_storage.filename, path=filepath)
    return jsonify(DataFileSchema().dump(data_file).data), 201

Он работает нормально, если я запускаю только один из тестов, но когда я добавляю второй тест, запуск завершается неудачно со следующей ошибкой:

___________________ TestDataFileList.test_create_data_file ____________________

self = <test.test_api.TestDataFileList testMethod=test_create_data_file>
fs_save_mock = <MagicMock name='save' id='268959928'>

    @patch.object(FileStorage, 'save')
    def test_create_data_file(self, fs_save_mock):
        """Test creation/upload of a data data file."""
        file_storage = FileStorage(
            name='Test Storage', filename='teststorage.ext')
        data = {'file': file_storage}

>       rv = self.client.post('/api/data/upload', data=data)

test\test_api.py:469:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\venv\lib\site-packages\werkzeug\test.py:840: in post
    return self.open(*args, **kw)
..\venv\lib\site-packages\flask\testing.py:192: in open
    environ = builder.get_environ()
..\venv\lib\site-packages\werkzeug\test.py:588: in get_environ
    stream_encode_multipart(values, charset=self.charset)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

values = [<FileStorage: 'teststorage.ext' (None)>], use_tempfile = True
threshold = 512000
boundary = '---------------WerkzeugFormPart_1534924385.40363360.05943907979559804'
charset = 'utf-8'

    def stream_encode_multipart(values, use_tempfile=True, threshold=1024 * 500,
                                boundary=None, charset='utf-8'):
        """Encode a dict of values (either strings or file descriptors or
        :class:`FileStorage` objects.) into a multipart encoded string stored
        in a file descriptor.
        """
        if boundary is None:
            boundary = '---------------WerkzeugFormPart_%s%s' % (time(), random())
        _closure = [BytesIO(), 0, False]

        if use_tempfile:
            def write_binary(string):
                stream, total_length, on_disk = _closure
                if on_disk:
                    stream.write(string)
                else:
                    length = len(string)
                    if length + _closure[1] <= threshold:
                        stream.write(string)
                    else:
                        new_stream = TemporaryFile('wb+')
                        new_stream.write(stream.getvalue())
                        new_stream.write(string)
                        _closure[0] = new_stream
                        _closure[2] = True
                    _closure[1] = total_length + length
        else:
            write_binary = _closure[0].write

        def write(string):
            write_binary(string.encode(charset))

        if not isinstance(values, MultiDict):
            values = MultiDict(values)

        for key, values in iterlists(values):
            for value in values:
                write('--%s\r\nContent-Disposition: form-data; name="%s"' %
                      (boundary, key))
                reader = getattr(value, 'read', None)
                if reader is not None:
                    filename = getattr(value, 'filename',
                                       getattr(value, 'name', None))
                    content_type = getattr(value, 'content_type', None)
                    if content_type is None:
                        content_type = filename and \
                            mimetypes.guess_type(filename)[0] or \
                            'application/octet-stream'
                    if filename is not None:
                        write('; filename="%s"\r\n' % filename)
                    else:
                        write('\r\n')
                    write('Content-Type: %s\r\n\r\n' % content_type)
                    while 1:
>                       chunk = reader(16384)
E                       ValueError: I/O operation on closed file.

..\venv\lib\site-packages\werkzeug\test.py:96: ValueError
===================== 1 failed, 31 passed in 3.89 seconds =====================

Не знаете, как решить эту проблему, кто-нибудь из вас может мне помочь с этим?

0 ответов

Другие вопросы по тегам