chore: refactor Spring Integration Pub/Sub Sample tests to use unique resources

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-polling-receiver/src/main/java/com/example/PollingReceiverApplication.java

@@ -23,6 +23,8 @@
 import com.google.cloud.spring.pubsub.support.GcpPubSubHeaders;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.Bean;
@@ -39,18 +41,22 @@
  */
 @SpringBootApplication
 public class PollingReceiverApplication {
-
   private static final Log LOGGER = LogFactory.getLog(PollingReceiverApplication.class);
 
   public static void main(String[] args) {
     SpringApplication.run(PollingReceiverApplication.class, args);
   }
 
+  @Bean
+  public String subscriptionName(@Value("${subscriptionName}") String subscriptionName) {
+    return subscriptionName;
+  }
+
   @Bean
   @InboundChannelAdapter(channel = "pubsubInputChannel", poller = @Poller(fixedDelay = "100"))
-  public MessageSource<Object> pubsubAdapter(PubSubTemplate pubSubTemplate) {
-    PubSubMessageSource messageSource =
-        new PubSubMessageSource(pubSubTemplate, "exampleSubscription");
+  public MessageSource<Object> pubsubAdapter(
+      PubSubTemplate pubSubTemplate, @Qualifier("subscriptionName") String subscriptionName) {
+    PubSubMessageSource messageSource = new PubSubMessageSource(pubSubTemplate, subscriptionName);
     messageSource.setMaxFetchSize(5);
     messageSource.setAckMode(AckMode.MANUAL);
     messageSource.setPayloadType(String.class);

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-polling-receiver/src/main/resources/application.properties

@@ -0,0 +1 @@
+subscriptionName=exampleSubscription

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-polling-receiver/src/test/java/com/example/PollingReceiverIntegrationTest.java

@@ -18,10 +18,20 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
 import com.google.cloud.spring.pubsub.core.PubSubTemplate;
+import com.google.pubsub.v1.ProjectName;
+import com.google.pubsub.v1.Subscription;
+import com.google.pubsub.v1.Topic;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -30,6 +40,7 @@
 import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
 /**
@@ -41,22 +52,61 @@
 @EnabledIfSystemProperty(named = "it.pubsub-integration", matches = "true")
 @ExtendWith(SpringExtension.class)
 @ExtendWith(OutputCaptureExtension.class)
-@SpringBootTest
+@SpringBootTest(
+    properties = {"spring.main.allow-bean-definition-overriding=true"},
+    classes = {PollingReceiverIntegrationTestConfiguration.class})
+@TestPropertySource(locations = "classpath:application-test.properties")
 @DirtiesContext
 class PollingReceiverIntegrationTest {
 
+  private static final String PROJECT_NAME =
+      ProjectName.of(ServiceOptions.getDefaultProjectId()).getProject();
+  @Autowired private Topic testTopic;
+  @Autowired private Subscription testSubscription;
   @Autowired private PubSubTemplate pubSubTemplate;
+  @Autowired private TopicAdminClient topicAdminClient;
+  @Autowired private SubscriptionAdminClient subscriptionAdminClient;
 
   @Test
-  void testSample(CapturedOutput capturedOutput) throws Exception {
+  void testSample(CapturedOutput capturedOutput) {
     String message = "test message " + UUID.randomUUID();
     String expectedString = "Message arrived by Synchronous Pull! Payload: " + message;
 
-    this.pubSubTemplate.publish("exampleTopic", message);
+    this.pubSubTemplate.publish(testTopic.getName(), message);
 
     Awaitility.await()
-        .atMost(60, TimeUnit.SECONDS)
+        .atMost(180, TimeUnit.SECONDS)
         .until(() -> capturedOutput.toString().contains(expectedString));
     assertThat(capturedOutput.toString()).contains(expectedString);
   }
+
+  @AfterEach
+  void cleanUp() {
+    List<String> projectTopics = fetchTopicNamesFromProject();
+    String topicName = testTopic.getName();
+    if (projectTopics.contains(topicName)) {
+      this.topicAdminClient.deleteTopic(topicName);
+    }
+    List<String> projectSubscriptions = fetchSubscriptionNamesFromProject();
+    String subscriptionName = testSubscription.getName();
+    if (projectSubscriptions.contains(subscriptionName)) {
+      this.subscriptionAdminClient.deleteSubscription(subscriptionName);
+    }
+  }
+
+  private List<String> fetchTopicNamesFromProject() {
+    TopicAdminClient.ListTopicsPagedResponse listTopicsResponse =
+        topicAdminClient.listTopics("projects/" + PROJECT_NAME);
+    return StreamSupport.stream(listTopicsResponse.iterateAll().spliterator(), false)
+        .map(Topic::getName)
+        .collect(Collectors.toList());
+  }
+
+  private List<String> fetchSubscriptionNamesFromProject() {
+    SubscriptionAdminClient.ListSubscriptionsPagedResponse response =
+        subscriptionAdminClient.listSubscriptions("projects/" + PROJECT_NAME);
+    return StreamSupport.stream(response.iterateAll().spliterator(), false)
+        .map(Subscription::getName)
+        .collect(Collectors.toList());
+  }
 }

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-polling-receiver/src/test/java/com/example/PollingReceiverIntegrationTestConfiguration.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.pubsub.v1.ProjectName;
+import com.google.pubsub.v1.PushConfig;
+import com.google.pubsub.v1.Subscription;
+import com.google.pubsub.v1.SubscriptionName;
+import com.google.pubsub.v1.Topic;
+import com.google.pubsub.v1.TopicName;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+
+@TestConfiguration
+public class PollingReceiverIntegrationTestConfiguration {
+
+  private String topicName;
+  private TopicAdminClient topicAdminClient;
+
+  private SubscriptionAdminClient subscriptionAdminClient;
+
+  public PollingReceiverIntegrationTestConfiguration(
+      TopicAdminClient topicAdminClient,
+      SubscriptionAdminClient subscriptionAdminClient,
+      @Value("${topicName}") String topicName) {
+    this.topicAdminClient = topicAdminClient;
+    this.subscriptionAdminClient = subscriptionAdminClient;
+    this.topicName = topicName;
+  }
+
+  @Bean
+  public Topic createTopic() {
+    String projectName = ProjectName.of(ServiceOptions.getDefaultProjectId()).getProject();
+    return topicAdminClient.createTopic(TopicName.of(projectName, this.topicName));
+  }
+
+  @Bean
+  public Subscription createSubscription(@Qualifier("subscriptionName") String subscriptionName) {
+    String projectName = ProjectName.of(ServiceOptions.getDefaultProjectId()).getProject();
+    return subscriptionAdminClient.createSubscription(
+        SubscriptionName.of(projectName, subscriptionName),
+        TopicName.of(projectName, this.topicName),
+        PushConfig.getDefaultInstance(),
+        10);
+  }
+}

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-polling-receiver/src/test/resources/application-test.properties

@@ -0,0 +1,2 @@
+topicName=topic-name-${random.uuid}
+subscriptionName=subscription-name-${random.uuid}

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-receiver/src/main/java/com/example/ReceiverApplication.java

@@ -25,6 +25,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.Bean;
@@ -49,16 +50,24 @@ public static void main(String[] args) throws IOException {
     System.in.read();
   }
 
+  @Bean
+  public String subscriptionName(@Value("${subscriptionName}") String subscriptionName) {
+    return subscriptionName;
+  }
+
   @Bean
   public MessageChannel pubsubInputChannel() {
     return new DirectChannel();
   }
 
+
   @Bean
   public PubSubInboundChannelAdapter messageChannelAdapter(
-      @Qualifier("pubsubInputChannel") MessageChannel inputChannel, PubSubTemplate pubSubTemplate) {
+      @Qualifier("pubsubInputChannel") MessageChannel inputChannel,
+      PubSubTemplate pubSubTemplate,
+      @Qualifier("subscriptionName") String subscriptionName) {
     PubSubInboundChannelAdapter adapter =
-        new PubSubInboundChannelAdapter(pubSubTemplate, "exampleSubscription");
+        new PubSubInboundChannelAdapter(pubSubTemplate, subscriptionName);
     adapter.setOutputChannel(inputChannel);
     adapter.setAckMode(AckMode.MANUAL);
     adapter.setPayloadType(String.class);

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-receiver/src/main/resources/application.properties

@@ -1 +1 @@
-
+subscriptionName=exampleSubscription

spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample/spring-cloud-gcp-integration-pubsub-sample-receiver/src/test/java/com/example/ReceiverIntegrationTest.java

@@ -18,10 +18,20 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
 import com.google.cloud.spring.pubsub.core.PubSubTemplate;
+import com.google.pubsub.v1.ProjectName;
+import com.google.pubsub.v1.Subscription;
+import com.google.pubsub.v1.Topic;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -30,6 +40,7 @@
 import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
 /**
@@ -40,22 +51,61 @@
 @EnabledIfSystemProperty(named = "it.pubsub-integration", matches = "true")
 @ExtendWith(SpringExtension.class)
 @ExtendWith(OutputCaptureExtension.class)
-@SpringBootTest
+@SpringBootTest(
+    properties = {"spring.main.allow-bean-definition-overriding=true"},
+    classes = {ReceiverTestConfiguration.class})
+@TestPropertySource(locations = "classpath:application-test.properties")
 @DirtiesContext
 class ReceiverIntegrationTest {
 
+  private static final String PROJECT_NAME =
+      ProjectName.of(ServiceOptions.getDefaultProjectId()).getProject();
+  @Autowired Subscription testSubscription;
   @Autowired private PubSubTemplate pubSubTemplate;
+  @Autowired private Topic testTopic;
+  @Autowired private TopicAdminClient topicAdminClient;
+  @Autowired private SubscriptionAdminClient subscriptionAdminClient;
 
   @Test
-  void testSample(CapturedOutput capturedOutput) throws Exception {
+  void testSample(CapturedOutput capturedOutput) {
     String message = "test message " + UUID.randomUUID();
     String expectedString = "Message arrived! Payload: " + message;
 
-    this.pubSubTemplate.publish("exampleTopic", message);
+    this.pubSubTemplate.publish(testTopic.getName(), message);
 
     Awaitility.await()
         .atMost(60, TimeUnit.SECONDS)
         .until(() -> capturedOutput.toString().contains(expectedString));
     assertThat(capturedOutput.toString()).contains(expectedString);
   }
+
+  @AfterEach
+  void cleanUp() {
+    List<String> projectTopics = fetchTopicNamesFromProject();
+    String topicName = testTopic.getName();
+    if (projectTopics.contains(topicName)) {
+      this.topicAdminClient.deleteTopic(topicName);
+    }
+    List<String> projectSubscriptions = fetchSubscriptionNamesFromProject();
+    String subscriptionName = testSubscription.getName();
+    if (projectSubscriptions.contains(subscriptionName)) {
+      this.subscriptionAdminClient.deleteSubscription(subscriptionName);
+    }
+  }
+
+  private List<String> fetchTopicNamesFromProject() {
+    TopicAdminClient.ListTopicsPagedResponse listTopicsResponse =
+        topicAdminClient.listTopics("projects/" + PROJECT_NAME);
+    return StreamSupport.stream(listTopicsResponse.iterateAll().spliterator(), false)
+        .map(Topic::getName)
+        .collect(Collectors.toList());
+  }
+
+  private List<String> fetchSubscriptionNamesFromProject() {
+    SubscriptionAdminClient.ListSubscriptionsPagedResponse response =
+        subscriptionAdminClient.listSubscriptions("projects/" + PROJECT_NAME);
+    return StreamSupport.stream(response.iterateAll().spliterator(), false)
+        .map(Subscription::getName)
+        .collect(Collectors.toList());
+  }
 }