Function test_upload_file

tests/units/test_app.py:993–1056

Test that file upload works correctly. Args: tmp_path: Temporary path. state: The state class. delta: Expected delta after processing all files. token: a Token. mocker: pytest mocker object. attached_mock_base_state_event_processor: BaseStateEvent

(
    tmp_path: Path,
    state,
    delta,
    token: str,
    mocker: MockerFixture,
    attached_mock_base_state_event_processor: BaseStateEventProcessor,
    mock_root_event_context: EventContext,
    clean_registration_context: RegistrationContext,
)

    ],
)
async def test_upload_file(
    tmp_path: Path,
    state,
    delta,
    token: str,
    mocker: MockerFixture,
    attached_mock_base_state_event_processor: BaseStateEventProcessor,
    mock_root_event_context: EventContext,
    clean_registration_context: RegistrationContext,
):
    """Test that file upload works correctly.

    Args:
        tmp_path: Temporary path.
        state: The state class.
        delta: Expected delta after processing all files.
        token: a Token.
        mocker: pytest mocker object.
        attached_mock_base_state_event_processor: BaseStateEventProcessor Fixture attached to the app instance to capture emitted events.
        mock_root_event_context: The mocked root event context, for accessing state_manager.
        clean_registration_context: Fixture to ensure clean registration context for each test, preventing cross-test contamination of state subclasses.
    """
    clean_registration_context.register_base_state(state)
    app = Mock(
        event_processor=attached_mock_base_state_event_processor,
    )
    async with mock_root_event_context.state_manager.modify_state(
        BaseStateToken(ident=token, cls=state)
    ) as root_state:
        (await root_state.get_state(state))._tmp_path = tmp_path
    data = b"This is binary data"

    request_mock = unittest.mock.Mock()
    request_mock.headers = {
        "reflex-client-token": token,
        "reflex-event-handler": f"{state.get_full_name()}.multi_handle_upload",
    }

    file1 = UploadFile(
        filename="image1.jpg",
        file=io.BytesIO(data),
    )
    file2 = UploadFile(
        filename="image2.jpg",
        file=io.BytesIO(data),
    )

    async def form():  # noqa: RUF029
        return FormData([("files", file1), ("files", file2)])

    request_mock.form = form

    upload_fn = upload(app)
    streaming_response = await upload_fn(request_mock)
    assert isinstance(streaming_response, StreamingResponse)
    # Handler yields after each file, producing intermediate + final updates.
    updates = []
    async for state_update in streaming_response.body_iterator: