chore: add await to in-memory messaging tests

hantsy · Apr 27, 2024 · 9411a96 · 9411a96
1 parent 7348c73
commit 9411a96
Show file tree

Hide file tree

Showing 12 changed files with 96 additions and 29 deletions.
diff --git a/amqp-rabbitmq/pom.xml b/amqp-rabbitmq/pom.xml
@@ -20,10 +20,11 @@
 
         <!-- dependencies -->
         <quarkus.version>3.10.0.CR1</quarkus.version>
-
-        <skipITs>true</skipITs>
         <lombok.version>1.18.32</lombok.version>
         <assertj.version>3.25.3</assertj.version>
+        <awaitility.version>4.2.0</awaitility.version>
+
+        <skipITs>true</skipITs>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -96,6 +97,12 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

diff --git a/amqp-rabbitmq/src/test/java/com/example/InMemoryProfile.java b/amqp-rabbitmq/src/test/java/com/example/InMemoryProfile.java
@@ -11,6 +11,7 @@ public InMemoryProfile() {
     @Override
     public Map<String, String> getConfigOverrides() {
         return Map.of(
+                "quarkus.amqp.devservices.enabled", "false",
                 "mp.messaging.outgoing.send.connector","smallrye-in-memory",
                 "mp.messaging.incoming.messages.connector","smallrye-in-memory",
                 "mp.messaging.outgoing.data-stream.connector","smallrye-in-memory"

diff --git a/amqp-rabbitmq/src/test/java/com/example/MessageHandlerTest.java b/amqp-rabbitmq/src/test/java/com/example/MessageHandlerTest.java
@@ -10,7 +10,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @QuarkusTest
 @TestProfile(InMemoryProfile.class)
@@ -33,11 +36,14 @@ void receive() {
         InMemorySink<String> sink = connector.sink("send");
         InMemorySink<Message> dataStream = connector.sink("data-stream");
 
-
         handler.send("hello");
-        assertThat(sink.received().get(0).getPayload()).isEqualTo("hello");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello")
+        );
 
         messages.send("hello-123");
-        assertThat(dataStream.received().get(0).getPayload().body()).isEqualTo("hello-123");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123")
+        );
     }
 }
diff --git a/amqp/pom.xml b/amqp/pom.xml
@@ -20,10 +20,11 @@
 
         <!-- dependencies -->
         <quarkus.version>3.10.0.CR1</quarkus.version>
-
-        <skipITs>true</skipITs>
         <lombok.version>1.18.32</lombok.version>
         <assertj.version>3.25.3</assertj.version>
+        <awaitility.version>4.2.0</awaitility.version>
+
+        <skipITs>true</skipITs>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -90,6 +91,12 @@
             <artifactId>smallrye-reactive-messaging-in-memory</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

diff --git a/amqp/src/test/java/com/example/InMemoryProfile.java b/amqp/src/test/java/com/example/InMemoryProfile.java
@@ -11,9 +11,10 @@ public InMemoryProfile() {
     @Override
     public Map<String, String> getConfigOverrides() {
         return Map.of(
-                "mp.messaging.outgoing.send.connector","smallrye-in-memory",
-                "mp.messaging.incoming.messages.connector","smallrye-in-memory",
-                "mp.messaging.outgoing.data-stream.connector","smallrye-in-memory"
+                "quarkus.amqp.devservices.enabled", "false",
+                "mp.messaging.outgoing.send.connector", "smallrye-in-memory",
+                "mp.messaging.incoming.messages.connector", "smallrye-in-memory",
+                "mp.messaging.outgoing.data-stream.connector", "smallrye-in-memory"
         );
     }
 

diff --git a/amqp/src/test/java/com/example/MessageHandlerTest.java b/amqp/src/test/java/com/example/MessageHandlerTest.java
@@ -10,7 +10,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @QuarkusTest
 @TestProfile(InMemoryProfile.class)
@@ -32,12 +35,15 @@ void receive() {
         InMemorySource<String> messages = connector.source("messages");
         InMemorySink<String> sink = connector.sink("send");
         InMemorySink<Message> dataStream = connector.sink("data-stream");
-
-
+
         handler.send("hello");
-        assertThat(sink.received().get(0).getPayload()).isEqualTo("hello");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(sink.received().get(0).getPayload()).isEqualTo("hello")
+        );
 
         messages.send("hello-123");
-        assertThat(dataStream.received().get(0).getPayload().body()).isEqualTo("hello-123");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(dataStream.received().get(0).getPayload().body()).isEqualTo("hello-123")
+        );
     }
 }
