Function poll_for_stopped_progress

tests/integration/test_upload.py:500–535

Poll for progress dictionaries to stop updating. Args: get_progress_dicts: A callable that returns the list of progress dictionary elements. iterations: Maximum number of iterations to poll for. delay: Delay in seconds between iterations. stable_iterations: Numbe

(
    get_progress_dicts: Callable[[], list[WebElement]],
    iterations: int = 20,
    delay: int | float = 1.0,
    stable_iterations: int = 3,
)

async def poll_for_stopped_progress(
    get_progress_dicts: Callable[[], list[WebElement]],
    iterations: int = 20,
    delay: int | float = 1.0,
    stable_iterations: int = 3,
) -> list[dict]:
    """Poll for progress dictionaries to stop updating.

    Args:
        get_progress_dicts: A callable that returns the list of progress dictionary elements.
        iterations: Maximum number of iterations to poll for.
        delay: Delay in seconds between iterations.
        stable_iterations: Number of consecutive iterations with no new progress dictionaries before considering it stopped.

    Returns:
        The stable list of deserialized progress dicts.

    Raises:
        TimeoutError: If progress dictionaries keep updating beyond the maximum iterations.
    """
    remaining_stable_iterations = stable_iterations
    last_progress_dicts_content = [p.text for p in get_progress_dicts()]
    for _ in range(iterations):
        await asyncio.sleep(delay)
        progress_dicts_content = [p.text for p in get_progress_dicts()]
        if progress_dicts_content == last_progress_dicts_content:
            # Content remains stable, decrement remaining_stable_iterations
            remaining_stable_iterations -= 1
            if remaining_stable_iterations <= 0:
                return [json.loads(t) for t in last_progress_dicts_content]
        else:
            # Progress dicts content changed, we must start over counting stable iterations.
            remaining_stable_iterations = stable_iterations
            last_progress_dicts_content = progress_dicts_content
    msg = f"Progress dictionaries kept updating after {iterations} iterations ({iterations * delay} seconds)."
    raise TimeoutError(msg)


def poll_for_token(driver: WebDriver, upload_file: AppHarness) -> str: