### Question:
SessionFetchQuery extends KeyedLocalStoreQuery<K, Windowed<K>, V> { @Override public Try<List<KV<Windowed<K>, V>>> execute(final KafkaStreamsContainer container, final long limit) { final LocalStoreAccessor<ReadOnlySessionStore<K, V>> accessor = container.localSessionStore(storeName()); final Reader<ReadOnlySessionStore<K, V>, List<KV<Windowed<K>, V>>> reader = reader(key()) .map(iterator -> LocalStoreQuery.toKeyValueListAndClose(iterator, limit)); return new LocalStoreQueryExecutor<>(accessor).execute(reader); } SessionFetchQuery(final String store,
                      final K key,
                      final Serializer<K> keySserializer); @Override StoreType storeType(); @Override StoreOperation operationType(); @Override Try<List<KV<Windowed<K>, V>>> execute(final KafkaStreamsContainer container, final long limit);  }### Answer:
@Test public void shouldFetchGivenKeyAndTime() { SessionFetchQuery<String, String> query = new SessionFetchQuery<>( STORE_NAME, KEY, Serdes.String().serializer()); final var mkContainer = Mockito.mock(KafkaStreamsContainer.class); ReadOnlySessionStore store = mock(ReadOnlySessionStore.class); when(store.fetch(KEY)).thenReturn(new InMemoryKeyValueIterator<>(new Windowed<>("key", null), "value")); when(mkContainer.localSessionStore(matches(STORE_NAME))).thenReturn(new LocalStoreAccessor<>(() -> store)); Try<List<KV<Windowed<String>, String>>> result = query.execute(mkContainer); Assertions.assertEquals(1, result.get().size()); Assertions.assertEquals("value", result.get().get(0).value()); }