query: 2.0.0 - (RED) Formulated expectations for JSON-values

[demos/kafka/wordcount] / src / test / java / de / juplo / kafka / wordcount / query / QueryStreamProcessorTopologyTest.java

diff --git a/src/test/java/de/juplo/kafka/wordcount/query/QueryStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/query/QueryStreamProcessorTopologyTest.java
index 8439be1..eef1eec 100644 (file)
--- a/src/test/java/de/juplo/kafka/wordcount/query/QueryStreamProcessorTopologyTest.java
+++ b/src/test/java/de/juplo/kafka/wordcount/query/QueryStreamProcessorTopologyTest.java
@@ -15,6 +15,8 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.kafka.support.serializer.JsonSerializer;
 
+import java.util.Map;
+
 import static de.juplo.kafka.wordcount.query.QueryApplicationConfiguration.serializationConfig;
 
 
@@ -45,12 +47,12 @@ public class QueryStreamProcessorTopologyTest
     top10In = testDriver.createInputTopic(
         TOP10_IN,
         new StringSerializer(),
-        new JsonSerializer());
+        jsonSerializer(TestRanking.class));
 
     userIn = testDriver.createInputTopic(
         USERS_IN,
         new StringSerializer(),
-        new JsonSerializer());
+        jsonSerializer(TestUserData.class));
   }
 
 
@@ -64,8 +66,8 @@ public class QueryStreamProcessorTopologyTest
         .getTop10Messages()
         .forEach(kv -> top10In.pipeInput(kv.key, kv.value));
 
-    KeyValueStore<String, String> store = testDriver.getKeyValueStore(STORE_NAME);
-    TestData.assertExpectedState(store);
+    KeyValueStore<String, UserRanking> store = testDriver.getKeyValueStore(STORE_NAME);
+    TestData.assertExpectedState(user -> store.get(user));
   }
 
   @AfterEach
@@ -73,4 +75,16 @@ public class QueryStreamProcessorTopologyTest
   {
     testDriver.close();
   }
+
+  private <T> JsonSerializer<T> jsonSerializer(Class<T> type)
+  {
+    JsonSerializer<T> jsonSerializer = new JsonSerializer<>();
+    jsonSerializer.configure(
+        Map.of(
+            JsonSerializer.TYPE_MAPPINGS,
+            "userdata:" + TestUserData.class.getName() + "," +
+            "ranking:" + TestRanking.class.getName()),
+        false);
+    return jsonSerializer;
+  }
 }