X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Ftop10%2FTestData.java;h=7a3a27edddb2722dae8dbf4f53d59c9ebcddd58f;hb=478b25d6700bd0a48b55647f09386f3b1d59fa29;hp=d8c603ae1ba6c688376456b47702de0a91bb0b1c;hpb=e32a74e6a989ba8fc1061cffd1bed814edb4f72f;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java index d8c603a..7a3a27e 100644 --- a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java @@ -2,16 +2,15 @@ package de.juplo.kafka.wordcount.top10; import de.juplo.kafka.wordcount.counter.TestCounter; import de.juplo.kafka.wordcount.counter.TestWord; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.Headers; +import de.juplo.kafka.wordcount.query.TestEntry; +import de.juplo.kafka.wordcount.query.TestRanking; +import de.juplo.kafka.wordcount.query.TestUser; 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.Map; -import java.util.Properties; -import java.util.stream.Collectors; +import java.util.Arrays; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -19,10 +18,15 @@ import static org.assertj.core.api.Assertions.assertThat; class TestData { - static final User PETER = User.of("peter"); - static final User KLAUS = User.of("klaus"); + static final TestUser PETER = TestUser.of("peter"); + static final TestUser KLAUS = TestUser.of("klaus"); - static final KeyValue[] INPUT_MESSAGES = new KeyValue[] + static final Stream> getInputMessages() + { + return Stream.of(INPUT_MESSAGES); + } + + private static final KeyValue[] INPUT_MESSAGES = new KeyValue[] { new KeyValue<>( TestWord.of(PETER.getUser(),"Hallo"), @@ -59,7 +63,7 @@ class TestData TestCounter.of(KLAUS.getUser(),"s",3)), }; - static void assertExpectedMessages(MultiValueMap receivedMessages) + static void assertExpectedMessages(MultiValueMap receivedMessages) { expectedMessages().forEach( (user, rankings) -> @@ -69,101 +73,137 @@ class TestData static void assertExpectedState(ReadOnlyKeyValueStore store) { - assertThat(store.get(EXPECTED_MESSAGES[9].key)).isEqualTo(EXPECTED_MESSAGES[9].value); - assertThat(store.get(EXPECTED_MESSAGES[10].key)).isEqualTo(EXPECTED_MESSAGES[10].value); + assertRankingEqualsRankingFromLastMessage(PETER, store.get(userOf(PETER))); + assertRankingEqualsRankingFromLastMessage(KLAUS, store.get(userOf(KLAUS))); + } + + private static User userOf(TestUser user) + { + return User.of(user.getUser()); + } + + static void assertExpectedNumberOfMessagesForUsers(MultiValueMap receivedMessages) + { + assertThat(countMessagesForUser(PETER, receivedMessages)); + assertThat(countMessagesForUser(KLAUS, receivedMessages)); + } + + private static int countMessagesForUser(TestUser user, MultiValueMap messagesForUsers) + { + return messagesForUsers.get(user) == null + ? 0 + : messagesForUsers.get(user).size(); + } + + + static void assertExpectedLastMessagesForUsers(MultiValueMap receivedMessages) + { + assertRankingEqualsRankingFromLastMessage(PETER, getLastMessageFor(PETER, receivedMessages)); + assertRankingEqualsRankingFromLastMessage(KLAUS, getLastMessageFor(KLAUS, receivedMessages)); } - static KeyValue[] EXPECTED_MESSAGES = new KeyValue[] + private static void assertRankingEqualsRankingFromLastMessage(TestUser user, Ranking ranking) + { + TestRanking testRanking = TestRanking.of(testEntriesOf(ranking.getEntries())); + assertRankingEqualsRankingFromLastMessage(user, testRanking); + } + + private static TestEntry[] testEntriesOf(Entry... entries) + { + return Arrays + .stream(entries) + .map(entry -> TestEntry.of( + entry.getWord(), + entry.getCounter() == null + ? -1l + : entry.getCounter())) + .toArray(size -> new TestEntry[size]); + } + + private static void assertRankingEqualsRankingFromLastMessage(TestUser user, TestRanking ranking) + { + assertThat(ranking).isEqualTo(getLastMessageFor(user)); + } + + private static TestRanking getLastMessageFor(TestUser user) + { + return getLastMessageFor(user, expectedMessages()); + } + + private static TestRanking getLastMessageFor(TestUser user, MultiValueMap messagesForUsers) + { + return messagesForUsers + .get(user) + .stream() + .reduce(null, (left, right) -> right); + } + + private static KeyValue[] EXPECTED_MESSAGES = new KeyValue[] { KeyValue.pair( // 0 PETER, - Ranking.of( - Entry.of("Hallo", 1l))), + TestRanking.of( + TestEntry.of("Hallo", 1l))), KeyValue.pair( // 1 KLAUS, - Ranking.of( - Entry.of("Müsch", 1l))), + TestRanking.of( + TestEntry.of("Müsch", 1l))), KeyValue.pair( // 2 PETER, - Ranking.of( - Entry.of("Hallo", 1l), - Entry.of("Welt", 1l))), + TestRanking.of( + TestEntry.of("Hallo", 1l), + TestEntry.of("Welt", 1l))), KeyValue.pair( // 3 KLAUS, - Ranking.of( - Entry.of("Müsch", 2l))), + TestRanking.of( + TestEntry.of("Müsch", 2l))), KeyValue.pair( // 4 KLAUS, - Ranking.of( - Entry.of("Müsch", 2l), - Entry.of("s", 1l))), + TestRanking.of( + TestEntry.of("Müsch", 2l), + TestEntry.of("s", 1l))), KeyValue.pair( // 5 PETER, - Ranking.of( - Entry.of("Hallo", 1l), - Entry.of("Welt", 1l), - Entry.of("Boäh", 1l))), + TestRanking.of( + TestEntry.of("Hallo", 1l), + TestEntry.of("Welt", 1l), + TestEntry.of("Boäh", 1l))), KeyValue.pair( // 6 PETER, - Ranking.of( - Entry.of("Welt", 2l), - Entry.of("Hallo", 1l), - Entry.of("Boäh", 1l))), + TestRanking.of( + TestEntry.of("Welt", 2l), + TestEntry.of("Hallo", 1l), + TestEntry.of("Boäh", 1l))), KeyValue.pair( // 7 PETER, - Ranking.of( - Entry.of("Welt", 2l), - Entry.of("Boäh", 2l), - Entry.of("Hallo", 1l))), + TestRanking.of( + TestEntry.of("Welt", 2l), + TestEntry.of("Boäh", 2l), + TestEntry.of("Hallo", 1l))), KeyValue.pair( // 8 KLAUS, - Ranking.of( - Entry.of("Müsch", 2l), - Entry.of("s", 2l))), + TestRanking.of( + TestEntry.of("Müsch", 2l), + TestEntry.of("s", 2l))), KeyValue.pair( // 9 PETER, - Ranking.of( - Entry.of("Boäh", 3l), - Entry.of("Welt", 2l), - Entry.of("Hallo", 1l))), + TestRanking.of( + TestEntry.of("Boäh", 3l), + TestEntry.of("Welt", 2l), + TestEntry.of("Hallo", 1l))), KeyValue.pair( // 10 KLAUS, - Ranking.of( - Entry.of("s", 3l), - Entry.of("Müsch", 2l))), + TestRanking.of( + TestEntry.of("s", 3l), + TestEntry.of("Müsch", 2l))), }; - static MultiValueMap expectedMessages() + private static MultiValueMap expectedMessages() { - MultiValueMap expectedMessages = new LinkedMultiValueMap<>(); + MultiValueMap expectedMessages = new LinkedMultiValueMap<>(); Stream .of(EXPECTED_MESSAGES) .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value)); return expectedMessages; } - - static Map convertToMap(Properties properties) - { - return properties - .entrySet() - .stream() - .collect( - Collectors.toMap( - entry -> (String)entry.getKey(), - entry -> entry.getValue() - )); - } - - static String parseHeader(Headers headers, String key) - { - Header header = headers.lastHeader(key); - if (header == null) - { - return key + "=null"; - } - else - { - return key + "=" + new String(header.value()); - } - } }