Skip to content

Commit 28bcdf6

Browse files
yjhawaryjhawar
committed
resolved comments
1 parent 700cf89 commit 28bcdf6

7 files changed

Lines changed: 64 additions & 22 deletions

File tree

‎src/e2e-test/features/Pipeline.feature‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
#
1616

1717
Feature: Oracle - Verify Oracle source data transfer to Big Query
18-
@ORACLE_SOURCE @BIGQUERY_TARGET
18+
@ENV_VARIABLES #@ORACLE_SOURCE @ORACLE_DELETE @BIGQUERY_DELETE
1919
Scenario: To verify replication of snapshot and cdc data from Oracle to Big Query successfully with Sanity test
2020
Given Open DataFusion Project with replication to configure pipeline
2121
When Enter input plugin property: "name" with value: "pipelineName"

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

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import io.cdap.e2e.pages.locators.CdfPipelineRunLocators;
2121
import io.cdap.e2e.utils.*;
2222
import io.cdap.plugin.locators.ReplicationLocators;
23+
import io.cdap.plugin.utils.BigQuery;
2324
import io.cdap.plugin.utils.OracleClient;
2425
import io.cdap.plugin.utils.ValidationHelper;
2526
import org.apache.commons.lang.StringUtils;
@@ -59,7 +60,7 @@ public static void clickOnOraclePlugin() {
5960

6061
public static void selectTable() {
6162
String table = schemaName + "." + tableName;
62-
WaitHelper.waitForElementToBeDisplayed(ReplicationLocators.selectTable(table));
63+
WaitHelper.waitForElementToBeDisplayed(ReplicationLocators.selectTable(table),300);
6364
AssertionHelper.verifyElementDisplayed(ReplicationLocators.selectTable(table));
6465
ElementHelper.clickOnElement(ReplicationLocators.selectTable(table));
6566
}
@@ -113,7 +114,7 @@ public static void waitTillPipelineIsRunningAndCheckForErrors() throws Interrupt
113114
int defaultTimeout = Integer.parseInt(PluginPropertyUtils.pluginProp("pipeline-initialization"));
114115
TimeUnit time = TimeUnit.SECONDS;
115116
time.sleep(defaultTimeout);
116-
ValidationHelper.waitForFlush();
117+
BigQuery.waitForFlush();
117118
// Checking if an error message is displayed.
118119
Assert.assertFalse(ElementHelper.isElementDisplayed(ReplicationLocators.error));
119120
}
@@ -140,18 +141,18 @@ public static void insertRecordAndWait()
140141
throws InterruptedException, SQLException, ClassNotFoundException {
141142
OracleClient.insertRow(tableName, schemaName, datatypeValues);
142143
OracleClient.forceFlushCDC();
143-
ValidationHelper.waitForFlush();
144+
BigQuery.waitForFlush();
144145
}
145146

146147
public static void deleteRecordAndWait() throws SQLException, ClassNotFoundException, InterruptedException {
147148
OracleClient.deleteRow(tableName, schemaName, deleteCondition);
148149
OracleClient.forceFlushCDC();
149-
ValidationHelper.waitForFlush();
150+
BigQuery.waitForFlush();
150151
}
151152

152153
public static void updateRecordAndWait() throws SQLException, ClassNotFoundException, InterruptedException {
153154
OracleClient.updateRow(tableName, schemaName, updateCondition, updatedValue );
154155
OracleClient.forceFlushCDC();
155-
ValidationHelper.waitForFlush();
156+
BigQuery.waitForFlush();
156157
}
157158
}

