Function test_upload_file

tests/integration/test_upload.py:579–632

Submit a file upload and check that it arrived on the backend. Args: tmp_path: pytest tmp_path fixture upload_file: harness for UploadFile app. driver: WebDriver instance. progress_dicts: callable to retrieve progress dictionary elements. upload_root_id:

(
    tmp_path,
    upload_file: AppHarness,
    driver: WebDriver,
    progress_dicts: Callable[[], list[WebElement]],
    upload_root_id: str | None,
)

@pytest.mark.parametrize("upload_root_id", [None, "secondary"])
def test_upload_file(
    tmp_path,
    upload_file: AppHarness,
    driver: WebDriver,
    progress_dicts: Callable[[], list[WebElement]],
    upload_root_id: str | None,
):
    """Submit a file upload and check that it arrived on the backend.

    Args:
        tmp_path: pytest tmp_path fixture
        upload_file: harness for UploadFile app.
        driver: WebDriver instance.
        progress_dicts: callable to retrieve progress dictionary elements.
        upload_root_id: ID of the upload root element, or None for the default.
    """
    assert upload_file.app_instance is not None
    poll_for_token(driver, upload_file)
    clear_btn = driver.find_element(By.ID, "clear_uploads")
    clear_btn.click()

    suffix = f"_{upload_root_id}" if upload_root_id else ""

    upload_box = get_upload_box(driver, upload_root_id=upload_root_id)
    assert upload_box
    upload_button = driver.find_element(By.ID, f"upload_button{suffix}")
    assert upload_button

    exp_name = "test.txt"
    exp_contents = "test file contents!"
    target_file = tmp_path / exp_name
    target_file.write_text(exp_contents)

    upload_box.send_keys(str(target_file))
    upload_button.click()

    # check that the selected files are displayed
    selected_files = driver.find_element(By.ID, f"selected_files{suffix}")
    assert Path(selected_files.text).name == Path(exp_name).name

    # Wait for the upload to complete.
    upload_done = driver.find_element(By.ID, "upload_done")
    assert upload_file.poll_for_value(upload_done, exp_not_equal="false") == "true"

    if upload_root_id == "secondary":
        event_order_displayed = driver.find_element(By.ID, "event-order")
        AppHarness.expect(lambda: "chain_event" in event_order_displayed.text)
        final_progress = progress_dicts()
        assert len(final_progress) > 0
        assert json.loads(final_progress[-1].text)["progress"] == 1

    # look up the backend state and assert on uploaded contents
    actual_contents = (rx.get_upload_dir() / exp_name).read_text()
    assert actual_contents == exp_contents


@pytest.mark.asyncio