Function test_process_events

tests/units/test_app.py:1987–2028

Test that an event is processed properly and that it is postprocessed n+1 times. Also check that the processing flag of the last stateupdate is set to False. Args: token: a Token. app_module_mock: The mock for the app module, used to patch the app instance. mock_

(
    token: str,
    app_module_mock: unittest.mock.Mock,
    mock_base_state_event_processor: BaseStateEventProcessor,
    mock_root_event_context: EventContext,
    emitted_deltas: list[tuple[str, dict[str, dict[str, Any]]]],
)

@pytest.mark.asyncio
async def test_process_events(
    token: str,
    app_module_mock: unittest.mock.Mock,
    mock_base_state_event_processor: BaseStateEventProcessor,
    mock_root_event_context: EventContext,
    emitted_deltas: list[tuple[str, dict[str, dict[str, Any]]]],
):
    """Test that an event is processed properly and that it is postprocessed
    n+1 times. Also check that the processing flag of the last stateupdate is set to
    False.

    Args:
        token: a Token.
        app_module_mock: The mock for the app module, used to patch the app instance.
        mock_base_state_event_processor: BaseStateEventProcessor Fixture.
        mock_root_event_context: The mock for the root event context, used to patch the app
            state manager.
        emitted_deltas: List to store emitted deltas.
    """
    event = Event(
        name=f"{GenState.get_name()}.go",
        payload={"c": 5},
        router_data={},
    )

    async with mock_base_state_event_processor as processor:
        await processor.enqueue(
            token,
            event,
        )

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

    gen_state = await mock_root_event_context.state_manager.get_state(
        BaseStateToken(ident=token, cls=GenState),
    )
    assert isinstance(gen_state, GenState)
    assert gen_state.value == 5
    assert len(emitted_deltas) == 5

    await mock_root_event_context.state_manager.close()


@pytest.fixture