test: Überprüfung der Fachlogik ergänzt

* Überall, wo die Fachlogik geprüft wird, wird jetzt sichergestellt,
  dass die berechneten Ergebnisse den Erwartungen entsprechen.
* Überprüft, werden nur die zu dem Zeitpunkt tatsächlich vollständig
  berechneten Ergebnisse. Dadurch wird durch die Überprüfung kein Fehler
  ausgelöst, wenn wegen einem simulierten Fehler noch nicht alle durch
  die erzeugten Nachrichten angeforderten Berechnungen erfolgt sind.

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 1a336d6..1336050 100644 (file)
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -3,21 +3,33 @@ package de.juplo.kafka;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.Bytes;
+import org.springframework.beans.factory.annotation.Autowired;
 
+import java.util.*;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 
 public class ApplicationTests extends GenericApplicationTests<String, String>
 {
+  @Autowired
+  AdderResults results;
+
+
   public ApplicationTests()
   {
     super(new ApplicationTestRecrodGenerator());
+    ((ApplicationTestRecrodGenerator)recordGenerator).tests = this;
   }
 
 
   static class ApplicationTestRecrodGenerator implements RecordGenerator
   {
+    ApplicationTests tests;
+
     final int[] numbers = {1, 7, 3, 2, 33, 6, 11};
     final String[] dieWilden13 =
         IntStream
@@ -29,6 +41,8 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
 
     int counter = 0;
 
+    Map<String, List<AdderResult>> state;
+
     @Override
     public int generate(
         boolean poisonPills,
@@ -36,6 +50,12 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
         Consumer<ProducerRecord<Bytes, Bytes>> messageSender)
     {
       counter = 0;
+      state =
+          Arrays
+              .stream(dieWilden13)
+              .collect(Collectors.toMap(
+                  seeräuber -> seeräuber,
+                  seeräuber -> new LinkedList()));
 
       for (int i = 0; i < 33; i++)
       {
@@ -50,6 +70,8 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
           send(key, value, logicErrors, messageSender);
         }
         send(key, calculateMessage, logicErrors, messageSender);
+
+        state.get(seeräuber).add(new AdderResult(number, (number + 1) * number / 2));
       }
 
       return counter;
@@ -79,5 +101,24 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
     {
       return false;
     }
+
+    @Override
+    public void assertBusinessLogic()
+    {
+      tests.results
+          .getState()
+          .values()
+          .stream()
+          .flatMap(map -> map.entrySet().stream())
+          .forEach(entry ->
+          {
+            String user = entry.getKey();
+            List<AdderResult> resultsForUser = entry.getValue();
+
+            assertThat(state.get(user))
+                .describedAs("Unexpected results for user {}", user)
+                .containsExactlyElementsOf(resultsForUser);
+          });
+    }
   }
 }