Instance of state manager parametrized for redis and in-process. Args: request: pytest request object. Yields: A state manager instance
(request)
| 1413 | |
| 1414 | @pytest.fixture(scope="function", params=["in_process", "redis"]) |
| 1415 | def state_manager(request) -> Generator[StateManager, None, None]: |
| 1416 | """Instance of state manager parametrized for redis and in-process. |
| 1417 | |
| 1418 | Args: |
| 1419 | request: pytest request object. |
| 1420 | |
| 1421 | Yields: |
| 1422 | A state manager instance |
| 1423 | """ |
| 1424 | state_manager = StateManager.create(state=TestState) |
| 1425 | if request.param == "redis": |
| 1426 | if not isinstance(state_manager, StateManagerRedis): |
| 1427 | pytest.skip("Test requires redis") |
| 1428 | else: |
| 1429 | # explicitly NOT using redis |
| 1430 | state_manager = StateManagerMemory(state=TestState) |
| 1431 | assert not state_manager._states_locks |
| 1432 | |
| 1433 | yield state_manager |
| 1434 | |
| 1435 | if isinstance(state_manager, StateManagerRedis): |
| 1436 | asyncio.get_event_loop().run_until_complete(state_manager.close()) |
| 1437 | |
| 1438 | |
| 1439 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected