Use AssertJ conditions in Java stream test suite

4 files changed, 45 insertions, 18 deletions

diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/ClusterSizeTest.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/ClusterSizeTest.java
index 8903ee0c62..6ecf2b4ae4 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/ClusterSizeTest.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/ClusterSizeTest.java
@@ -16,6 +16,9 @@
 
 package com.rabbitmq.stream;
 
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.ko;
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.ok;
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.responseCode;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.rabbitmq.stream.impl.Client;
@@ -40,8 +43,7 @@ public class ClusterSizeTest {
     String s = UUID.randomUUID().toString();
     Response response =
         client.create(s, Collections.singletonMap("initial-cluster-size", clusterSize));
-    assertThat(response.isOk()).isFalse();
-    assertThat(response.getResponseCode()).isEqualTo(Constants.RESPONSE_CODE_PRECONDITION_FAILED);
+    assertThat(response).is(ko()).has(responseCode(Constants.RESPONSE_CODE_PRECONDITION_FAILED));
   }
 
   @ParameterizedTest
@@ -53,7 +55,7 @@ public class ClusterSizeTest {
     try {
       Response response =
           client.create(s, Collections.singletonMap("initial-cluster-size", requestedClusterSize));
-      assertThat(response.isOk()).isTrue();
+      assertThat(response).is(ok());
       StreamMetadata metadata = client.metadata(s).get(s);
       assertThat(metadata).isNotNull();
       assertThat(metadata.getResponseCode()).isEqualTo(Constants.RESPONSE_CODE_OK);
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java
index 29918c1ce7..bc804b9e47 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java
@@ -16,6 +16,8 @@
 
 package com.rabbitmq.stream;
 
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.ok;
+import static com.rabbitmq.stream.TestUtils.waitAtMost;
 import static com.rabbitmq.stream.TestUtils.waitUntil;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
@@ -110,7 +112,7 @@ public class FailureTest {
       assertThat(metadataLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
       // wait until there's a new leader
-      TestUtils.waitAtMost(
+      waitAtMost(
           Duration.ofSeconds(10),
           () -> {
             Client.StreamMetadata m = publisher.metadata(stream).get(stream);
@@ -136,7 +138,7 @@ public class FailureTest {
     }
 
     // wait until all the replicas are there
-    TestUtils.waitAtMost(
+    waitAtMost(
         Duration.ofSeconds(10),
         () -> {
           LOGGER.info("Getting metadata for {}", stream);
@@ -167,7 +169,7 @@ public class FailureTest {
                       consumeLatch.countDown();
                     }));
 
-    TestUtils.waitAtMost(
+    waitAtMost(
         Duration.ofSeconds(5),
         () -> {
           Client.Response response =
@@ -222,7 +224,7 @@ public class FailureTest {
                       cf.get(new Client.ClientParameters().port(TestUtils.streamPortNode2()));
                   // wait until there's a new leader
                   try {
-                    TestUtils.waitAtMost(
+                    waitAtMost(
                         Duration.ofSeconds(5),
                         () -> {
                           Client.StreamMetadata m = locator.metadata(stream).get(stream);
@@ -317,7 +319,7 @@ public class FailureTest {
 
     Client metadataClient = cf.get(new Client.ClientParameters().port(TestUtils.streamPortNode2()));
     // wait until all the replicas are there
-    TestUtils.waitAtMost(
+    waitAtMost(
         Duration.ofSeconds(5),
         () -> {
           Client.StreamMetadata m = metadataClient.metadata(stream).get(stream);
@@ -353,7 +355,7 @@ public class FailureTest {
 
     Client.Response response =
         consumer.subscribe((byte) 1, stream, OffsetSpecification.first(), 10);
-    assertThat(response.isOk()).isTrue();
+    assertThat(response).is(ok());
 
     assertThat(consumedLatch.await(5, TimeUnit.SECONDS)).isTrue();
     assertThat(generations).hasSize(2).contains(0L, 1L);
@@ -499,7 +501,7 @@ public class FailureTest {
 
     Client.Response response =
         consumer.subscribe((byte) 1, stream, OffsetSpecification.first(), 10);
-    assertThat(response.isOk()).isTrue();
+    assertThat(response).is(ok());
 
     // let's publish for a bit of time
     Thread.sleep(2000);
@@ -523,7 +525,7 @@ public class FailureTest {
     confirmedCount = confirmed.size();
 
     // wait until all the replicas are there
-    TestUtils.waitAtMost(
+    waitAtMost(
         Duration.ofSeconds(10),
         () -> {
           Client.StreamMetadata m = metadataClient.metadata(stream).get(stream);
@@ -539,7 +541,7 @@ public class FailureTest {
 
     assertThat(publishingLatch.await(5, TimeUnit.SECONDS)).isTrue();
 
-    TestUtils.waitAtMost(Duration.ofSeconds(5), () -> consumed.size() >= confirmed.size());
+    waitAtMost(Duration.ofSeconds(5), () -> consumed.size() >= confirmed.size());
 
     assertThat(generations).hasSize(2).contains(0L, 1L);
     assertThat(consumed).hasSizeGreaterThanOrEqualTo(confirmed.size());
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/LeaderLocatorTest.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/LeaderLocatorTest.java
index 104f5772c1..7d387c09c2 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/LeaderLocatorTest.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/LeaderLocatorTest.java
@@ -16,6 +16,9 @@
 
 package com.rabbitmq.stream;
 
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.ko;
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.ok;
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.responseCode;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -47,8 +50,7 @@ public class LeaderLocatorTest {
     Client client = cf.get(new Client.ClientParameters().port(TestUtils.streamPortNode1()));
     String s = UUID.randomUUID().toString();
     Response response = client.create(s, Collections.singletonMap("queue-leader-locator", "foo"));
-    assertThat(response.isOk()).isFalse();
-    assertThat(response.getResponseCode()).isEqualTo(Constants.RESPONSE_CODE_PRECONDITION_FAILED);
+    assertThat(response).is(ko()).has(responseCode(Constants.RESPONSE_CODE_PRECONDITION_FAILED));
   }
 
   @Test
@@ -60,7 +62,7 @@ public class LeaderLocatorTest {
       try {
         Response response =
             client.create(s, Collections.singletonMap("queue-leader-locator", "client-local"));
-        assertThat(response.isOk()).isTrue();
+        assertThat(response).is(ok());
         StreamMetadata metadata = client.metadata(s).get(s);
         assertThat(metadata).isNotNull();
         assertThat(metadata.getResponseCode()).isEqualTo(Constants.RESPONSE_CODE_OK);
@@ -136,7 +138,7 @@ public class LeaderLocatorTest {
                 Response response =
                     client.create(
                         s, Collections.singletonMap("queue-leader-locator", "least-leaders"));
-                assertThat(response.isOk()).isTrue();
+                assertThat(response).is(ok());
                 createdStreams.add(s);
               });
 
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java
index e0a5afee67..03015a0c44 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java
@@ -16,11 +16,13 @@
 
 package com.rabbitmq.stream;
 
+import static com.rabbitmq.stream.TestUtils.ResponseConditions.ok;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import com.rabbitmq.stream.impl.Client;
+import com.rabbitmq.stream.impl.Client.Response;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import java.lang.reflect.Field;
@@ -30,6 +32,7 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BooleanSupplier;
+import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.*;
 
@@ -106,7 +109,7 @@ public class TestUtils {
                     .eventLoopGroup(eventLoopGroup(context))
                     .port(streamPortNode1()));
         Client.Response response = client.create(stream);
-        assertThat(response.isOk()).isTrue();
+        assertThat(response).is(ok());
         client.close();
         store(context).put("testMethodStream", stream);
       } catch (NoSuchFieldException e) {
@@ -136,7 +139,7 @@ public class TestUtils {
                     .eventLoopGroup(eventLoopGroup(context))
                     .port(streamPortNode1()));
         Client.Response response = client.delete(stream);
-        assertThat(response.isOk()).isTrue();
+        assertThat(response).is(ok());
         client.close();
         store(context).remove("testMethodStream");
       } catch (NoSuchFieldException e) {
@@ -197,4 +200,22 @@ public class TestUtils {
       }
     }
   }
+
+  static class ResponseConditions {
+
+    static Condition<Response> ok() {
+      return new Condition<>(Response::isOk, "Response should be OK");
+    }
+
+    static Condition<Response> ko() {
+      return new Condition<>(response -> !response.isOk(), "Response should be OK");
+    }
+
+    static Condition<Response> responseCode(short expectedResponse) {
+      return new Condition<>(
+          response -> response.getResponseCode() == expectedResponse,
+          "response code %d",
+          expectedResponse);
+    }
+  }
 }