Pubsub new tests e2e

src/e2e-test/features/pubsub/pubsubadditionalfeature/PubSubToPubSub.feature

@@ -0,0 +1,166 @@
+# Copyright © 2024 Cask Data, Inc.
+#
+# 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
+#
+# http://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.
+
+@PubSub_DataStream
+Feature: PubSub - Verification of successful data transfer from DataStream PubSub source to PubSub sink using different file formats
+
+  @PUBSUB_SOURCE_TEST @PUBSUB_SINK_TEST @PUBSUB_SUBSCRIPTION_TEST
+  Scenario: Validate successful transfer of records from PubSub(source) to PubSub(sink).
+    Given Open Datafusion Project to configure pipeline
+    When Select data pipeline type as: "Realtime"
+    When Expand Plugin group in the LHS plugins list: "Source"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Source"
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Sink"
+    Then Connect plugins: "Pub/Sub" and "Pub/Sub2" to establish connection
+    Then Navigate to the properties page of plugin: "Pub/Sub"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSourceReferenceName"
+    Then Enter PubSub source property subscription name
+    Then Enter PubSub source property topic name
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "Pub/Sub2"
+    Then Replace input plugin property: "project" with value: "projectId"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSinkReferenceName"
+    Then Enter PubSub sink property topic name
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    And Click on configure button
+    And Click on pipeline config
+    And Click on batch time and select format
+    Then Save the pipeline
+    Then Deploy the pipeline
+    Then Run the Pipeline in Runtime
+    Then Wait for pipeline to be in status: "Running" with a timeout of 240 seconds
+    Then Publish the messages
+    Then Subscribe to the messages
+    Then Validate OUT record count is equal to IN record count
+    And Stop the pipeline
+    Then Verify the pipeline status is "Stopped"
+
+  @PUBSUB_SOURCE_TEST @PUBSUB_SINK_TEST @PUBSUB_SUBSCRIPTION_TEST
+  Scenario: Validate successful transfer of records from PubSub(source) to PubSub(sink) using macros.
+    Given Open Datafusion Project to configure pipeline
+    When Select data pipeline type as: "Realtime"
+    When Expand Plugin group in the LHS plugins list: "Source"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Source"
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Sink"
+    Then Connect plugins: "Pub/Sub" and "Pub/Sub2" to establish connection
+    Then Navigate to the properties page of plugin: "Pub/Sub"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSourceReferenceName"
+    Then Click on the Macro button of Property: "topic" and set the value to: "pubSubSourceTopic"
+    Then Click on the Macro button of Property: "subscription" and set the value to: "pubSubSourceSubscription"
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "Pub/Sub2"
+    Then Replace input plugin property: "project" with value: "projectId"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSinkReferenceName"
+    Then Enter PubSub sink property topic name
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    And Click on configure button
+    And Click on pipeline config
+    And Click on batch time and select format
+    Then Save the pipeline
+    Then Deploy the pipeline
+    Then Enter runtime argument value for PubSub source property topic key "pubSubSourceTopic"
+    Then Enter runtime argument value for PubSub source property subscription key "pubSubSourceSubscription"
+    Then Run the Pipeline in Runtime with runtime arguments
+    Then Wait for pipeline to be in status: "Running" with a timeout of 240 seconds
+    Then Publish the messages
+    Then Subscribe to the messages
+    Then Validate OUT record count is equal to IN record count
+    And Stop the pipeline
+    Then Verify the pipeline status is "Stopped"
+
+  @PUBSUB_SOURCE_TEST @PUBSUB_SINK_TEST @PUBSUB_SUBSCRIPTION_TEST
+  Scenario: Validate the successful transfer of records from a pubSub source to a pubSub sink with format Text at both source and sink
+    Given Open Datafusion Project to configure pipeline
+    When Select data pipeline type as: "Realtime"
+    When Expand Plugin group in the LHS plugins list: "Source"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Source"
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Sink"
+    Then Connect plugins: "Pub/Sub" and "Pub/Sub2" to establish connection
+    Then Navigate to the properties page of plugin: "Pub/Sub"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSourceReferenceName"
+    Then Enter PubSub source property subscription name
+    Then Enter PubSub source property topic name
+    Then Select dropdown plugin property: "select-format" with option value: "text"
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "Pub/Sub2"
+    Then Replace input plugin property: "project" with value: "projectId"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSinkReferenceName"
+    Then Enter PubSub sink property topic name
+    Then Select dropdown plugin property: "select-format" with option value: "text"
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    And Click on configure button
+    And Click on pipeline config
+    And Click on batch time and select format
+    Then Save the pipeline
+    Then Deploy the pipeline
+    Then Run the Pipeline in Runtime
+    Then Wait for pipeline to be in status: "Running" with a timeout of 240 seconds
+    Then Publish the messages
+    Then Subscribe to the messages
+    Then Validate OUT record count is equal to IN record count
+    And Stop the pipeline
+    Then Verify the pipeline status is "Stopped"
+
+  @PUBSUB_SOURCE_TEST @PUBSUB_SINK_TEST @PUBSUB_SUBSCRIPTION_TEST
+  Scenario: Validate the successful transfer of records from a pubSub source to a pubSub sink with format Text at source and Json at sink
+    Given Open Datafusion Project to configure pipeline
+    When Select data pipeline type as: "Realtime"
+    When Expand Plugin group in the LHS plugins list: "Source"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Source"
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "Pub/Sub" from the plugins list as: "Sink"
+    Then Connect plugins: "Pub/Sub" and "Pub/Sub2" to establish connection
+    Then Navigate to the properties page of plugin: "Pub/Sub"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSourceReferenceName"
+    Then Enter PubSub source property subscription name
+    Then Enter PubSub source property topic name
+    Then Select dropdown plugin property: "select-format" with option value: "text"
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "Pub/Sub2"
+    Then Replace input plugin property: "project" with value: "projectId"
+    Then Override Service account details if set in environment variables
+    Then Enter input plugin property: "referenceName" with value: "PubSubSinkReferenceName"
+    Then Enter PubSub sink property topic name
+    Then Select dropdown plugin property: "select-format" with option value: "json"
+    Then Validate "Pub/Sub" plugin properties
+    And Close the Plugin Properties page
+    And Click on configure button
+    And Click on pipeline config
+    And Click on batch time and select format
+    Then Save the pipeline
+    Then Deploy the pipeline
+    Then Run the Pipeline in Runtime
+    Then Wait for pipeline to be in status: "Running" with a timeout of 240 seconds
+    Then Publish the messages
+    Then Subscribe to the messages
+    Then Validate OUT record count is equal to IN record count
+    And Stop the pipeline
+    Then Verify the pipeline status is "Stopped"

