Add docker related unit tests

servicebus/spring-cloud-azure-stream-binder-servicebus/servicebus-queue-binder/pom.xml

@@ -21,7 +21,7 @@
   <properties>
     <maven.compiler.source>17</maven.compiler.source>
     <maven.compiler.target>17</maven.compiler.target>
-    <version.spring.cloud.azure>7.0.0</version.spring.cloud.azure>
+    <version.spring.cloud.azure>7.1.0-beta.1</version.spring.cloud.azure>
   </properties>
 
   <dependencyManagement>
@@ -60,6 +60,46 @@
       <groupId>org.hibernate.validator</groupId>
       <artifactId>hibernate-validator</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.azure.spring</groupId>
+      <artifactId>spring-cloud-azure-autoconfigure</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.azure.spring</groupId>
+      <artifactId>spring-cloud-azure-docker-compose</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-testcontainers</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers-azure</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers-junit-jupiter</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.azure.spring</groupId>
+      <artifactId>spring-cloud-azure-testcontainers</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.microsoft.sqlserver</groupId>
+      <artifactId>mssql-jdbc</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
 </project>

servicebus/spring-cloud-azure-stream-binder-servicebus/servicebus-queue-binder/src/test/java/com/azure/spring/sample/servicebus/queue/binder/ServiceBusDockerComposeTest.java

@@ -0,0 +1,94 @@
+package com.azure.spring.sample.servicebus.queue.binder;
+
+import com.azure.spring.cloud.autoconfigure.implementation.context.AzureGlobalPropertiesAutoConfiguration;
+import com.azure.spring.cloud.autoconfigure.implementation.servicebus.AzureServiceBusAutoConfiguration;
+import com.azure.spring.cloud.autoconfigure.implementation.servicebus.AzureServiceBusMessagingAutoConfiguration;
+import com.azure.spring.messaging.checkpoint.Checkpointer;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.waitAtMost;
+
+@SpringBootTest(properties = {
+        "spring.docker.compose.skip.in-tests=false",
+        "spring.docker.compose.file=classpath:servicebus-compose.yaml",
+        "spring.docker.compose.stop.command=down",
+        "spring.cloud.function.definition=consume;supply",
+        "spring.cloud.stream.bindings.consume-in-0.destination=queue.1",
+        "spring.cloud.stream.bindings.supply-out-0.destination=queue.1",
+        "spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete=false",
+        "spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type=queue",
+        "spring.cloud.stream.poller.fixed-delay=1000",
+        "spring.cloud.stream.poller.initial-delay=0"
+})
+class ServiceBusDockerComposeTest {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusDockerComposeTest.class);
+    private static final Set<String> RECEIVED_MESSAGES = ConcurrentHashMap.newKeySet();
+    private static final AtomicInteger messageSequence = new AtomicInteger(0);
+
+    @Test
+    void supplierAndConsumerShouldWorkThroughServiceBusQueue() {
+        waitAtMost(Duration.ofSeconds(60))
+            .pollDelay(Duration.ofSeconds(2))
+            .untilAsserted(() -> {
+                assertThat(RECEIVED_MESSAGES).isNotEmpty();
+                LOGGER.info("✓ Test passed - Consumer received {} message(s)", RECEIVED_MESSAGES.size());
+            });
+    }
+
+    @Configuration(proxyBeanMethods = false)
+    @EnableAutoConfiguration
+    @ImportAutoConfiguration(classes = {
+            AzureGlobalPropertiesAutoConfiguration.class,
+            AzureServiceBusAutoConfiguration.class,
+            AzureServiceBusMessagingAutoConfiguration.class})
+    static class Config {
+
+        @Bean
+        public Supplier<Message<String>> supply() {
+            return () -> {
+                int sequence = messageSequence.getAndIncrement();
+                String payload = "Hello world, " + sequence;
+                LOGGER.info("[Supplier] Invoked - message sequence: {}", sequence);
+                return MessageBuilder.withPayload(payload).build();
+            };
+        }
+
+        @Bean
+        public Consumer<Message<String>> consume() {
+            return message -> {
+                String payload = message.getPayload();
+                RECEIVED_MESSAGES.add(payload);
+                LOGGER.info("[Consumer] Received message: {}", payload);
+
+                Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
+                if (checkpointer != null) {
+                    checkpointer.success()
+                            .doOnSuccess(s -> LOGGER.info("[Consumer] Message checkpointed"))
+                            .doOnError(e -> LOGGER.error("[Consumer] Checkpoint failed", e))
+                            .block();
+                }
+            };
+        }
+    }
+}
+
+

