Function test_state_manager_contend

tests/test_state.py:1468–1497

Multiple coroutines attempting to access the same state. Args: state_manager: A state manager instance. token: A token.

(state_manager: StateManager, token: str)

@pytest.mark.asyncio
async def test_state_manager_contend(state_manager: StateManager, token: str):
    """Multiple coroutines attempting to access the same state.

    Args:
        state_manager: A state manager instance.
        token: A token.
    """
    n_coroutines = 10
    exp_num1 = 10

    async with state_manager.modify_state(token) as state:
        state.num1 = 0

    async def _coro():
        async with state_manager.modify_state(token) as state:
            await asyncio.sleep(0.01)
            state.num1 += 1

    tasks = [asyncio.create_task(_coro()) for _ in range(n_coroutines)]

    for f in asyncio.as_completed(tasks):
        await f

    assert (await state_manager.get_state(token)).num1 == exp_num1

    if isinstance(state_manager, StateManagerRedis):
        assert (await state_manager.redis.get(f"{token}_lock")) is None
    elif isinstance(state_manager, StateManagerMemory):
        assert token in state_manager._states_locks
        assert not state_manager._states_locks[token].locked()


@pytest.fixture(scope="function")