### Question:
WindowFetchKeyRangeQuery implements LocalStoreQuery<Windowed<K>, V> { @Override public Try<List<KV<Windowed<K>, V>>> execute(final KafkaStreamsContainer container, final long limit) { final LocalStoreAccessor<ReadOnlyWindowStore<K, V>> accessor = container.localWindowStore(store); final Reader<ReadOnlyWindowStore<K, V>, List<KV<Windowed<K>, V>>> reader = reader(keyFrom, keyTo, timeFrom, timeTo) .map(iterator -> LocalStoreQuery.toKeyValueListAndClose(iterator, limit)); return new LocalStoreQueryExecutor<>(accessor).execute(reader); } WindowFetchKeyRangeQuery(final String storeName,
                             final K fromKey,
                             final K toKey,
                             final Instant fromTime,
                             final Instant toTime); @Override StoreType storeType(); @Override StoreOperation operationType(); @Override Try<List<KV<Windowed<K>, V>>> execute(final KafkaStreamsContainer container, final long limit);  }### Answer:
@Test public void shouldFetchKeyRange() { WindowFetchKeyRangeQuery<String, String> query = new WindowFetchKeyRangeQuery<>( STORE_NAME, KEY_FROM, KEY_TO, Instant.MIN, Instant.MAX); final var mkContainer = Mockito.mock(KafkaStreamsContainer.class); Windowed<String> windowed = new Windowed<>(KEY_FROM, new TimeWindow(0L, 1L)); ReadOnlyWindowStore store = mock(ReadOnlyWindowStore.class); when(store.fetch(KEY_FROM, KEY_TO, Instant.MIN, Instant.MAX)).thenReturn( new InMemoryKeyValueIterator<>(windowed, "value") ); when(mkContainer.localWindowStore(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(KV.of(windowed, "value"), result.get().get(0)); }