Function test_process_event_generator

tests/units/test_state.py:1013–1037

Test event handlers that generate multiple updates. Args: token: A token. mock_base_state_event_processor: The event processor. emitted_deltas: List to capture emitted deltas.

(
    token: str,
    mock_base_state_event_processor: BaseStateEventProcessor,
    emitted_deltas: list,
)

@pytest.mark.asyncio
async def test_process_event_generator(
    token: str,
    mock_base_state_event_processor: BaseStateEventProcessor,
    emitted_deltas: list,
):
    """Test event handlers that generate multiple updates.

    Args:
        token: A token.
        mock_base_state_event_processor: The event processor.
        emitted_deltas: List to capture emitted deltas.
    """
    event = Event(
        name=f"{GenState.get_full_name()}.go",
        payload={"c": 5},
    )
    async with mock_base_state_event_processor as processor:
        await processor.enqueue(token, event)
    # Generator yields 5 deltas (one per increment).
    assert len(emitted_deltas) == 5
    for count, (delta_token, delta) in enumerate(emitted_deltas, 1):
        assert delta_token == token
        assert delta == {
            GenState.get_full_name(): {"value" + FIELD_MARKER: count},
        }


def test_get_client_token(test_state, router_data):