adjust batch delivery tests to new framework

allegro · Dec 1, 2023 · 4065a9f · 4065a9f
1 parent 191091a
commit 4065a9f
Show file tree

Hide file tree

Showing 9 changed files with 755 additions and 611 deletions.
diff --git a/...rc/common/java/pl/allegro/tech/hermes/consumers/EndpointAddressResolverConfiguration.java b/...rc/common/java/pl/allegro/tech/hermes/consumers/EndpointAddressResolverConfiguration.java
@@ -0,0 +1,19 @@
+package pl.allegro.tech.hermes.consumers;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Profile;
+import pl.allegro.tech.hermes.consumers.consumer.sender.resolver.EndpointAddressResolver;
+import pl.allegro.tech.hermes.test.helper.endpoint.MultiUrlEndpointAddressResolver;
+
+@Configuration
+public class EndpointAddressResolverConfiguration {
+
+    @Bean
+    @Primary
+    @Profile("integration")
+    public EndpointAddressResolver testMultiUrlEndpointAddressResolver() {
+        return new MultiUrlEndpointAddressResolver();
+    }
+}
diff --git a/...ts/src/common/java/pl/allegro/tech/hermes/integrationtests/subscriber/TestSubscriber.java b/...ts/src/common/java/pl/allegro/tech/hermes/integrationtests/subscriber/TestSubscriber.java
@@ -55,6 +55,10 @@ public void waitUntilAnyMessageReceived() {
         await().atMost(adjust(new Duration(DEFAULT_WAIT_TIME_IN_SEC, SECONDS))).until(() ->
             assertThat(receivedRequests.size()).isPositive());
     }
+    public void waitUntilReceived(Duration duration, int numberOfExpectedMessages) {
+        await().atMost(adjust(duration)).until(() ->
+                assertThat(receivedRequests.size()).isEqualTo(numberOfExpectedMessages));
+    }
 
     public void waitUntilRequestReceived(Consumer<LoggedRequest> requestConsumer) {
         waitUntilAnyMessageReceived();
@@ -64,6 +68,16 @@ public void waitUntilRequestReceived(Consumer<LoggedRequest> requestConsumer) {
         }
     }
 
+    public void waitUntilRequestsReceived(Consumer<List<LoggedRequest>> requestsConsumer) {
+        await().atMost(adjust(new Duration(DEFAULT_WAIT_TIME_IN_SEC, SECONDS))).until(
+                () -> {
+                    synchronized (receivedRequests) {
+                        requestsConsumer.accept(receivedRequests);
+                    }
+                }
+        );
+    }
+
     public void noMessagesReceived() {
         assertThat(receivedRequests).isEmpty();
     }
@@ -97,7 +111,7 @@ private LoggedRequest getFirstReceivedRequest() {
         }
     }
 
-    private LoggedRequest getLastReceivedRequest() {
+    public LoggedRequest getLastReceivedRequest() {
         synchronized (receivedRequests) {
             return Streams.findLast(receivedRequests.stream()).orElseThrow(NoSuchElementException::new);
         }

diff --git a/...mon/java/pl/allegro/tech/hermes/integrationtests/subscriber/TestSubscribersExtension.java b/...mon/java/pl/allegro/tech/hermes/integrationtests/subscriber/TestSubscribersExtension.java
@@ -39,16 +39,25 @@ public TestSubscribersExtension() {
     }
 
     public TestSubscriber createSubscriber(String endpointPathSuffix) {
+        return createSubscriber(OK.getStatusCode(), endpointPathSuffix);
+    }
+
+    public TestSubscriber createSubscriber() {
+        return createSubscriber("");
+    }
+
+    public TestSubscriber createSubscriber(int statusCode) {
+        return createSubscriber(statusCode, "");
+    }
+
+    public TestSubscriber createSubscriber(int statusCode, String endpointPathSuffix) {
         String path = createPath(endpointPathSuffix);
-        service.addStubMapping(post(urlPathEqualTo(path)).willReturn(aResponse().withStatus(OK.getStatusCode())).build());
+        service.addStubMapping(post(urlPathEqualTo(path)).willReturn(aResponse().withStatus(statusCode)).build());
         TestSubscriber subscriber = new TestSubscriber(createSubscriberURI(path));
         subscribersPerPath.put(path, subscriber);
         return subscriber;
     }
 
-    public TestSubscriber createSubscriber() {
-        return createSubscriber("");
-    }
 
     public TestSubscriber createSubscriberWithRetry(String message, int delay) {
         String path = createPath("");
@@ -84,7 +93,18 @@ public TestSubscriber createSubscriberWithRetry(String message, int delay) {
 
     private String createPath(String pathSuffix) {
         return "/subscriber-" + subscriberIndex.incrementAndGet() + pathSuffix;
+    }
+
+    public interface SubscriberScenario {
+        void apply(WireMockServer subscriber, String endpoint);
+    }
 
+    public TestSubscriber createSubscriber(SubscriberScenario scenario) {
+        String path = createPath("");
+        scenario.apply(service, path);
+        TestSubscriber subscriber = new TestSubscriber(createSubscriberURI(path));
+        subscribersPerPath.put(path, subscriber);
+        return subscriber;
     }
 
     private URI createSubscriberURI(String path) {