top10: 1.1.2 - (RED) Added test, that asserts the expectated processing

diff --git a/pom.xml b/pom.xml
index 6749738..86b4290 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -10,7 +10,7 @@
        </parent>
        <groupId>de.juplo.kafka.wordcount</groupId>
        <artifactId>top10</artifactId>
-       <version>1.1.1</version>
+       <version>1.1.2</version>
        <name>Wordcount-Top-10</name>
        <description>Top-10 stream-processor of the multi-user wordcount-example</description>
        <properties>
@@ -50,15 +50,34 @@
                        <artifactId>lombok</artifactId>
                        <optional>true</optional>
                </dependency>
+
                <dependency>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-starter-test</artifactId>
                        <scope>test</scope>
                </dependency>
+               <dependency>
+                       <groupId>org.springframework.kafka</groupId>
+                       <artifactId>spring-kafka-test</artifactId>
+                       <scope>test</scope>
+               </dependency>
+               <dependency>
+                       <groupId>org.awaitility</groupId>
+                       <artifactId>awaitility</artifactId>
+                       <scope>test</scope>
+               </dependency>
+               <dependency>
+                       <groupId>org.assertj</groupId>
+                       <artifactId>assertj-core</artifactId>
+                       <scope>test</scope>
+               </dependency>
        </dependencies>
 
        <build>
                <plugins>
+                       <plugin>
+                               <artifactId>maven-failsafe-plugin</artifactId>
+                       </plugin>
                        <plugin>
                                <groupId>org.springframework.boot</groupId>
                                <artifactId>spring-boot-maven-plugin</artifactId>

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java
new file mode 100644 (file)
index 0000000..d98ae64
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java
@@ -0,0 +1,21 @@
+package de.juplo.kafka.wordcount.counter;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor(staticName = "of")
+public class TestCounter
+{
+  String user;
+  String word;
+  long counter;
+
+  public static TestCounter of(TestWord word, long counter)
+  {
+    return new TestCounter(word.getUser(), word.getWord(), counter);
+  }
+}

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java
new file mode 100644 (file)
index 0000000..8008e12
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java
@@ -0,0 +1,17 @@
+package de.juplo.kafka.wordcount.counter;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+@Data
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class TestWord
+{
+  private String user;
+  private String word;
+}

diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java
new file mode 100644 (file)
index 0000000..73a405e
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java
@@ -0,0 +1,159 @@
+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 org.apache.kafka.streams.KeyValue;
+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.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+class TestData
+{
+       static final KeyValue<TestWord, TestCounter>[] INPUT_MESSAGES = new KeyValue[]
+       {
+                       new KeyValue<>(
+                                       TestWord.of("peter","Hallo"),
+                                       TestCounter.of("peter","Hallo",1)),
+                       new KeyValue<>(
+                                       TestWord.of("klaus","Müsch"),
+                                       TestCounter.of("klaus","Müsch",1)),
+                       new KeyValue<>(
+                                       TestWord.of("peter","Welt"),
+                                       TestCounter.of("peter","Welt",1)),
+                       new KeyValue<>(
+                                       TestWord.of("klaus","Müsch"),
+                                       TestCounter.of("klaus","Müsch",2)),
+                       new KeyValue<>(
+                                       TestWord.of("klaus","s"),
+                                       TestCounter.of("klaus","s",1)),
+                       new KeyValue<>(
+                                       TestWord.of("peter","Boäh"),
+                                       TestCounter.of("peter","Boäh",1)),
+                       new KeyValue<>(
+                                       TestWord.of("peter","Welt"),
+                                       TestCounter.of("peter","Welt",2)),
+                       new KeyValue<>(
+                                       TestWord.of("peter","Boäh"),
+                                       TestCounter.of("peter","Boäh",2)),
+                       new KeyValue<>(
+                                       TestWord.of("klaus","s"),
+                                       TestCounter.of("klaus","s",2)),
+                       new KeyValue<>(
+                                       TestWord.of("peter","Boäh"),
+                                       TestCounter.of("peter","Boäh",3)),
+                       new KeyValue<>(
+                                       TestWord.of("klaus","s"),
+                                       TestCounter.of("klaus","s",3)),
+       };
+
+       static void assertExpectedMessages(MultiValueMap<String, Ranking> receivedMessages)
+       {
+               expectedMessages().forEach(
+                               (user, rankings) ->
+                                               assertThat(receivedMessages.get(user))
+                                                               .containsExactlyElementsOf(rankings));
+       }
+
+       static KeyValue<String, Ranking>[] EXPECTED_MESSAGES = new KeyValue[]
+       {
+                       KeyValue.pair( // 0
+                                       "peter",
+                                       Ranking.of(
+                                                       Entry.of("Hallo", 1l))),
+                       KeyValue.pair( // 1
+                                       "klaus",
+                                       Ranking.of(
+                                                       Entry.of("Müsch", 1l))),
+                       KeyValue.pair( // 2
+                                       "peter",
+                                       Ranking.of(
+                                                       Entry.of("Hallo", 1l),
+                                                       Entry.of("Welt", 1l))),
+                       KeyValue.pair( // 3
+                                       "klaus",
+                                       Ranking.of(
+                                                       Entry.of("Müsch", 2l))),
+                       KeyValue.pair( // 4
+                                       "klaus",
+                                       Ranking.of(
+                                                       Entry.of("Müsch", 2l),
+                                                       Entry.of("s", 1l))),
+                       KeyValue.pair( // 5
+                                       "peter",
+                                       Ranking.of(
+                                                       Entry.of("Hallo", 1l),
+                                                       Entry.of("Welt", 1l),
+                                                       Entry.of("Boäh", 1l))),
+                       KeyValue.pair( // 6
+                                       "peter",
+                                       Ranking.of(
+                                                       Entry.of("Welt", 2l),
+                                                       Entry.of("Hallo", 1l),
+                                                       Entry.of("Boäh", 1l))),
+                       KeyValue.pair( // 7
+                                       "peter",
+                                       Ranking.of(
+                                                       Entry.of("Welt", 2l),
+                                                       Entry.of("Boäh", 2l),
+                                                       Entry.of("Hallo", 1l))),
+                       KeyValue.pair( // 8
+                                       "klaus",
+                                       Ranking.of(
+                                                       Entry.of("Müsch", 2l),
+                                                       Entry.of("s", 2l))),
+                       KeyValue.pair( // 9
+                                       "peter",
+                                       Ranking.of(
+                                                       Entry.of("Boäh", 3l),
+                                                       Entry.of("Welt", 2l),
+                                                       Entry.of("Hallo", 1l))),
+                       KeyValue.pair( // 10
+                                       "klaus",
+                                       Ranking.of(
+                                                       Entry.of("s", 3l),
+                                                       Entry.of("Müsch", 2l))),
+       };
+
+       static MultiValueMap<String, Ranking> expectedMessages()
+       {
+               MultiValueMap<String, Ranking> expectedMessages = new LinkedMultiValueMap<>();
+               Stream
+                               .of(EXPECTED_MESSAGES)
+                               .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value));
+               return expectedMessages;
+       }
+
+       static Map<String, Object> 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());
+               }
+       }
+}

diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java
new file mode 100644 (file)
index 0000000..8ecf9fa
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java
@@ -0,0 +1,83 @@
+package de.juplo.kafka.wordcount.top10;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.junit.jupiter.api.Test;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+import static de.juplo.kafka.wordcount.top10.TestData.convertToMap;
+import static de.juplo.kafka.wordcount.top10.TestData.parseHeader;
+import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
+import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
+
+
+@Slf4j
+public class Top10StreamProcessorTopologyTest
+{
+  public final static String IN = "TEST-IN";
+  public final static String OUT = "TEST-OUT";
+
+  @Test
+  public void test()
+  {
+    Topology topology = Top10StreamProcessor.buildTopology(IN, OUT);
+
+    Top10ApplicationConfiguration applicationConfiguriation =
+        new Top10ApplicationConfiguration();
+    Properties streamProcessorProperties =
+        applicationConfiguriation.streamProcessorProperties(new Top10ApplicationProperties());
+    Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
+
+    JsonSerde<?> keySerde = new JsonSerde<>();
+    keySerde.configure(propertyMap, true);
+    JsonSerde<?> valueSerde = new JsonSerde<>();
+    valueSerde.configure(propertyMap, false);
+
+    TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
+
+    TestInputTopic<Key, Entry> in = testDriver.createInputTopic(
+        IN,
+        (JsonSerializer<Key>)keySerde.serializer(),
+        (JsonSerializer<Entry>)valueSerde.serializer());
+
+    TestOutputTopic<String, Ranking> out = testDriver.createOutputTopic(
+        OUT,
+        (JsonDeserializer<String>)keySerde.deserializer(),
+        (JsonDeserializer<Ranking>)valueSerde.deserializer());
+
+    Stream
+        .of(TestData.INPUT_MESSAGES)
+        .forEach(kv -> in.pipeInput(
+            Key.of(kv.key.getUser(), kv.key.getWord()),
+            Entry.of(kv.value.getWord(), kv.value.getCounter())));
+
+    MultiValueMap<String, Ranking> receivedMessages = new LinkedMultiValueMap<>();
+    out
+        .readRecordsToList()
+        .forEach(record ->
+        {
+          log.debug(
+              "OUT: {} -> {}, {}, {}",
+              record.key(),
+              record.value(),
+              parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
+              parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
+          receivedMessages.add(record.key(), record.value());
+        });
+
+    TestData.assertExpectedMessages(receivedMessages);
+
+    testDriver.close();
+  }
+}