servicebus/spring-cloud-azure-stream-binder-servicebus/servicebus-queue-binder/src/test/java/com/azure/spring/sample/servicebus/queue/binder/ServiceBusTestContainerTest.java

@@ -0,0 +1,117 @@
+package com.azure.spring.sample.servicebus.queue.binder;
+
+import com.azure.spring.cloud.autoconfigure.implementation.context.AzureGlobalPropertiesAutoConfiguration;
+import com.azure.spring.cloud.autoconfigure.implementation.servicebus.AzureServiceBusAutoConfiguration;
+import com.azure.spring.cloud.autoconfigure.implementation.servicebus.AzureServiceBusMessagingAutoConfiguration;
+import com.azure.spring.messaging.checkpoint.Checkpointer;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.testcontainers.azure.ServiceBusEmulatorContainer;
+import org.testcontainers.containers.MSSQLServerContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.MountableFile;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.waitAtMost;
+
+@SpringJUnitConfig
+@TestPropertySource(properties = {
+        "spring.cloud.function.definition=consume;supply",
+        "spring.cloud.stream.bindings.consume-in-0.destination=queue.1",
+        "spring.cloud.stream.bindings.supply-out-0.destination=queue.1",
+        "spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete=false",
+        "spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type=queue",
+        "spring.cloud.stream.poller.fixed-delay=1000",
+        "spring.cloud.stream.poller.initial-delay=0"})
+@Testcontainers
+class ServiceBusTestContainerTest {
+
+    private static final Network NETWORK = Network.newNetwork();
+
+    private static final MSSQLServerContainer<?> SQLSERVER = new MSSQLServerContainer<>(
+            "mcr.microsoft.com/mssql/server:2022-CU14-ubuntu-22.04")
+            .acceptLicense()
+            .withNetwork(NETWORK)
+            .withNetworkAliases("sqlserver");
+
+    @Container
+    @ServiceConnection
+    private static final ServiceBusEmulatorContainer SERVICE_BUS = new ServiceBusEmulatorContainer(
+            "mcr.microsoft.com/azure-messaging/servicebus-emulator:latest")
+            .acceptLicense()
+            .withCopyFileToContainer(MountableFile.forClasspathResource("Config.json"),
+                    "/ServiceBus_Emulator/ConfigFiles/Config.json")
+            .withNetwork(NETWORK)
+            .withMsSqlServerContainer(SQLSERVER);
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusDockerComposeTest.class);
+    private static final Set<String> RECEIVED_MESSAGES = ConcurrentHashMap.newKeySet();
+    private static final AtomicInteger messageSequence = new AtomicInteger(0);
+
+    @Test
+    void supplierAndConsumerShouldWorkThroughServiceBusQueue() {
+        waitAtMost(Duration.ofSeconds(60))
+                .pollDelay(Duration.ofSeconds(2))
+                .untilAsserted(() -> {
+                    assertThat(RECEIVED_MESSAGES).isNotEmpty();
+                    LOGGER.info("✓ Test passed - Consumer received {} message(s)", RECEIVED_MESSAGES.size());
+                });
+    }
+
+    @Configuration(proxyBeanMethods = false)
+    @EnableAutoConfiguration
+    @ImportAutoConfiguration(classes = {
+            AzureGlobalPropertiesAutoConfiguration.class,
+            AzureServiceBusAutoConfiguration.class,
+            AzureServiceBusMessagingAutoConfiguration.class})
+    static class Config {
+
+        @Bean
+        public Supplier<Message<String>> supply() {
+            return () -> {
+                int sequence = messageSequence.getAndIncrement();
+                String payload = "Hello world, " + sequence;
+                LOGGER.info("[Supplier] Invoked - message sequence: {}", sequence);
+                return MessageBuilder.withPayload(payload).build();
+            };
+        }
+
+        @Bean
+        public Consumer<Message<String>> consume() {
+            return message -> {
+                String payload = message.getPayload();
+                RECEIVED_MESSAGES.add(payload);
+                LOGGER.info("[Consumer] Received message: {}", payload);
+
+                Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
+                if (checkpointer != null) {
+                    checkpointer.success()
+                            .doOnSuccess(s -> LOGGER.info("[Consumer] Message checkpointed"))
+                            .doOnError(e -> LOGGER.error("[Consumer] Checkpoint failed", e))
+                            .block();
+                }
+            };
+        }
+    }
+}
+

