Function test_process_event_substate

tests/units/test_state.py:961–1009

Test processing an event on a substate. 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_substate(
    token: str,
    mock_base_state_event_processor: BaseStateEventProcessor,
    emitted_deltas: list,
):
    """Test processing an event on a substate.

    Args:
        token: A token.
        mock_base_state_event_processor: The event processor.
        emitted_deltas: List to capture emitted deltas.
    """
    # Events should bubble down to the substate.
    event = Event(
        name=f"{ChildState.get_full_name()}.change_both",
        payload={"value": "hi", "count": 12},
    )
    async with mock_base_state_event_processor as processor:
        await processor.enqueue(token, event)
    assert emitted_deltas == [
        (
            token,
            {
                ChildState.get_full_name(): {
                    "value" + FIELD_MARKER: "HI",
                    "count" + FIELD_MARKER: 24,
                },
                GrandchildState3.get_full_name(): {"computed" + FIELD_MARKER: ""},
            },
        )
    ]
    emitted_deltas.clear()

    # Test with the grandchild state.
    event = Event(
        name=f"{GrandchildState.get_full_name()}.set_value2",
        payload={"value": "new"},
    )
    async with mock_base_state_event_processor as processor:
        await processor.enqueue(token, event)
    assert emitted_deltas == [
        (
            token,
            {
                GrandchildState.get_full_name(): {"value2" + FIELD_MARKER: "new"},
                GrandchildState3.get_full_name(): {"computed" + FIELD_MARKER: ""},
            },
        )
    ]


@pytest.mark.asyncio