From 1127988a1b631aa9e0c0107c1a3ed9f99edd188b Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sat, 8 Jun 2024 10:35:34 +0200
Subject: [PATCH] counter: 1.2.15 - Added assertion for the expected number of
 messages

---
 .../counter/CounterApplicationIT.java         |  9 +++++++++
 .../CounterStreamProcessorTopologyTest.java   |  2 ++
 .../kafka/wordcount/counter/TestData.java     | 19 +++++++++++++++++++
 3 files changed, 30 insertions(+)

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
index 41e5b62..a771904 100644
--- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
@@ -82,6 +82,15 @@ public class CounterApplicationIT
 				});
 	}
 
+	@DisplayName("Await the expected number of messages")
+	@Test
+	public void testAwaitExpectedNumberOfMessagesForUsers()
+	{
+		await("Expected number of messages")
+				.atMost(Duration.ofSeconds(5))
+				.untilAsserted(() -> consumer.enforceAssertion(TestData.expectedNumberOfMessagesForWordAssertion()));
+	}
+
 	@DisplayName("Await the expected output messages")
 	@Test
 	void testSendMessage()
diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
index a1b4c7f..e761679 100644
--- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
@@ -72,6 +72,8 @@ public class CounterStreamProcessorTopologyTest
         .forEach(record -> receivedMessages.add(record.key(), record.value()));
 
     TestData.assertExpectedMessages(receivedMessages);
+
+    TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
   }
 
   @AfterEach
diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
index 714e585..c9d871a 100644
--- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
@@ -85,6 +85,25 @@ class TestData
 								.containsExactlyElementsOf(counter));
 	}
 
+	static Consumer<MultiValueMap<TestOutputWord, TestOutputWordCounter>> expectedNumberOfMessagesForWordAssertion()
+	{
+		return receivedMessages -> assertExpectedNumberOfMessagesForWord(receivedMessages);
+	}
+
+	static void assertExpectedNumberOfMessagesForWord(MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages)
+	{
+		assertThat(countMessagesForWord(PETER_HALLO, receivedMessages));
+		assertThat(countMessagesForWord(PETER_WELT, receivedMessages));
+		assertThat(countMessagesForWord(PETER_BOÄH, receivedMessages));
+		assertThat(countMessagesForWord(KLAUS_MÜSCH, receivedMessages));
+		assertThat(countMessagesForWord(KLAUS_S, receivedMessages));
+	}
+
+	private static int countMessagesForWord(TestOutputWord word, MultiValueMap<TestOutputWord, TestOutputWordCounter> messagesForUsers)
+	{
+		return messagesForUsers.get(word).size();
+	}
+
 	private static final KeyValue<TestOutputWord, TestOutputWordCounter>[] EXPECTED_MESSAGES = new KeyValue[]
 	{
 			KeyValue.pair(
-- 
2.20.1