counter: 1.1.11 - Added a test, that is based on `TopologyTestDriver`

[demos/kafka/wordcount] / src / test / java / de / juplo / kafka / wordcount / counter / CounterStreamProcessorTopologyTest.java

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
new file mode 100644 (file)
index 0000000..c2ada6f
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
@@ -0,0 +1,54 @@
+package de.juplo.kafka.wordcount.counter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.serialization.*;
+import org.apache.kafka.streams.*;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Properties;
+
+
+public class CounterStreamProcessorTopologyTest
+{
+  public final static String IN = "TEST-IN";
+  public final static String OUT = "TEST-OUT";
+
+  @Test
+  public void test()
+  {
+    Topology topology = CounterStreamProcessor.buildTopology(
+        IN,
+        OUT,
+        Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"),
+        new ObjectMapper());
+
+    CounterApplicationConfiguriation config =
+        new CounterApplicationConfiguriation();
+    Properties properties =
+        config.propertyMap(new CounterApplicationProperties());
+
+    TopologyTestDriver testDriver = new TopologyTestDriver(topology, properties);
+
+    TestInputTopic<String, String> in = testDriver.createInputTopic(
+        IN,
+        new StringSerializer(),
+        new StringSerializer());
+
+    TestOutputTopic<String, String> out = testDriver.createOutputTopic(
+        OUT,
+        new StringDeserializer(),
+        new StringDeserializer());
+
+    TestData.writeInputData((key, value) -> in.pipeInput(key, value));
+
+    List<Message> receivedMessages = out
+        .readRecordsToList()
+        .stream()
+        .map(record -> Message.of(record.key(), record.value()))
+        .toList();
+
+    TestData.assertExpectedResult(receivedMessages);
+  }
+}