Test persistence of session pool state to KVS and validate stored data integrity.
(event_manager: EventManager, kvs_name: str | None)
| 114 | |
| 115 | @pytest.mark.parametrize('kvs_name', [KVS_NAME, None]) |
| 116 | async def test_session_pool_persist(event_manager: EventManager, kvs_name: str | None) -> None: |
| 117 | """Test persistence of session pool state to KVS and validate stored data integrity.""" |
| 118 | service_locator.set_event_manager(event_manager) |
| 119 | |
| 120 | async with SessionPool( |
| 121 | max_pool_size=MAX_POOL_SIZE, |
| 122 | persistence_enabled=True, |
| 123 | persist_state_kvs_name=kvs_name, |
| 124 | persist_state_key=PERSIST_STATE_KEY, |
| 125 | ) as sp: |
| 126 | # Emit persist state event and wait for the persistence to complete |
| 127 | event_manager.emit(event=Event.PERSIST_STATE, event_data=EventPersistStateData(is_migrating=False)) |
| 128 | await event_manager.wait_for_all_listeners_to_complete() |
| 129 | |
| 130 | # Get the persisted state from the key-value store |
| 131 | kvs = await KeyValueStore.open(name=kvs_name) |
| 132 | previous_state = await kvs.get_value(key=PERSIST_STATE_KEY) |
| 133 | assert isinstance(previous_state, dict) |
| 134 | sp_model = SessionPoolModel(**previous_state) |
| 135 | |
| 136 | # Check if the state is correctly persisted |
| 137 | assert sp_model.session_count == sp.session_count |
| 138 | assert sp_model.usable_session_count == sp.usable_session_count |
| 139 | assert sp_model.retired_session_count == sp.retired_session_count |
| 140 | |
| 141 | # Check if all the sessions are correctly persisted |
| 142 | for kvs_session in sp_model.sessions.values(): |
| 143 | session = await sp.get_session_by_id(kvs_session.id) |
| 144 | assert kvs_session == session |
| 145 | |
| 146 | |
| 147 | async def test_session_pool_persist_and_restore(event_manager: EventManager, kvs: KeyValueStore) -> None: |
nothing calls this directly
no test coverage detected