Function test_background_task_reset

tests/units/test_state.py:2578–2608

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_base_state_event_processor: The event processor. state_manager: A state manager instance.

(
    mock_app: rx.App,
    token: str,
    mock_base_state_event_processor: BaseStateEventProcessor,
    state_manager: StateManager,
)

@pytest.mark.asyncio
async def test_background_task_reset(
    mock_app: rx.App,
    token: str,
    mock_base_state_event_processor: BaseStateEventProcessor,
    state_manager: StateManager,
):
    """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_base_state_event_processor: The event processor.
        state_manager: A state manager instance.
    """
    async with mock_base_state_event_processor as processor:
        await processor.enqueue(
            token,
            Event(
                name=f"{BackgroundTaskState.get_full_name()}.background_task_reset",
                payload={},
            ),
        )

    if environment.REFLEX_OPLOCK_ENABLED.get():
        await state_manager.close()

    background_task_state = await state_manager.get_state(
        BaseStateToken(ident=token, cls=BackgroundTaskState)
    )
    assert isinstance(background_task_state, BackgroundTaskState)
    assert background_task_state.order == ["reset"]


@pytest.mark.asyncio