X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterStreamProcessorTopologyTest.java;h=e80e383108569f4f1fb35babda6b152b854b1826;hb=refs%2Fheads%2Fcounter;hp=06a0798198493851127b02bd88b1bec49632c2ca;hpb=29899898af28dcbb50326ae5837ada195829c592;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java index 06a0798..e80e383 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java @@ -11,8 +11,9 @@ import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.Stores; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; @@ -20,7 +21,6 @@ import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import java.util.Map; -import java.util.stream.Collectors; import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig; import static de.juplo.kafka.wordcount.counter.CounterStreamProcessor.STORE_NAME; @@ -33,13 +33,12 @@ public class CounterStreamProcessorTopologyTest public static final String OUT = "TEST-OUT"; - TopologyTestDriver testDriver; - TestInputTopic in; - TestOutputTopic out; + static TopologyTestDriver testDriver; + static MultiValueMap receivedMessages = new LinkedMultiValueMap<>(); - @BeforeEach - public void setUpTestDriver() + @BeforeAll + public static void setUpTestDriver() { Topology topology = CounterStreamProcessor.buildTopology( IN, @@ -48,34 +47,53 @@ public class CounterStreamProcessorTopologyTest testDriver = new TopologyTestDriver(topology, serializationConfig()); - in = testDriver.createInputTopic(IN, serializer(), serializer()); - out = testDriver.createOutputTopic(OUT, keyDeserializer(), valueDeserializer()); - } - + TestInputTopic in = + testDriver.createInputTopic(IN, serializer(), serializer()); + TestOutputTopic out = + testDriver.createOutputTopic(OUT, keyDeserializer(), valueDeserializer()); - @Test - public void test() - { TestData .getInputMessages() .forEach(kv -> in.pipeInput(kv.key, kv.value)); - MultiValueMap receivedMessages = new LinkedMultiValueMap<>(); + receivedMessages = new LinkedMultiValueMap<>(); out .readRecordsToList() .forEach(record -> receivedMessages.add(record.key(), record.value())); + } + + @DisplayName("Assert the expected output messages") + @Test + public void testExpectedMessages() + { TestData.assertExpectedMessages(receivedMessages); + } + @DisplayName("Assert the expected number of messages") + @Test + public void testExpectedNumberOfMessagesForWord() + { TestData.assertExpectedNumberOfMessagesForWord(receivedMessages); + } + + @DisplayName("Await the expected final output messages") + @Test + public void testExpectedLastMessagesForWord() + { TestData.assertExpectedLastMessagesForWord(receivedMessages); + } + @DisplayName("Assert the expected state in the state-store") + @Test + public void testExpectedState() + { KeyValueStore store = testDriver.getKeyValueStore(STORE_NAME); TestData.assertExpectedState(store); } - @AfterEach - public void tearDown() + @AfterAll + public static void tearDown() { testDriver.close(); } @@ -86,7 +104,7 @@ public class CounterStreamProcessorTopologyTest return new JsonSerializer().noTypeInfo(); } - private JsonDeserializer keyDeserializer() + private static JsonDeserializer keyDeserializer() { return deserializer(true); } @@ -107,17 +125,6 @@ public class CounterStreamProcessorTopologyTest private static String typeMappingsConfig() { - return typeMappings() - .entrySet() - .stream() - .map(entry -> entry.getKey() + ":" + entry.getValue().getName()) - .collect(Collectors.joining(",")); - } - - private static Map typeMappings() - { - return Map.of( - "word", TestOutputWord.class, - "counter", TestOutputWordCounter.class); + return CounterStreamProcessor.typeMappingsConfig(TestOutputWord.class, TestOutputWordCounter.class); } }