Function test_upload_file

tests/test_app.py:738–794

Test that file upload works correctly. Args: tmp_path: Temporary path. state: The state class. delta: Expected delta token: a Token. mocker: pytest mocker object.

(tmp_path, state, delta, token: str, mocker)

    ],
)
async def test_upload_file(tmp_path, state, delta, token: str, mocker):
    """Test that file upload works correctly.

    Args:
        tmp_path: Temporary path.
        state: The state class.
        delta: Expected delta
        token: a Token.
        mocker: pytest mocker object.
    """
    mocker.patch(
        "nextpy.state.State.class_subclasses",
        {state if state is FileUploadState else FileStateBase1},
    )
    state._tmp_path = tmp_path
    # The App state must be the "root" of the state tree
    app = App(state=State)
    app.event_namespace.emit = AsyncMock()  # type: ignore
    current_state = await app.state_manager.get_state(token)
    data = b"This is binary data"

    # Create a binary IO object and write data to it
    bio = io.BytesIO()
    bio.write(data)

    state_name = state.get_full_name().partition(".")[2] or state.get_name()
    request_mock = unittest.mock.Mock()
    request_mock.headers = {
        "nextpy-client-token": token,
        "nextpy-event-handler": f"state.{state_name}.multi_handle_upload",
    }

    file1 = UploadFile(
        filename=f"image1.jpg",
        file=bio,
    )
    file2 = UploadFile(
        filename=f"image2.jpg",
        file=bio,
    )
    upload_fn = upload(app)
    streaming_response = await upload_fn(request_mock, [file1, file2])
    async for state_update in streaming_response.body_iterator:
        assert (
            state_update
            == StateUpdate(delta=delta, events=[], final=True).json() + "\n"
        )

    current_state = await app.state_manager.get_state(token)
    state_dict = current_state.dict()[state.get_full_name()]
    assert state_dict["img_list"] == [
        "image1.jpg",
        "image2.jpg",
    ]

    if isinstance(app.state_manager, StateManagerRedis):
        await app.state_manager.close()