Function test_state_proxy

tests/units/test_state.py:2293–2415

Test that the state proxy works. Args: grandchild_state: A grandchild state. token: A token. attached_mock_base_state_event_processor: The event processor attached for this test. emitted_deltas: A list to capture emitted deltas. attached_mock_event_contex

(
    grandchild_state: GrandchildState,
    token: str,
    attached_mock_base_state_event_processor: BaseStateEventProcessor,
    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
    attached_mock_event_context: EventContext,
)

@pytest.mark.asyncio
async def test_state_proxy(
    grandchild_state: GrandchildState,
    token: str,
    attached_mock_base_state_event_processor: BaseStateEventProcessor,
    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
    attached_mock_event_context: EventContext,
):
    """Test that the state proxy works.

    Args:
        grandchild_state: A grandchild state.
        token: A token.
        attached_mock_base_state_event_processor: The event processor attached for this test.
        emitted_deltas: A list to capture emitted deltas.
        attached_mock_event_context: The event context attached for this test.
    """
    child_state = grandchild_state.parent_state
    assert child_state is not None
    parent_state = child_state.parent_state
    assert parent_state is not None
    router_data = RouterData.from_router_data({
        "query": {},
        "token": token,
        "sid": "test_sid",
    })
    grandchild_state.router = router_data
    state_manager = attached_mock_event_context.state_manager
    if isinstance(state_manager, (StateManagerMemory, StateManagerDisk)):
        state_manager.states[parent_state.router.session.client_token] = parent_state
    elif isinstance(state_manager, StateManagerRedis):
        pickle_state = parent_state._serialize()
        if pickle_state:
            await state_manager.redis.set(
                str(
                    BaseStateToken(
                        ident=parent_state.router.session.client_token,
                        cls=type(parent_state),
                    )
                ),
                pickle_state,
                ex=state_manager.token_expiration,
            )

    sp = StateProxy(grandchild_state)
    assert sp.__wrapped__ == grandchild_state
    assert sp._self_substate_path == tuple(grandchild_state.get_full_name().split("."))
    assert not sp._self_mutable
    assert sp._self_actx is None

    # cannot use normal contextmanager protocol
    with pytest.raises(TypeError), sp:
        pass

    with pytest.raises(ImmutableStateError):
        # cannot directly modify state proxy outside of async context
        sp.value2 = "16"

    with pytest.raises(ImmutableStateError):