diff --git a/spring-pulsar-test/spring-pulsar-test.gradle b/spring-pulsar-test/spring-pulsar-test.gradle index 5cf27a17..9fd1c7c8 100644 --- a/spring-pulsar-test/spring-pulsar-test.gradle +++ b/spring-pulsar-test/spring-pulsar-test.gradle @@ -11,4 +11,5 @@ dependencies { implementation project(':spring-pulsar') testImplementation 'org.assertj:assertj-core' testImplementation 'org.junit.jupiter:junit-jupiter' + testImplementation 'org.mockito:mockito-core' } diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java index 2c0b76ac..52dd21ff 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java @@ -42,8 +42,10 @@ import org.springframework.util.Assert; * * @param the type of the message payload * @author Jonas Geiregat + * @author Kartik Shrivastava + * @author Chris Bono */ -public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, ConditionsSpec { +public final class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, ConditionsSpec { private final PulsarConsumerFactory consumerFactory; @@ -57,6 +59,13 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C private boolean untilMethodAlreadyCalled = false; + /** + * Begin a builder which will consume messages from the + * {@link PulsarTestContainerSupport} (if available) or the default Pulsar broker url + * {@code pulsar://localhost:6650}. + * @param the message payload type + * @return the {@link TopicSpec topic step} of the builder + */ public static TopicSpec consumeMessages() { if (PulsarTestContainerSupport.isContainerStarted()) { return PulsarConsumerTestUtil.consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl()); @@ -64,6 +73,12 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C return PulsarConsumerTestUtil.consumeMessages("pulsar://localhost:6650"); } + /** + * Begin a builder which will consume messages from the specified Pulsar broker url. + * @param the message payload type + * @param url the Pulsar broker url + * @return the {@link TopicSpec topic step} of the builder + */ public static TopicSpec consumeMessages(String url) { Assert.notNull(url, "url must not be null"); try { @@ -74,16 +89,28 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C } } + /** + * Begin a builder which will consume messages with a provided Pulsar client. + * @param the message payload type + * @param pulsarClient the client to consume with + * @return the {@link TopicSpec topic step} of the builder + */ public static TopicSpec consumeMessages(PulsarClient pulsarClient) { Assert.notNull(pulsarClient, "pulsarClient must not be null"); return PulsarConsumerTestUtil.consumeMessages(new DefaultPulsarConsumerFactory<>(pulsarClient, List.of())); } + /** + * Begin a builder which will consume messages with a provided consumer factory. + * @param the message payload type + * @param pulsarConsumerFactory the consumer factory to consume with + * @return the {@link TopicSpec topic step} of the builder + */ public static TopicSpec consumeMessages(PulsarConsumerFactory pulsarConsumerFactory) { return new PulsarConsumerTestUtil<>(pulsarConsumerFactory); } - public PulsarConsumerTestUtil(PulsarConsumerFactory consumerFactory) { + private PulsarConsumerTestUtil(PulsarConsumerFactory consumerFactory) { Assert.notNull(consumerFactory, "PulsarConsumerFactory must not be null"); this.consumerFactory = consumerFactory; } diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java index 03e3b6b7..7fa9fbdd 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java @@ -16,6 +16,7 @@ package org.springframework.pulsar.test.support; +import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.testcontainers.containers.PulsarContainer; import org.testcontainers.junit.jupiter.Testcontainers; @@ -25,6 +26,7 @@ import org.testcontainers.utility.DockerImageName; * Provides a static {@link PulsarContainer} that can be shared across test classes. * * @author Chris Bono + * @author Kartik Shrivastava */ @Testcontainers(disabledWithoutDocker = true) public interface PulsarTestContainerSupport { @@ -40,6 +42,11 @@ public interface PulsarTestContainerSupport { PULSAR_CONTAINER.start(); } + @AfterAll + static void stopContainer() { + PULSAR_CONTAINER.stop(); + } + static String getPulsarBrokerUrl() { return PULSAR_CONTAINER.getPulsarBrokerUrl(); } @@ -52,8 +59,4 @@ public interface PulsarTestContainerSupport { return PULSAR_CONTAINER.isRunning(); } - static void stopContainer() { - PULSAR_CONTAINER.stop(); - } - } diff --git a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java index 4da7d0cf..c79caae8 100644 --- a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java +++ b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java @@ -29,9 +29,13 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.springframework.pulsar.PulsarException; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.PulsarConsumerFactory; @@ -41,6 +45,8 @@ import org.springframework.pulsar.core.PulsarTemplate; * Tests for {@link PulsarConsumerTestUtil}. * * @author Jonas Geiregat + * @author Kartik Shrivastava + * @author Chris Bono */ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { @@ -61,9 +67,16 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { this.pulsarTemplate = new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient)); } + @AfterEach + void cleanupFromTest() throws PulsarClientException { + if (this.pulsarClient != null) { + this.pulsarClient.close(); + } + } + @Test - void whenConditionIsSpecifiedMessagesAreConsumedUntilConditionIsMet() { - var topic = testTopic("a"); + void whenConditionIsSpecifiedThenMessagesConsumedUntilConditionMet() { + var topic = testTopic("cond"); IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) .fromTopic(topic) @@ -75,8 +88,8 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { } @Test - void whenConditionIsNotSpecifiedMessagesAreConsumedUntilAwaitDuration() { - var topic = testTopic("b"); + void whenConditionIsNotSpecifiedThenMessagesAreConsumedUntilAwaitDuration() { + var topic = testTopic("no-cond"); IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) .fromTopic(topic) @@ -89,81 +102,8 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { } @Test - void exceptionIsThrownWhenConditionNotMetWithinAwaitDuration() { - assertThatExceptionOfType(ConditionTimeoutException.class) - .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) - .fromTopic(testTopic("c")) - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(5)) - .until(ConsumedMessagesConditions.desiredMessageCount(3)) - .get()) - .withMessage("Condition was not met within 5 seconds"); - } - - @Test - void messagesAreConsumedWhenContainerIsRunningAndConsumeMessagesIsCalledWithoutArguments() { - // depends upon pulsarClient created in prepareForTest - var topic = testTopic("e1"); - IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); - var msgs = PulsarConsumerTestUtil.consumeMessages() - .fromTopic(topic) - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(5)) - .until(desiredMessageCount(2)) - .get(); - assertThat(msgs).hasSize(2); - } - - @Test - void messagesAreConsumedWhenContainerIsStoppedAndConsumeMessagesIsCalledWithoutArguments() { - PulsarTestContainerSupport.stopContainer(); - PulsarConsumerTestUtil.consumeMessages(); - // TODO: Complete this test - } - - @Test - void messagesAreConsumedWhenConsumeMessagesIsCalledWithBrokerUrl() { - var topic = testTopic("e2"); - IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); - var msgs = PulsarConsumerTestUtil.consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .fromTopic(topic) - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(5)) - .until(desiredMessageCount(2)) - .get(); - assertThat(msgs).hasSize(2); - } - - @Test - void exceptionIsThrownWhenConsumeMessagesIsCalledWithNullBrokerUrl() { - String url = null; - assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(url)) - .withMessage("url must not be null"); - } - - @Test - void messagesAreConsumedWhenConsumeMessagesIsCalledWithPulsarClient() { - var topic = testTopic("e3"); - IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); - var msgs = PulsarConsumerTestUtil.consumeMessages(this.pulsarClient) - .fromTopic(topic) - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(5)) - .until(desiredMessageCount(2)) - .get(); - assertThat(msgs).hasSize(2); - } - - @Test - void exceptionIsThrownWhenConsumeMessagesIsCalledWithNullPulsarClient() { - PulsarClient localPulsarClient = null; - assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(localPulsarClient)) - .withMessage("pulsarClient must not be null"); - } - - @Test - void whenChainedConditionAreSpecifiedMessagesAreConsumedUntilTheyAreMet() { - var topic = testTopic("d"); + void whenChainedConditionsAreSpecifiedThenMessagesConsumedUntilAllConditionsMet() { + var topic = testTopic("chained-cond"); IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); ConsumedMessagesCondition condition1 = ConsumedMessagesConditions.desiredMessageCount(5); ConsumedMessagesCondition condition2 = ConsumedMessagesConditions.atLeastOneMessageMatches("message-1"); @@ -177,9 +117,75 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { } @Test - void exceptionIsThrownWhenUntilIsCalledMultipleTimes() { - var topic = testTopic("e"); - IntStream.range(0, 1).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + void whenConditionNotMetWithinAwaitDurationThenExceptionIsThrown() { + assertThatExceptionOfType(ConditionTimeoutException.class) + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic(testTopic("cond-not-met")) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(ConsumedMessagesConditions.desiredMessageCount(3)) + .get()) + .withMessage("Condition was not met within 5 seconds"); + } + + @Test + void consumeMessagesWithNoArgsUsesPulsarContainerIfAvailable() { + var topic = testTopic("no-arg"); + IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages() + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(2)) + .get(); + assertThat(msgs).hasSize(2); + } + + @Test + void consumeMessagesWithNoArgsUsesDefaultUrlWhenPulsarContainerNotAvailable() { + try (MockedStatic containerSupport = Mockito + .mockStatic(PulsarTestContainerSupport.class)) { + containerSupport.when(PulsarTestContainerSupport::isContainerStarted).thenReturn(false); + var topic = testTopic("no-arg-dft-url"); + assertThatExceptionOfType(PulsarException.class) + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages() + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(2)) + .get()) + .withStackTraceContaining("Connection refused: localhost"); + } + } + + @Test + void consumeMessagesWithBrokerUrl() { + var topic = testTopic("url-arg"); + IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(2)) + .get(); + assertThat(msgs).hasSize(2); + } + + @Test + void consumeMessagesWithPulsarClient() { + var topic = testTopic("client-arg"); + IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages(this.pulsarClient) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(2)) + .get(); + assertThat(msgs).hasSize(2); + } + + @Test + void untilCannotBeCalledMultipleTimes() { + var topic = testTopic("until-multi"); assertThatExceptionOfType(IllegalStateException.class) .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) .fromTopic(topic) @@ -191,6 +197,20 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { .withMessage("Multiple calls to 'until' are not allowed. Use 'and' to combine conditions."); } + @Test + void brokerUrlCannotBeNull() { + String url = null; + assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(url)) + .withMessage("url must not be null"); + } + + @Test + void pulsarClientCannotBeNull() { + PulsarClient localPulsarClient = null; + assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(localPulsarClient)) + .withMessage("pulsarClient must not be null"); + } + @Test void consumerFactoryCannotBeNull() { PulsarConsumerFactory consumerFactory = null;