Skip to content

Commit 1ef12f7

Browse files
AnkitCLIAnkitCLI
committed
Pubsub new e2e tests
1 parent 0555650 commit 1ef12f7

12 files changed

Lines changed: 1349 additions & 4 deletions

File tree

‎pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@
6969
<properties>
7070
<jee.version>7</jee.version>
7171
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
72-
<avro.version>1.8.2</avro.version>
72+
<avro.version>1.11.3</avro.version>
7373
<bigquery.connector.hadoop2.version>hadoop2-1.2.0</bigquery.connector.hadoop2.version>
7474
<commons.codec.version>1.4</commons.codec.version>
7575
<cdap.version>6.9.1</cdap.version>
@@ -80,7 +80,7 @@
8080
<google.cloud.bigtable.version>1.17.1</google.cloud.bigtable.version>
8181
<google.cloud.bigquery.version>1.137.1</google.cloud.bigquery.version>
8282
<google.cloud.kms.version>2.0.2</google.cloud.kms.version>
83-
<google.cloud.pubsub.version>1.108.1</google.cloud.pubsub.version>
83+
<google.cloud.pubsub.version>1.112.1</google.cloud.pubsub.version>
8484
<google.cloud.spanner.version>6.10.1</google.cloud.spanner.version>
8585
<google.cloud.speech.version>1.24.7</google.cloud.speech.version>
8686
<google.cloud.storage.version>2.3.0</google.cloud.storage.version>
@@ -638,7 +638,7 @@
638638
<dependency>
639639
<groupId>org.apache.avro</groupId>
640640
<artifactId>avro-mapred</artifactId>
641-
<classifier>hadoop2</classifier>
641+
<!-- <classifier>hadoop2</classifier>-->
642642
<version>${avro.version}</version>
643643
</dependency>
644644
<dependency>

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

Lines changed: 456 additions & 0 deletions
Large diffs are not rendered by default.

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

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import com.google.cloud.bigquery.BigQueryException;
1919
import com.google.cloud.storage.Blob;
2020
import com.google.cloud.storage.StorageException;
21+
import com.google.pubsub.v1.Encoding;
2122
import io.cdap.e2e.pages.actions.CdfConnectionActions;
2223
import io.cdap.e2e.pages.actions.CdfPluginPropertiesActions;
2324
import io.cdap.e2e.utils.BigQueryClient;
@@ -62,6 +63,9 @@ public class TestSetupHooks {
6263
public static String bqSourceTable2 = StringUtils.EMPTY;
6364
public static String bqSourceView = StringUtils.EMPTY;
6465
public static String pubSubTargetTopic = StringUtils.EMPTY;
66+
public static String pubSubSourceTopic = StringUtils.EMPTY;
67+
public static String pubSubSourceSubscription = StringUtils.EMPTY;
68+
public static String pubSubSchemaId = StringUtils.EMPTY;
6569
public static String spannerInstance = StringUtils.EMPTY;
6670
public static String spannerDatabase = StringUtils.EMPTY;
6771
public static String spannerSourceTable = StringUtils.EMPTY;
@@ -481,12 +485,102 @@ private static String createGCSBucketWithFilesAndFolder(String folderPath) throw
481485
return bucketName;
482486
}
483487