‎src/e2e-test/java/io.cdap.plugin/hooks/TestSetUpHooks.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import com.google.cloud.bigquery.BigQueryException;
1919
import io.cdap.e2e.utils.BigQueryClient;
2020
import io.cdap.e2e.utils.PluginPropertyUtils;
21+
import io.cdap.plugin.utils.BigQuery;
2122
import io.cdap.plugin.utils.OracleClient;
2223
import io.cucumber.java.After;
2324
import io.cucumber.java.Before;
@@ -43,7 +44,7 @@ public class TestSetUpHooks {
4344
public static String row1 = PluginPropertyUtils.pluginProp("datatypeValuesRow1");
4445
public static String row2= PluginPropertyUtils.pluginProp("datatypeValuesRow2");
4546

46-
@Before(order = 1, value = "@ORACLE_SOURCE")
47+
@Before(order = 1, value = "@ENV_VARIABLES")
4748
public static void overridePropertiesFromEnvVarsIfProvided() {
4849
String projectId = System.getenv("PROJECT_ID");
4950
if (projectId != null && !projectId.isEmpty()) {
@@ -89,12 +90,12 @@ public static void getOracleRecordsAsMap() throws SQLException, ClassNotFoundExc
8990
BeforeActions.scenario.write("Expected Oracle records : " + sourceOracleRecords);
9091
}
9192

92-
@After(order = 1, value = "@ORACLE_SOURCE")
93+
@After(order = 1, value = "@ORACLE_DELETE")
9394
public static void dropTables() throws SQLException, ClassNotFoundException {
9495
OracleClient.deleteTables(schemaName, tableName);
9596
}
9697

97-
@After(order = 1, value = "@BIGQUERY_TARGET")
98+
@After(order = 1, value = "@BIGQUERY_DELETE")
9899
public static void deleteTempTargetBQTable() throws IOException, InterruptedException {
99100
try {
100101
BigQueryClient.dropBqQuery(tableName);

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ public void selectSourceAsOracle() {
4444
}
4545

4646
@Then("Validate Source table is available and select it")
47-
public void selectTable1() {
47+
public void selectTable() {
4848
ReplicationActions.selectTable();
4949
}
5050
@Then("Deploy the replication pipeline")

‎src/e2e-test/java/io.cdap.plugin/utils/BigQuery.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
/*
2+
* Copyright (c) 2023.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package io.cdap.plugin.utils;
18+
19+
import com.google.cloud.bigquery.BigQueryException;
20+
import io.cdap.e2e.utils.BigQueryClient;
21+
import io.cdap.e2e.utils.PluginPropertyUtils;
22+
import org.junit.Assert;
23+
import stepsdesign.BeforeActions;
24+
25+
import java.io.IOException;
26+
import java.util.concurrent.TimeUnit;
27+
28+
public class BigQuery {
29+
public static void waitForFlush() throws InterruptedException {
30+
int flushInterval = Integer.parseInt(PluginPropertyUtils.pluginProp("loadInterval"));
31+
TimeUnit time = TimeUnit.SECONDS;
32+
time.sleep(2 * flushInterval + 60);
33+
}
34+
35+
public static void deleteTable(String tableName) throws IOException, InterruptedException{
36+
try {
37+
BigQueryClient.dropBqQuery(tableName);
38+
BeforeActions.scenario.write("BQ Target table - " + tableName + " deleted successfully");
39+
} catch (BigQueryException e) {
40+
if (e.getMessage().contains("Not found: Table")) {
41+
BeforeActions.scenario.write("BQ Target Table does not exist");
42+
} else {
43+
Assert.fail(e.getMessage());
44+
}
45+
}
46+
}
47+
}

‎src/e2e-test/java/io.cdap.plugin/utils/ValidationHelper.java‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ public static void validateRecords(List<Map<String, Object>> sourceOracleRecords
6666

6767
String uniqueField = PluginPropertyUtils.pluginProp("primaryKey");
6868
// Logic to maintain the order of both lists and validate records based on that order.
69-
Map BqUniqueIdMap = (Map)targetBigQueryRecords.stream()
69+
Map bqUniqueIdMap = (Map)targetBigQueryRecords.stream()
7070
.filter(t -> t.get("_is_deleted")==null)
7171
.collect(Collectors.toMap(
7272
t -> t.get(uniqueField),
@@ -84,11 +84,10 @@ public static void validateRecords(List<Map<String, Object>> sourceOracleRecords
8484
for (int record = 0; record < sourceOracleRecords.size(); record++) {
8585
Map<String, Object> oracleRecord = sourceOracleRecords.get(record);
8686
Object uniqueId = oracleRecord.get(uniqueField);
87-
Map<String, Object> bqRow = (Map<String, Object>) BqUniqueIdMap.get(uniqueId);
87+
Map<String, Object> bqRow = (Map<String, Object>) bqUniqueIdMap.get(uniqueId);
8888
if (bqRow != null) {
8989
bqRow.remove("_is_deleted");
9090
bqRow.remove("_sequence_num");
91-
bqRow.forEach((key, value) -> System.out.println("Bq record" + key + ":" + value));
9291
}
9392
compareRecords(bqRow, oracleRecord);
9493
}
@@ -106,10 +105,4 @@ public static void compareRecords(Map<String, Object> targetBigQueryRecords,
106105
}
107106
}
108107

109-
public static void waitForFlush() throws InterruptedException {
110-
int flushInterval = Integer.parseInt(PluginPropertyUtils.pluginProp("loadInterval"));
111-
TimeUnit time = TimeUnit.SECONDS;
112-
time.sleep(2 * flushInterval);
113-
}
114-
115108
}

‎src/e2e-test/resources/pluginParameters.properties‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
11

22

33
projectId=PROJECT_ID
4-
dataset=XE
5-
schema=ABC
4+
dataset=ORCL
5+
schema=HR
66
sourceTable=SOURCE_TABLE
77
host=ORACLE_HOST
88
port=ORACLE_PORT
99
username=ORACLE_USERNAME
1010
password=ORACLE_PASSWORD
1111

12-
pipelineName=e2e-sanity
12+
pipelineName=e2e-sanity-existing-table
1313
pipeline-initialization=200
1414
loadInterval=30
1515

0 commit comments

Comments
 (0)