diff --git a/kafka/pom.xml b/kafka/pom.xml
@@ -20,10 +20,11 @@
 
         <!-- dependencies -->
         <quarkus.version>3.10.0.CR1</quarkus.version>
-
-        <skipITs>true</skipITs>
         <lombok.version>1.18.32</lombok.version>
         <assertj.version>3.25.3</assertj.version>
+        <awaitility.version>4.2.0</awaitility.version>
+
+        <skipITs>true</skipITs>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -86,6 +87,12 @@
             <artifactId>smallrye-reactive-messaging-in-memory</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

diff --git a/kafka/src/test/java/com/example/MessageHandlerTest.java b/kafka/src/test/java/com/example/MessageHandlerTest.java
@@ -10,7 +10,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @QuarkusTest
 @TestProfile(InMemoryProfile.class)
@@ -33,11 +36,14 @@ void receive() {
         InMemorySink<String> sink = connector.sink("send");
         InMemorySink<Message> dataStream = connector.sink("data-stream");
 
-
         handler.send("hello");
-        assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello")
+        );
 
         messages.send("hello-123");
-        assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123")
+        );
     }
 }
diff --git a/pulsar/pom.xml b/pulsar/pom.xml
@@ -20,10 +20,11 @@
 
         <!-- dependencies -->
         <quarkus.version>3.10.0.CR1</quarkus.version>
-
-        <skipITs>true</skipITs>
         <lombok.version>1.18.32</lombok.version>
         <assertj.version>3.25.3</assertj.version>
+        <awaitility.version>4.2.0</awaitility.version>
+
+        <skipITs>true</skipITs>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -86,6 +87,12 @@
             <artifactId>smallrye-reactive-messaging-in-memory</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

diff --git a/pulsar/src/test/java/com/example/MessageHandlerTest.java b/pulsar/src/test/java/com/example/MessageHandlerTest.java
@@ -12,7 +12,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @QuarkusTest
 @TestProfile(InMemoryProfile.class)
@@ -44,11 +47,14 @@ void receive() {
         InMemorySink<String> sink = connector.sink("send");
         InMemorySink<Message> dataStream = connector.sink("data-stream");
 
-
         handler.send("hello");
-        assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello")
+        );
 
         messages.send("hello-123");
-        assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123")
+        );
     }
 }
diff --git a/rabbitmq/pom.xml b/rabbitmq/pom.xml
@@ -20,10 +20,11 @@
 
         <!-- dependencies -->
         <quarkus.version>3.10.0.CR1</quarkus.version>
-
-        <skipITs>true</skipITs>
         <lombok.version>1.18.32</lombok.version>
         <assertj.version>3.25.3</assertj.version>
+        <awaitility.version>4.2.0</awaitility.version>
+
+        <skipITs>true</skipITs>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -87,6 +88,12 @@
             <artifactId>smallrye-reactive-messaging-in-memory</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

diff --git a/rabbitmq/src/test/java/com/example/MessageHandlerTest.java b/rabbitmq/src/test/java/com/example/MessageHandlerTest.java
@@ -10,7 +10,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @QuarkusTest
 @TestProfile(InMemoryProfile.class)
@@ -33,11 +36,14 @@ void receive() {
         InMemorySink<String> sink = connector.sink("send");
         InMemorySink<Message> dataStream = connector.sink("data-stream");
 
-
         handler.send("hello");
-        assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(sink.received().getFirst().getPayload()).isEqualTo("hello")
+        );
 
         messages.send("hello-123");
-        assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123");
+        await().atMost(Duration.ofMillis(1000)).untilAsserted(() ->
+                assertThat(dataStream.received().getFirst().getPayload().body()).isEqualTo("hello-123")
+        );
     }
 }