488+
@Before(order = 1, value = "@PUBSUB_SOURCE_TEST")
489+
public static void createSourcePubSubTopic() throws IOException {
490+
pubSubSourceTopic = "cdf-e2e-test-" + UUID.randomUUID();
491+
PubSubClient.createTopic(pubSubSourceTopic);
492+
BeforeActions.scenario.write("Source PubSub topic " + pubSubSourceTopic);
493+
}
494+
495+
@Before(order = 1, value = "@PUBSUB_SCHEMA_TEST")
496+
public static void createSourcePubSubSchema() throws IOException {
497+
pubSubSchemaId = "cdf-e2e-test-" + UUID.randomUUID();
498+
PubSubClient.createAvroSchema(pubSubSchemaId, PluginPropertyUtils.pluginProp("avrofile"));
499+
BeforeActions.scenario.write("Source Schema " + pubSubSchemaId);
500+
}
501+
502+
@Before(order = 2, value = "@PUBSUB_SCHEMA_TOPIC_TEST")
503+
public static void createSourcePubSubSchemaTopic() throws IOException, InterruptedException {
504+
pubSubSourceTopic = "cdf-e2e-test-" + UUID.randomUUID();
505+
PubSubClient.createTopicWithSchema(pubSubSourceTopic, pubSubSchemaId, Encoding.BINARY);
506+
BeforeActions.scenario.write("Schema Topic " + pubSubSourceTopic);
507+
}
508+
509+
@Before(order = 3, value = "@PUBSUB_SUBSCRIPTION_TEST")
510+
public static void createSubscriptionPubSubTopic() throws IOException {
511+
pubSubSourceSubscription = "cdf-e2e-test-" + UUID.randomUUID();
512+
PubSubClient.createSubscription(pubSubSourceSubscription, pubSubSourceTopic);
513+
BeforeActions.scenario.write("Source PubSub subscription " + pubSubSourceSubscription);
514+
}
515+
516+
@After(order = 1, value = "@PUBSUB_SOURCE_TEST")
517+
public static void deleteSourcePubSubTopic() {
518+
try {
519+
PubSubClient.deleteTopic(pubSubSourceTopic);
520+
BeforeActions.scenario.write("Deleted target PubSub topic " + pubSubSourceTopic);
521+
pubSubSourceTopic = StringUtils.EMPTY;
522+
} catch (Exception e) {
523+
if (e.getMessage().contains("Invalid resource name given") || e.getMessage().contains("Resource not found")) {
524+
BeforeActions.scenario.write("Source PubSub topic " + pubSubSourceTopic + " does not exist.");
525+
} else {
526+
Assert.fail(e.getMessage());
527+
}
528+
}
529+
}
530+
531+
@After(order = 2, value = "@PUBSUB_SCHEMA_TOPIC_TEST")
532+
public static void deleteSourcePubSubSchemaTopic() {
533+
try {
534+
PubSubClient.deleteTopic(pubSubSourceTopic);
535+
BeforeActions.scenario.write("Deleted target PubSub topic " + pubSubSourceTopic);
536+
pubSubSourceTopic = StringUtils.EMPTY;
537+
} catch (Exception e) {
538+
if (e.getMessage().contains("Invalid resource name given") || e.getMessage().contains("Resource not found")) {
539+
BeforeActions.scenario.write("Schema " + pubSubSchemaId + " does not exist.");
540+
} else {
541+
Assert.fail(e.getMessage());
542+
}
543+
}
544+
}
545+
484546
@Before(order = 1, value = "@PUBSUB_SINK_TEST")
485547
public static void createTargetPubSubTopic() {
486548
pubSubTargetTopic = "cdf-e2e-test-" + UUID.randomUUID();
487549
BeforeActions.scenario.write("Target PubSub topic " + pubSubTargetTopic);
488550
}
489551

