taipy/tests/gui/server/http/test_file_upload.py

# Copyright 2021-2025 Avaiga Private Limited
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.

import inspect
import io
import os
import pathlib
import tempfile

import pytest

from taipy.gui import Gui
from taipy.gui.data.data_scope import _DataScopes
from taipy.gui.utils import _get_non_existent_file_path


def test_file_upload_no_varname(gui: Gui, helpers):
    gui.run(run_server=False)
    flask_client = gui._server.test_client()
    # Get the jsx once so that the page will be evaluated -> variable will be registered
    sid = helpers.create_scope_and_get_sid(gui)
    with pytest.warns(UserWarning):
        ret = flask_client.post(f"/taipy-uploads?client_id={sid}")
        assert ret.status_code == 400


def test_file_upload_no_blob(gui: Gui, helpers):
    gui.run(run_server=False)
    flask_client = gui._server.test_client()
    # Get the jsx once so that the page will be evaluated -> variable will be registered
    sid = helpers.create_scope_and_get_sid(gui)
    with pytest.warns(UserWarning):
        ret = flask_client.post(f"/taipy-uploads?client_id={sid}", data={"var_name": "varname"})
        assert ret.status_code == 400


def test_file_upload_no_filename(gui: Gui, helpers):
    gui.run(run_server=False)
    flask_client = gui._server.test_client()
    file = (io.BytesIO(b"abcdef"), "")
    # Get the jsx once so that the page will be evaluated -> variable will be registered
    sid = helpers.create_scope_and_get_sid(gui)
    with pytest.warns(UserWarning):
        ret = flask_client.post(f"/taipy-uploads?client_id={sid}", data={"var_name": "varname", "blob": file})
        assert ret.status_code == 400


def test_file_upload_simple(gui: Gui, helpers):
    gui.run(run_server=False)
    flask_client = gui._server.test_client()
    # Get the jsx once so that the page will be evaluated -> variable will be registered
    sid = helpers.create_scope_and_get_sid(gui)
    file_name = "test.jpg"
    file = (io.BytesIO(b"abcdef"), file_name)
    upload_path = pathlib.Path(gui._get_config("upload_folder", tempfile.gettempdir()))
    file_name = _get_non_existent_file_path(upload_path, file_name).name
    ret = flask_client.post(
        f"/taipy-uploads?client_id={sid}",
        data={"var_name": "varname", "blob": file},
        content_type="multipart/form-data",
    )
    assert ret.status_code == 200
    created_file = upload_path / file_name
    assert created_file.exists()


def test_file_upload_multi_part(gui: Gui, helpers):
    gui.run(run_server=False)
    flask_client = gui._server.test_client()
    # Get the jsx once so that the page will be evaluated -> variable will be registered
    sid = helpers.create_scope_and_get_sid(gui)
    file_name = "test2.jpg"
    file0 = (io.BytesIO(b"abcdef"), file_name)
    file1 = (io.BytesIO(b"abcdef"), file_name)
    upload_path = pathlib.Path(gui._get_config("upload_folder", tempfile.gettempdir()))
    file_name = _get_non_existent_file_path(upload_path, file_name).name
    ret = flask_client.post(
        f"/taipy-uploads?client_id={sid}",
        data={"var_name": "varname", "blob": file0, "total": "2", "part": "0"},
        content_type="multipart/form-data",
    )
    assert ret.status_code == 200
    ret = flask_client.post(
        f"/taipy-uploads?client_id={sid}",
        data={"var_name": "varname", "blob": file1, "total": "2", "part": "1"},
        content_type="multipart/form-data",
    )
    assert ret.status_code == 200
    file_path = upload_path / file_name
    assert file_path.exists()


def test_file_upload_multiple(gui: Gui, helpers):
    var_name = "varname"
    gui._set_frame(inspect.currentframe())
    gui.run(run_server=False, single_client=True)
    flask_client = gui._server.test_client()
    with gui.get_server_instance().app_context():
        gui._bind_var_val(var_name, None)
    # Get the jsx once so that the page will be evaluated -> variable will be registered
    sid = _DataScopes._GLOBAL_ID
    file = (io.BytesIO(b"abcdef"), "test.jpg")
    ret = flask_client.post(
        f"/taipy-uploads?client_id={sid}", data={"var_name": var_name, "blob": file}, content_type="multipart/form-data"
    )
    assert ret.status_code == 200
    created_file = pathlib.Path(gui._get_config("upload_folder", tempfile.gettempdir())) / "test.jpg"
    assert created_file.exists()
    file2 = (io.BytesIO(b"abcdef"), "test2.jpg")
    ret = flask_client.post(
        f"/taipy-uploads?client_id={sid}",
        data={"var_name": var_name, "blob": file2, "multiple": "True"},
        content_type="multipart/form-data",
    )
    assert ret.status_code == 200
    created_file = pathlib.Path(gui._get_config("upload_folder", tempfile.gettempdir())) / "test2.jpg"
    assert created_file.exists()
    value = getattr(gui._bindings()._get_all_scopes()[sid], var_name)
    assert len(value) == 2


def test_file_upload_folder(gui: Gui, helpers):
    gui._set_frame(inspect.currentframe())
    gui.run(run_server=False, single_client=True)
    flask_client = gui._server.test_client()

    sid = _DataScopes._GLOBAL_ID
    files = [(io.BytesIO(b"(^~^)"), "cutey.txt"), (io.BytesIO(b"(^~^)"), "cute_nested.txt")]
    folders = [["folder"], ["folder", "nested"]]
    for file, folder in zip(files, folders):
        path = os.path.join(*folder, file[1])
        response = flask_client.post(
            f"/taipy-uploads?client_id={sid}",
            data={"var_name": "cute_varname", "blob": file, "path": path},
            content_type="multipart/form-data",
        )
        assert response.status_code == 200
        assert os.path.isfile(os.path.join(gui._get_config("upload_folder", tempfile.gettempdir()), path))