src/e2e-test/java/io/cdap/plugin/common/stepsdesign/TestSetupHooks.java

@@ -21,6 +21,7 @@
 import io.cdap.e2e.pages.actions.CdfConnectionActions;
 import io.cdap.e2e.pages.actions.CdfPluginPropertiesActions;
 import io.cdap.e2e.utils.BigQueryClient;
+import io.cdap.e2e.utils.ConstantsUtil;
 import io.cdap.e2e.utils.PluginPropertyUtils;
 import io.cdap.e2e.utils.StorageClient;
 import io.cdap.plugin.utils.PubSubClient;
@@ -62,6 +63,8 @@ public class TestSetupHooks {
   public static String bqSourceTable2 = StringUtils.EMPTY;
   public static String bqSourceView = StringUtils.EMPTY;
   public static String pubSubTargetTopic = StringUtils.EMPTY;
+  public static String pubSubSourceTopic = StringUtils.EMPTY;
+  public static String pubSubSourceSubscription = StringUtils.EMPTY;
   public static String spannerInstance = StringUtils.EMPTY;
   public static String spannerDatabase = StringUtils.EMPTY;
   public static String spannerSourceTable = StringUtils.EMPTY;
@@ -480,6 +483,56 @@ private static String createGCSBucketWithFilesAndFolder(String folderPath) throw
     return bucketName;
   }
 
