Function test_cancel_upload

tests/integration/test_upload.py:809–869

Submit a large file upload and cancel it. Covers both the standard upload form and the streaming-chunk upload form; the latter additionally writes a partial file to a subdirectory under ``rx.get_upload_dir()`` which is verified to be smaller than the source. Args: request:

(
    request: pytest.FixtureRequest,
    tmp_path,
    upload_file: AppHarness,
    driver: WebDriver,
    upload_root_id: str,
    progress_fixture: str,
    exp_name: str,
    file_size_bytes: int,
    partial_subdir: str | None,
)

    ],
)
async def test_cancel_upload(
    request: pytest.FixtureRequest,
    tmp_path,
    upload_file: AppHarness,
    driver: WebDriver,
    upload_root_id: str,
    progress_fixture: str,
    exp_name: str,
    file_size_bytes: int,
    partial_subdir: str | None,
):
    """Submit a large file upload and cancel it.

    Covers both the standard upload form and the streaming-chunk upload form;
    the latter additionally writes a partial file to a subdirectory under
    ``rx.get_upload_dir()`` which is verified to be smaller than the source.

    Args:
        request: pytest request fixture, used to resolve the parametrized progress fixture.
        tmp_path: pytest tmp_path fixture
        upload_file: harness for UploadFile app.
        driver: WebDriver instance.
        upload_root_id: id of the rx.upload.root component to drive; also the suffix used for its upload/cancel button ids.
        progress_fixture: name of the fixture providing the progress dicts callable.
        exp_name: name of the file to upload.
        file_size_bytes: size of the file to create, in bytes.
        partial_subdir: subdirectory under the upload dir where a partial file is expected, or None if no partial file is written.
    """
    assert upload_file.app_instance is not None
    progress_dicts: Callable[[], list[WebElement]] = request.getfixturevalue(
        progress_fixture
    )
    poll_for_token(driver, upload_file)

    upload_box = get_upload_box(driver, upload_root_id=upload_root_id)
    upload_button = driver.find_element(By.ID, f"upload_button_{upload_root_id}")
    cancel_button = driver.find_element(By.ID, f"cancel_button_{upload_root_id}")

    target_file = tmp_path / exp_name
    with target_file.open("wb") as f:
        f.seek(file_size_bytes)
        f.write(b"0")

    upload_box.send_keys(str(target_file))
    upload_button.click()
    # Check for at least 2 progress updates to ensure the upload is active.
    AppHarness.expect(lambda: len(progress_dicts()) >= 2)
    cancel_button.click()

    # There should never be a final progress record for a cancelled upload.
    for p in await poll_for_stopped_progress(progress_dicts):
        assert p["progress"] != 1

    assert not (rx.get_upload_dir() / exp_name).exists()

    if partial_subdir is not None:
        partial_path = rx.get_upload_dir() / partial_subdir / exp_name
        assert partial_path.exists()