From: Kai Moritz <kai@juplo.de>
Date: Sat, 25 May 2024 14:59:42 +0000 (+0200)
Subject: top10: 1.1.2 - (RED) Added test, that asserts the expectated processing
X-Git-Tag: top10-1.1.2~3
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=21b1d0cffa0bfe47d7d78cef4e4dc23588d69a1b;p=demos%2Fkafka%2Fwordcount
top10: 1.1.2 - (RED) Added test, that asserts the expectated processing
---
diff --git a/pom.xml b/pom.xml
index 6749738..86b4290 100644
--- 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
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
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
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
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();
+ }
+}