Function test_background_task_no_block

tests/test_state.py:1773–1887

Test that a background task does not block other events. 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_no_block(mock_app: xt.App, token: str):
    """Test that a background task does not block other events.

    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",
            router_data=router_data,
            payload={},
        ),
        sid="",
        headers={},
        client_ip="",
    ):
        # background task returns empty update immediately
        assert update == StateUpdate()

    # wait for the coroutine to start
    await asyncio.sleep(0.5 if CI else 0.1)
    assert len(mock_app.background_tasks) == 1

    # Process another normal event
    async for update in xt.app.process(  # type: ignore
        mock_app,
        Event(
            token=token,
            name=f"{BackgroundTaskState.get_name()}.other",
            router_data=router_data,
            payload={},
        ),
        sid="",
        headers={},
        client_ip="",
    ):
        # other task returns delta
        assert update == StateUpdate(
            delta={
                BackgroundTaskState.get_name(): {
                    "order": [
                        "background_task:start",
                        "other",
                    ],
                    "computed_order": [
                        "background_task:start",
                        "other",
                    ],
                }
            }
        )

    # Explicit wait for background tasks