Function test_background_task_reset

tests/test_state.py:1891–1922

Test that a background task calling reset is protected by the state proxy. Args: mock_app: An app that will be returned by `get_app()` token: A token.

(mock_app: xt.App, token: str)

@pytest.mark.asyncio
async def test_background_task_reset(mock_app: xt.App, token: str):
    """Test that a background task calling reset is protected by the state proxy.

    Args:
        mock_app: An app that will be returned by `get_app()`
        token: A token.
    """
    router_data = {"query": {}}
    mock_app.state_manager.state = mock_app.state = BackgroundTaskState
    async for update in xt.app.process(  # type: ignore
        mock_app,
        Event(
            token=token,
            name=f"{BackgroundTaskState.get_name()}.background_task_reset",
            router_data=router_data,
            payload={},
        ),
        sid="",
        headers={},
        client_ip="",
    ):
        # background task returns empty update immediately
        assert update == StateUpdate()

    # Explicit wait for background tasks
    for task in tuple(mock_app.background_tasks):
        await task
    assert not mock_app.background_tasks

    assert (await mock_app.state_manager.get_state(token)).order == [
        "reset",
    ]


@pytest.mark.asyncio