servicebus/spring-cloud-azure-stream-binder-servicebus/servicebus-queue-binder/src/test/resources/Config.json

@@ -0,0 +1,108 @@
+{
+  "UserConfig": {
+    "Namespaces": [
+      {
+        "Name": "sbemulatorns",
+        "Queues": [
+          {
+            "Name": "queue.1",
+            "Properties": {
+              "DeadLetteringOnMessageExpiration": false,
+              "DefaultMessageTimeToLive": "PT1H",
+              "DuplicateDetectionHistoryTimeWindow": "PT20S",
+              "ForwardDeadLetteredMessagesTo": "",
+              "ForwardTo": "",
+              "LockDuration": "PT1M",
+              "MaxDeliveryCount": 10,
+              "RequiresDuplicateDetection": false,
+              "RequiresSession": false
+            }
+          }
+        ],
+
+        "Topics": [
+          {
+            "Name": "topic.1",
+            "Properties": {
+              "DefaultMessageTimeToLive": "PT1H",
+              "DuplicateDetectionHistoryTimeWindow": "PT20S",
+              "RequiresDuplicateDetection": false
+            },
+            "Subscriptions": [
+              {
+                "Name": "subscription.1",
+                "Properties": {
+                  "DeadLetteringOnMessageExpiration": false,
+                  "DefaultMessageTimeToLive": "PT1H",
+                  "LockDuration": "PT1M",
+                  "MaxDeliveryCount": 10,
+                  "ForwardDeadLetteredMessagesTo": "",
+                  "ForwardTo": "",
+                  "RequiresSession": false
+                },
+                "Rules": [
+                  {
+                    "Name": "app-prop-filter-1",
+                    "Properties": {
+                      "FilterType": "Correlation",
+                      "CorrelationFilter": {
+                        "ContentType": "application/text",
+                        "CorrelationId": "id1",
+                        "Label": "subject1",
+                        "MessageId": "msgid1",
+                        "ReplyTo": "someQueue",
+                        "ReplyToSessionId": "sessionId",
+                        "SessionId": "session1",
+                        "To": "xyz"
+                      }
+                    }
+                  }
+                ]
+              },
+              {
+                "Name": "subscription.2",
+                "Properties": {
+                  "DeadLetteringOnMessageExpiration": false,
+                  "DefaultMessageTimeToLive": "PT1H",
+                  "LockDuration": "PT1M",
+                  "MaxDeliveryCount": 10,
+                  "ForwardDeadLetteredMessagesTo": "",
+                  "ForwardTo": "",
+                  "RequiresSession": false
+                },
+                "Rules": [
+                  {
+                    "Name": "user-prop-filter-1",
+                    "Properties": {
+                      "FilterType": "Correlation",
+                      "CorrelationFilter": {
+                        "Properties": {
+                          "prop3": "value3"
+                        }
+                      }
+                    }
+                  }
+                ]
+              },
+              {
+                "Name": "subscription.3",
+                "Properties": {
+                  "DeadLetteringOnMessageExpiration": false,
+                  "DefaultMessageTimeToLive": "PT1H",
+                  "LockDuration": "PT1M",
+                  "MaxDeliveryCount": 10,
+                  "ForwardDeadLetteredMessagesTo": "",
+                  "ForwardTo": "",
+                  "RequiresSession": false
+                }
+              }
+            ]
+          }
+        ]
+      }
+    ],
+    "Logging": {
+      "Type": "File"
+    }
+  }
+}