X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FTestData.java;h=862eb2bc67463437168bc94db062c62d8d3cdb5f;hb=refs%2Fheads%2Fcounter;hp=f795e7765f6dba7e69ce24b01ddf81c0c8873660;hpb=f9106e5166ce84caa2a910aa21fb41efaebcc73f;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java index f795e77..9b38dbc 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -1,105 +1,204 @@ package de.juplo.kafka.wordcount.counter; -import java.util.List; -import java.util.function.BiConsumer; +import de.juplo.kafka.wordcount.splitter.TestInputUser; +import de.juplo.kafka.wordcount.splitter.TestInputWord; +import de.juplo.kafka.wordcount.top10.TestOutputWord; +import de.juplo.kafka.wordcount.top10.TestOutputWordCounter; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; +import java.util.stream.Stream; + +import static de.juplo.kafka.wordcount.counter.CounterStreamProcessor.TYPE; import static org.assertj.core.api.Assertions.assertThat; class TestData { - static void writeInputData(BiConsumer consumer) + static final String PETER = "peter"; + static final String KLAUS = "klaus"; + + static final String WORD_HALLO = "Hallo"; + static final String WORD_MÜSCH = "Müsch"; + static final String WORD_WELT = "Welt"; + static final String WORD_S = "s"; + static final String WORD_BOÄH = "Boäh"; + + static final TestOutputWord PETER_HALLO = TestOutputWord.of(TYPE, PETER, WORD_HALLO); + static final TestOutputWord PETER_WELT = TestOutputWord.of(TYPE, PETER, WORD_WELT); + static final TestOutputWord PETER_BOÄH = TestOutputWord.of(TYPE, PETER, WORD_BOÄH); + static final TestOutputWord KLAUS_MÜSCH = TestOutputWord.of(TYPE, KLAUS, WORD_MÜSCH); + static final TestOutputWord KLAUS_S = TestOutputWord.of(TYPE, KLAUS, WORD_S); + + private static final KeyValue[] INPUT_MESSAGES = new KeyValue[] + { + KeyValue.pair( + TestInputUser.of(PETER), + TestInputWord.of(PETER, WORD_HALLO)), + KeyValue.pair( + TestInputUser.of(KLAUS), + TestInputWord.of(KLAUS, WORD_MÜSCH)), + KeyValue.pair( + TestInputUser.of(PETER), + TestInputWord.of(PETER, WORD_WELT)), + KeyValue.pair( + TestInputUser.of(KLAUS), + TestInputWord.of(KLAUS, WORD_MÜSCH)), + KeyValue.pair( + TestInputUser.of(KLAUS), + TestInputWord.of(KLAUS, WORD_S)), + KeyValue.pair( + TestInputUser.of(PETER), + TestInputWord.of(PETER, WORD_BOÄH)), + KeyValue.pair( + TestInputUser.of(PETER), + TestInputWord.of(PETER, WORD_WELT)), + KeyValue.pair( + TestInputUser.of(PETER), + TestInputWord.of(PETER, WORD_BOÄH)), + KeyValue.pair( + TestInputUser.of(KLAUS), + TestInputWord.of(KLAUS, WORD_S)), + KeyValue.pair( + TestInputUser.of(PETER), + TestInputWord.of(PETER, WORD_BOÄH)), + KeyValue.pair( + TestInputUser.of(KLAUS), + TestInputWord.of(KLAUS, WORD_S)), + }; + + static Stream> getInputMessages() + { + return Stream.of(TestData.INPUT_MESSAGES); + } + + static void assertExpectedMessages(MultiValueMap receivedMessages) + { + expectedMessages().forEach( + (word, counter) -> + assertThat(receivedMessages.get(word)) + .containsExactlyElementsOf(counter)); + } + + static void assertExpectedNumberOfMessagesForWord(MultiValueMap receivedMessages) + { + assertThat(countMessagesForWord(PETER_HALLO, receivedMessages)); + assertThat(countMessagesForWord(PETER_WELT, receivedMessages)); + assertThat(countMessagesForWord(PETER_BOÄH, receivedMessages)); + assertThat(countMessagesForWord(KLAUS_MÜSCH, receivedMessages)); + assertThat(countMessagesForWord(KLAUS_S, receivedMessages)); + } + + private static int countMessagesForWord(TestOutputWord word, MultiValueMap messagesForUsers) + { + return messagesForUsers.get(word) == null + ? 0 + : messagesForUsers.get(word).size(); + } + + static void assertExpectedState(ReadOnlyKeyValueStore store) { - consumer.accept( - "peter", - Word.of("peter","Hallo")); - consumer.accept( - "klaus", - Word.of("klaus","Müsch")); - consumer.accept( - "peter", - Word.of("peter","Welt")); - consumer.accept( - "klaus", - Word.of("klaus","Müsch")); - consumer.accept( - "klaus", - Word.of("klaus","s")); - consumer.accept( - "peter", - Word.of("peter","Boäh")); - consumer.accept( - "peter", - Word.of("peter","Welt")); - consumer.accept( - "peter", - Word.of("peter","Boäh")); - consumer.accept( - "klaus", - Word.of("klaus","s")); - consumer.accept( - "peter", - Word.of("peter","Boäh")); - consumer.accept( - "klaus", - Word.of("klaus","s")); + assertWordCountEqualsWordCountFromLastMessage(PETER_HALLO, store.get(wordOf(PETER_HALLO))); + assertWordCountEqualsWordCountFromLastMessage(PETER_WELT, store.get(wordOf(PETER_WELT))); + assertWordCountEqualsWordCountFromLastMessage(PETER_BOÄH, store.get(wordOf(PETER_BOÄH))); + assertWordCountEqualsWordCountFromLastMessage(KLAUS_MÜSCH, store.get(wordOf(KLAUS_MÜSCH))); + assertWordCountEqualsWordCountFromLastMessage(KLAUS_S, store.get(wordOf(KLAUS_S))); } - static void assertExpectedResult(List receivedMessages) + private static Word wordOf(TestOutputWord testOutputWord) { - assertThat(receivedMessages).hasSize(11); - assertThat(receivedMessages).containsSubsequence( - expectedMessages[0]); // Hallo - assertThat(receivedMessages).containsSubsequence( - expectedMessages[1], - expectedMessages[3]); // Müsch - assertThat(receivedMessages).containsSubsequence( - expectedMessages[2], - expectedMessages[6]); - assertThat(receivedMessages).containsSubsequence( - expectedMessages[4], - expectedMessages[8], - expectedMessages[10]); // s - assertThat(receivedMessages).containsSubsequence( - expectedMessages[5], - expectedMessages[7], - expectedMessages[9]); // Boäh + return Word.of( + testOutputWord.getChannel(), + testOutputWord.getKey()); } - static Message[] expectedMessages = + static void assertExpectedLastMessagesForWord(MultiValueMap receivedMessages) { - Message.of( - "{\"user\":\"peter\",\"word\":\"Hallo\"}", - "{\"user\":\"peter\",\"word\":\"Hallo\",\"count\":1}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"Müsch\"}", - "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":1}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Welt\"}", - "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":1}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"Müsch\"}", - "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":2}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":1}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":1}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Welt\"}", - "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":2}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":2}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":2}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":3}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":3}"), + assertWordCountEqualsWordCountFromLastMessage(PETER_HALLO, getLastMessageFor(PETER_HALLO, receivedMessages)); + assertWordCountEqualsWordCountFromLastMessage(PETER_WELT, getLastMessageFor(PETER_WELT, receivedMessages)); + assertWordCountEqualsWordCountFromLastMessage(PETER_BOÄH, getLastMessageFor(PETER_BOÄH, receivedMessages)); + assertWordCountEqualsWordCountFromLastMessage(KLAUS_MÜSCH, getLastMessageFor(KLAUS_MÜSCH, receivedMessages)); + assertWordCountEqualsWordCountFromLastMessage(KLAUS_S, getLastMessageFor(KLAUS_S, receivedMessages)); + } + + private static void assertWordCountEqualsWordCountFromLastMessage( + TestOutputWord word, + Long counter) + { + TestOutputWordCounter testOutputWordCounter = TestOutputWordCounter.of( + word.getChannel(), + word.getKey(), + counter); + assertWordCountEqualsWordCountFromLastMessage(word, testOutputWordCounter); + } + + private static void assertWordCountEqualsWordCountFromLastMessage( + TestOutputWord word, + TestOutputWordCounter counter) + { + assertThat(counter).isEqualTo(getLastMessageFor(word)); + } + + private static TestOutputWordCounter getLastMessageFor(TestOutputWord word) + { + return getLastMessageFor(word, expectedMessages()); + } + + private static TestOutputWordCounter getLastMessageFor( + TestOutputWord user, + MultiValueMap messagesForWord) + { + return messagesForWord + .get(user) + .stream() + .reduce(null, (left, right) -> right); + } + + private static final KeyValue[] EXPECTED_MESSAGES = new KeyValue[] + { + KeyValue.pair( + PETER_HALLO, + TestOutputWordCounter.of(PETER, WORD_HALLO,1)), + KeyValue.pair( + KLAUS_MÜSCH, + TestOutputWordCounter.of(KLAUS, WORD_MÜSCH,1)), + KeyValue.pair( + PETER_WELT, + TestOutputWordCounter.of(PETER, WORD_WELT,1)), + KeyValue.pair( + KLAUS_MÜSCH, + TestOutputWordCounter.of(KLAUS, WORD_MÜSCH,2)), + KeyValue.pair( + KLAUS_S, + TestOutputWordCounter.of(KLAUS, WORD_S,1)), + KeyValue.pair( + PETER_BOÄH, + TestOutputWordCounter.of(PETER, WORD_BOÄH,1)), + KeyValue.pair( + PETER_WELT, + TestOutputWordCounter.of(PETER, WORD_WELT,2)), + KeyValue.pair( + PETER_BOÄH, + TestOutputWordCounter.of(PETER, WORD_BOÄH,2)), + KeyValue.pair( + KLAUS_S, + TestOutputWordCounter.of(KLAUS, WORD_S,2)), + KeyValue.pair( + PETER_BOÄH, + TestOutputWordCounter.of(PETER, WORD_BOÄH,3)), + KeyValue.pair( + KLAUS_S, + TestOutputWordCounter.of(KLAUS, WORD_S,3)), }; + + static MultiValueMap expectedMessages() + { + MultiValueMap expectedMessages = new LinkedMultiValueMap<>(); + Stream + .of(EXPECTED_MESSAGES) + .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value)); + return expectedMessages; + } }