Function state_manager

tests/units/conftest.py:230–260

Instance of state manager parametrized for redis and in-process. Args: request: pytest request object. mock_root_event_context: The mock root event context to use for the state manager. Yields: A state manager instance

(
    request: pytest.FixtureRequest, mock_root_event_context: EventContext
)

    loop_scope="function", scope="function", params=["in_process", "disk", "redis"]
)
async def state_manager(
    request: pytest.FixtureRequest, mock_root_event_context: EventContext
) -> AsyncGenerator[StateManager, None]:
    """Instance of state manager parametrized for redis and in-process.

    Args:
        request: pytest request object.
        mock_root_event_context: The mock root event context to use for the state manager.

    Yields:
        A state manager instance
    """
    state_manager = StateManager.create()
    if request.param == "redis":
        if not isinstance(state_manager, StateManagerRedis):
            state_manager = StateManagerRedis(redis=mock_redis())
    elif request.param == "disk":
        # explicitly NOT using redis
        state_manager = StateManagerDisk()
        assert not state_manager._states_locks
    else:
        state_manager = StateManagerMemory()
        assert not state_manager._states_locks

    orig_state_manager = mock_root_event_context.state_manager
    object.__setattr__(mock_root_event_context, "state_manager", state_manager)

    yield state_manager

    await state_manager.close()
    object.__setattr__(mock_root_event_context, "state_manager", orig_state_manager)


@pytest.fixture