+  @Before(order = 1, value = "@PUBSUB_SOURCE_TEST")
+  public static void createSourcePubSubTopic() throws IOException {
+    pubSubSourceTopic = "cdf-e2e-test-" + UUID.randomUUID();
+    PubSubClient.createTopic(pubSubSourceTopic);
+    BeforeActions.scenario.write("Source PubSub topic " + pubSubSourceTopic);
+  }
+
+  @Before(order = 1, value = "@PUBSUB_SUBSCRIPTION_TEST")
+  public static void createSubscriptionPubSubTopic() throws IOException {
+    pubSubSourceSubscription = "cdf-e2e-test-" + UUID.randomUUID();
+    PubSubClient.createSubscription(pubSubSourceSubscription , pubSubSourceTopic);
+    BeforeActions.scenario.write("Source PubSub subscription " + pubSubSourceSubscription);
+  }
+
+  @After(order = 1, value = "@PUBSUB_SOURCE_TEST")
+  public static void deleteSourcePubSubTopic() {
+    try {
+      PubSubClient.deleteTopic(pubSubSourceTopic);
+      BeforeActions.scenario.write("Deleted target PubSub topic " + pubSubSourceTopic);
+      pubSubSourceTopic = StringUtils.EMPTY;
+    } catch (Exception e) {
+      if (e.getMessage().contains("Invalid resource name given") || e.getMessage().contains("Resource not found")) {
+
+      }
+    }
+  }
+
+  @After(order = 1, value = "@PUBSUB_SUBSCRIPTION_TEST")
+  public static void deleteSourcePubSubSubscription() {
+    try {
+      PubSubClient.deleteTopic(pubSubSourceSubscription);
+      BeforeActions.scenario.write("Deleted target PubSub topic " + pubSubSourceSubscription);
+      pubSubSourceSubscription = StringUtils.EMPTY;
+    } catch (Exception e) {
+      if (e.getMessage().contains("Invalid resource name given") || e.getMessage().contains("Resource not found")) {
+        BeforeActions.scenario.write("Source PubSub topic " + pubSubSourceTopic + " does not exist.");
+      } else {
+        Assert.fail(e.getMessage());
+      }
+    }
+  }
+
+  public static void publishMessage() throws IOException, InterruptedException {
+    String dataMessage1 = PluginPropertyUtils.pluginProp("firstMessage");
+    String dataMessage2 = PluginPropertyUtils.pluginProp("secondMessage");
+    List<String> dataMessagesList = Arrays.asList(dataMessage1, dataMessage2);
+    PubSubClient.publishMessagesWithPubSub(PluginPropertyUtils.pluginProp
+      (ConstantsUtil.PROJECT_ID), pubSubSourceTopic, dataMessagesList);
+  }
+
   @Before(order = 1, value = "@PUBSUB_SINK_TEST")
   public static void createTargetPubSubTopic() {
     pubSubTargetTopic = "cdf-e2e-test-" + UUID.randomUUID();
@@ -1141,7 +1194,7 @@ public static void createSinkBQDedupeTable() throws IOException, InterruptedExce
 
   @Before(value = "@BQ_INSERT_INT_SOURCE_TEST")
   public static void createSourceBQTable() throws IOException, InterruptedException {
-    bqSourceTable = "E2E_TARGET_" + UUID.randomUUID().toString().replaceAll("-", "_");
+    bqSourceTable = "E2E_SOURCE_" + UUID.randomUUID().toString().replaceAll("-", "_");
     PluginPropertyUtils.addPluginProp("bqSourceTable", bqSourceTable);
     BeforeActions.scenario.write("BQ source table name - " + bqSourceTable);
     BigQueryClient.getSoleQueryResult("create table `" + datasetName + "." + bqSourceTable + "` " +

src/e2e-test/java/io/cdap/plugin/pubsub/locators/PubSubLocators.java

@@ -74,4 +74,19 @@ public static WebElement selectedFormat(String format) {
     return SeleniumDriver.getDriver()
       .findElement(By.xpath("//*[@data-cy='select-format']/div[text()='" + format + "']"));
   }
+
+  @FindBy(how = How.XPATH, using = "//span[contains(text(), \"Batch interval\")]//following-sibling::div//select[1]")
+  public static WebElement batchTime;
+
+  @FindBy(how = How.XPATH, using = "//span[contains(text(), \"Batch interval\")]//following-sibling::div//select[2]")
+  public static WebElement timeSelect;
+
+  @FindBy(how = How.XPATH, using = "//button[@data-testid='config-apply-close']")
+  public static WebElement saveButton;
+
+  @FindBy(how = How.XPATH, using = "//*[@data-cy='pipeline-configure-modeless-btn']")
+  public static WebElement configButton;
+
+  @FindBy(how = How.XPATH, using = "//*[@data-cy='tab-content-Pipeline config']")
+  public static WebElement pipelineConfig;
 }

src/e2e-test/java/io/cdap/plugin/pubsub/runners/sinkrunner/TestRunner.java

@@ -27,7 +27,7 @@
   features = {"src/e2e-test/features"},
   glue = {"io.cdap.plugin.pubsub.stepsdesign", "io.cdap.plugin.gcs.stepsdesign",
     "io.cdap.plugin.common.stepsdesign", "io.cdap.plugin.bigquery.stepsdesign", "stepsdesign"},
-  tags = {"@PubSub_Sink"},
+  tags = {"@PubSub_Sink","@PubSub_DataStream"},
   monochrome = true,
   plugin = {"pretty", "html:target/cucumber-html-report/pubsub-sink",
     "json:target/cucumber-reports/cucumber-pubsub-sink.json",

src/e2e-test/java/io/cdap/plugin/pubsub/runners/sinkrunner/TestRunnerRequired.java

@@ -27,7 +27,7 @@
   features = {"src/e2e-test/features"},
   glue = {"io.cdap.plugin.pubsub.stepsdesign", "io.cdap.plugin.gcs.stepsdesign",
     "io.cdap.plugin.common.stepsdesign", "io.cdap.plugin.bigquery.stepsdesign", "stepsdesign"},
-  tags = {"@PubSub_Sink_Required"},
+  tags = {"@PubSub_Sink_Required","@PubSub_DataStream_Required"},
   monochrome = true,
   plugin = {"pretty", "html:target/cucumber-html-report/pubsub-sink-required",
     "json:target/cucumber-reports/cucumber-pubsub-sink-required.json",

src/e2e-test/java/io/cdap/plugin/pubsub/stepsdesign/PubSubSink.java

@@ -27,13 +27,16 @@
 import io.cdap.plugin.utils.E2EHelper;
 import io.cdap.plugin.utils.E2ETestConstants;
 import io.cdap.plugin.utils.PubSubClient;
+import io.cucumber.java.en.And;
 import io.cucumber.java.en.Then;
 import io.cucumber.java.en.When;
 import org.apache.commons.lang3.StringUtils;
 import org.junit.Assert;
+import org.openqa.selenium.support.ui.Select;
 import stepsdesign.BeforeActions;
 
 import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 
 /**
  * PubSub Sink Plugin related step design.
@@ -234,4 +237,31 @@ public void enterRuntimeArgumentValueForPubSubSinkPropertyTopicKey(String runtim
   public void clickOnPreviewDataForPubSubSink() {
     openSinkPluginPreviewData("GooglePublisher");
   }
+
+  @Then("Subscribe to the messages")
+  public void subscribeToTheMessages() throws InterruptedException {
+    TimeUnit time = TimeUnit.SECONDS;
+    time.sleep(60);
+    PubSubClient.subscribeAsyncExample(PluginPropertyUtils.pluginProp(ConstantsUtil.PROJECT_ID),
+                                       TestSetupHooks.pubSubSourceSubscription);
+  }
+
+  @And("Click on batch time and select format")
+  public void clickOnBatchTimeAndSelectFormat() {
+    Select select = new Select(PubSubLocators.batchTime);
+    select.selectByIndex(0);
+    Select selectformat = new Select(PubSubLocators.timeSelect);
+    selectformat.selectByIndex(1);
+    ElementHelper.clickOnElement(PubSubLocators.saveButton);
+  }
+
+  @And("Click on configure button")
+  public void clickOnConfigureButton() {
+    ElementHelper.clickOnElement(PubSubLocators.configButton);
+  }
+
+  @And("Click on pipeline config")
+  public void clickOnPipelineConfig() {
+    ElementHelper.clickOnElement(PubSubLocators.pipelineConfig);
+  }
 }

src/e2e-test/java/io/cdap/plugin/pubsub/stepsdesign/PubSubSource.java

@@ -16,10 +16,17 @@
 
 package io.cdap.plugin.pubsub.stepsdesign;
 
+import io.cdap.e2e.pages.locators.CdfStudioLocators;
 import io.cdap.e2e.utils.CdfHelper;
+import io.cdap.e2e.utils.ElementHelper;
+import io.cdap.plugin.common.stepsdesign.TestSetupHooks;
+import io.cdap.plugin.pubsub.actions.PubSubActions;
 import io.cucumber.java.en.Then;
 import io.cucumber.java.en.When;
 
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
 /**
  * PubSub Source Plugin related step design.
  */
@@ -35,4 +42,32 @@ public void sourceIsPubSub() {
   public void openThePubSubSourceProperties() {
     openSourcePluginProperties("GooglePublisher");
   }
+
+  @Then("Enter PubSub source property topic name")
+  public void enterPubSubSourcePropertyTopicName() {
+    PubSubActions.enterPubSubTopic(TestSetupHooks.pubSubSourceTopic);
+  }
+
+  @Then("Enter PubSub source property subscription name")
+  public void enterPubSubSourcePropertySubscriptionName() {
+    PubSubActions.enterSubscription(TestSetupHooks.pubSubSourceSubscription);
+  }
+
+  @Then("Publish the messages")
+  public void publishTheMessage() throws IOException, InterruptedException {
+    TimeUnit time = TimeUnit.SECONDS;
+    time.sleep(120);
+    TestSetupHooks.publishMessage();
+  }
+
+  @Then("Enter runtime argument value for PubSub source property topic key {string}")
+  public void enterRuntimeArgumentValueForPubSubSourcePropertyTopicKey(String runtimeArgumentKey) {
+    ElementHelper.sendKeys(CdfStudioLocators.runtimeArgsValue(runtimeArgumentKey), TestSetupHooks.pubSubSourceTopic);
+  }
+
+  @Then("Enter runtime argument value for PubSub source property subscription key {string}")
+  public void enterRuntimeArgumentValueForPubSubSourcePropertySubscriptionKey(String runtimeArgumentKey) {
+    ElementHelper.sendKeys(CdfStudioLocators.runtimeArgsValue(runtimeArgumentKey),
+                           TestSetupHooks.pubSubSourceSubscription);
+  }
 }