552+
@After(order = 1, value = "@PUBSUB_SCHEMA_TEST")
553+
public static void deletePubSubSchema() throws IOException {
554+
PubSubClient.deleteSchema(PluginPropertyUtils.pluginProp("projectId"), pubSubSchemaId);
555+
BeforeActions.scenario.write("Deleted PubSub schema " + pubSubSchemaId);
556+
}
557+
558+
@After(order = 2, value = "@PUBSUB_SUBSCRIPTION_TEST")
559+
public static void deletePubSubSubscription() throws IOException {
560+
PubSubClient.deleteSubscription(PluginPropertyUtils.pluginProp("projectId"), pubSubSourceSubscription);
561+
BeforeActions.scenario.write("Deleted PubSub subscription " + pubSubSourceSubscription);
562+
}
563+
564+
public static void publishMessageJsonFormat() throws IOException, InterruptedException {
565+
String jsonMessage = PluginPropertyUtils.pluginProp("message");
566+
String jsonMessage2 = PluginPropertyUtils.pluginProp("message2");
567+
List<String> jsonMessagesList = Arrays.asList(jsonMessage, jsonMessage2);
568+
PubSubClient.publishMessagesWithPubsub(PluginPropertyUtils.pluginProp("projectId"),
569+
pubSubSourceTopic, jsonMessagesList);
570+
}
571+
572+
public static void publishMessageAvroFormat() throws IOException, InterruptedException, ExecutionException {
573+
PubSubClient.publishAvroRecords(PluginPropertyUtils.pluginProp("projectId"), pubSubSourceTopic);
574+
}
575+
576+
public static void publishMessage() throws IOException, InterruptedException {
577+
String dataMessage1 = PluginPropertyUtils.pluginProp("firstMessage");
578+
String dataMessage2 = PluginPropertyUtils.pluginProp("secondMessage");
579+
List<String> dataMessagesList = Arrays.asList(dataMessage1, dataMessage2);
580+
PubSubClient.publishMessagesWithPubsub(PluginPropertyUtils.pluginProp
581+
("projectId"), pubSubSourceTopic, dataMessagesList);
582+
}
583+
490584
@After(order = 1, value = "@PUBSUB_SINK_TEST")
491585
public static void deleteTargetPubSubTopic() {
492586
try {

‎src/e2e-test/java/io/cdap/plugin/pubsub/actions/PubSubActions.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import io.cdap.e2e.utils.ElementHelper;
2121
import io.cdap.e2e.utils.SeleniumHelper;
2222
import io.cdap.plugin.pubsub.locators.PubSubLocators;
23+
import org.openqa.selenium.support.ui.Select;
2324

2425
import java.util.UUID;
2526

@@ -85,4 +86,9 @@ public static void enterSubscription(String subscription) {
8586
public static void enterNumberOfReaders(String numberOfReaders) {
8687
ElementHelper.replaceElementValue(PubSubLocators.numberOfReaders, numberOfReaders);
8788
}
89+
90+
public static void selectDataType() {
91+
Select select = new Select(PubSubLocators.messageDataType);
92+
select.selectByIndex(9);
93+
}
8894
}

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,24 @@ public class PubSubLocators {
6565
@FindBy(how = How.XPATH, using = "//input[@data-cy='numberOfReaders']")
6666
public static WebElement numberOfReaders;
6767

68+
@FindBy(how = How.XPATH, using = "//input[@placeholder='Field name' and @value='']")
69+
public static WebElement addField;
70+
@FindBy(how = How.XPATH, using = "//span[contains(text(), \"Batch interval\")]//following-sibling::div//select[1]")
71+
public static WebElement batchTime;
72+
@FindBy(how = How.XPATH, using = "//span[contains(text(), \"Batch interval\")]//following-sibling::div//select[2]")
73+
public static WebElement timeSelect;
74+
@FindBy(how = How.XPATH, using = "//button[@data-testid='config-apply-close']")
75+
public static WebElement saveButton;
76+
@FindBy(how = How.XPATH, using = "//*[@data-cy='pipeline-configure-modeless-btn']")
77+
public static WebElement configButton;
78+
@FindBy(how = How.XPATH, using = "//*[@data-cy='tab-content-Pipeline config']")
79+
public static WebElement pipelineConfig;
80+
@FindBy(how = How.XPATH, using = "//select[@title='bytes']")
81+
public static WebElement messageDataType;
82+
83+
@FindBy(how = How.XPATH, using = "//div[@class='btn pipeline-action-btn pipeline-run-btn']//*[name()='svg']")
84+
public static WebElement clickOnRuntimeArgumentButton;
85+
6886
public static WebElement formatType(String formatType) {
6987
return SeleniumDriver.getDriver()
7088
.findElement(By.xpath("//li[@data-value='" + formatType + "' and text()='" + formatType + "']"));

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

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,16 @@
2727
import io.cdap.plugin.utils.E2EHelper;
2828
import io.cdap.plugin.utils.E2ETestConstants;
2929
import io.cdap.plugin.utils.PubSubClient;
30+
import io.cucumber.java.en.And;
3031
import io.cucumber.java.en.Then;
3132
import io.cucumber.java.en.When;
3233
import org.apache.commons.lang3.StringUtils;
3334
import org.junit.Assert;
35+
import org.openqa.selenium.support.ui.Select;
3436
import stepsdesign.BeforeActions;
3537

3638
import java.io.IOException;
39+
import java.util.concurrent.TimeUnit;
3740

3841
/**
3942
* PubSub Sink Plugin related step design.
@@ -234,4 +237,31 @@ public void enterRuntimeArgumentValueForPubSubSinkPropertyTopicKey(String runtim
234237
public void clickOnPreviewDataForPubSubSink() {
235238
openSinkPluginPreviewData("GooglePublisher");
236239
}
240+
241+
@Then("Subscribe to the Pub/Sub messages")
242+
public void subscribeToTheMessages() throws InterruptedException {
243+
TimeUnit time = TimeUnit.SECONDS;
244+
time.sleep(60);
245+
PubSubClient.subscribeAsync(PluginPropertyUtils.pluginProp(ConstantsUtil.PROJECT_ID),
246+
TestSetupHooks.pubSubSourceSubscription);
247+
}
248+
249+
@And("Click on batch time and select format")
250+
public void clickOnBatchTimeAndSelectFormat() {
251+
Select select = new Select(PubSubLocators.batchTime);
252+
select.selectByIndex(0);
253+
Select selectformat = new Select(PubSubLocators.timeSelect);
254+
selectformat.selectByIndex(1);
255+
ElementHelper.clickOnElement(PubSubLocators.saveButton);
256+
}
257+
258+
@And("Click on configure button")
259+
public void clickOnConfigureButton() {
260+
ElementHelper.clickOnElement(PubSubLocators.configButton);
261+
}
262+
263+
@And("Click on pipeline config")
264+
public void clickOnPipelineConfig() {
265+
ElementHelper.clickOnElement(PubSubLocators.pipelineConfig);
266+
}
237267
}

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

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,21 @@
1616

1717
package io.cdap.plugin.pubsub.stepsdesign;
1818

19+
import io.cdap.e2e.pages.locators.CdfStudioLocators;
1920
import io.cdap.e2e.utils.CdfHelper;
21+
import io.cdap.e2e.utils.ElementHelper;
22+
import io.cdap.e2e.utils.SeleniumDriver;
23+
import io.cdap.plugin.common.stepsdesign.TestSetupHooks;
24+
import io.cdap.plugin.pubsub.actions.PubSubActions;
25+
import io.cdap.plugin.pubsub.locators.PubSubLocators;
2026
import io.cucumber.java.en.Then;
2127
import io.cucumber.java.en.When;
28+
import org.openqa.selenium.Keys;
29+
import org.openqa.selenium.interactions.Actions;
30+
31+
import java.io.IOException;
32+
import java.util.concurrent.ExecutionException;
33+
import java.util.concurrent.TimeUnit;
2234

2335
/**
2436
* PubSub Source Plugin related step design.
@@ -35,4 +47,59 @@ public void sourceIsPubSub() {
3547
public void openThePubSubSourceProperties() {
3648
openSourcePluginProperties("GooglePublisher");
3749
}
50+
51+
@Then("Enter PubSub source property topic name")
52+
public void enterPubSubSourcePropertyTopicName() {
53+
PubSubActions.enterPubSubTopic(TestSetupHooks.pubSubSourceTopic);
54+
}
55+
56+
@Then("Enter PubSub source property subscription name")
57+
public void enterPubSubSourcePropertySubscriptionName() {
58+
PubSubActions.enterSubscription(TestSetupHooks.pubSubSourceSubscription);
59+
}
60+
61+
@Then("Publish the messages")
62+
public void publishTheMessage() throws IOException, InterruptedException {
63+
TimeUnit time = TimeUnit.SECONDS;
64+
time.sleep(60);
65+
TestSetupHooks.publishMessageJsonFormat();
66+
time.sleep(20);
67+
}
68+
69+
@Then("Publish the messages for text format")
70+
public void publishTheMessageTextFormat() throws IOException, InterruptedException, IOException {
71+
TimeUnit time = TimeUnit.SECONDS;
72+
time.sleep(60);
73+
TestSetupHooks.publishMessage();
74+
time.sleep(20);
75+
}
76+
77+
@Then("Enter runtime argument value for PubSub source property topic key {string}")
78+
public void enterRuntimeArgumentValueForPubSubSourcePropertyTopicKey(String runtimeArgumentKey) {
79+
ElementHelper.clickOnElement(PubSubLocators.clickOnRuntimeArgumentButton);
80+
ElementHelper.sendKeys(CdfStudioLocators.runtimeArgsValue(runtimeArgumentKey), TestSetupHooks.pubSubSourceTopic);
81+
}
82+
83+
@Then("Enter runtime argument value for PubSub source property subscription key {string}")
84+
public void enterRuntimeArgumentValueForPubSubSourcePropertySubscriptionKey(String runtimeArgumentKey) {
85+
ElementHelper.sendKeys(CdfStudioLocators.runtimeArgsValue(runtimeArgumentKey),
86+
TestSetupHooks.pubSubSourceSubscription);
87+
}
88+
89+
@Then("Add schema for the message")
90+
public void addSchemaForTheMessageWithOptionValue() {
91+
PubSubActions.selectDataType();
92+
ElementHelper.sendKeys(PubSubLocators.addField, "name");
93+
Actions act = new Actions(SeleniumDriver.getDriver());
94+
act.sendKeys(Keys.ENTER).perform();
95+
ElementHelper.sendKeys(PubSubLocators.addField, "postabbr");
96+
}
97+
98+
@Then("Publish the messages with schema")
99+
public void publishTheMessagesWithSchema() throws InterruptedException, IOException, ExecutionException {
100+
TimeUnit time = TimeUnit.SECONDS;
101+
time.sleep(120);
102+
TestSetupHooks.publishMessageAvroFormat();
103+
time.sleep(30);
104+
}
38105
}

0 commit comments

Comments
 (0)