PLUGIN-72 SAP HANA Plugin integration tests

docker-compose/db-plugins-env/docker-compose.yml

@@ -80,7 +80,6 @@ services:
     sysctls:
       - kernel.shmmax=1073741824
       - net.ipv4.ip_local_port_range=60000 65535
-      - kernel.shmmni=524288
       - kernel.shmall=8388608
     extra_hosts:
       # Alter this if running on non-Linux machine

saphana-plugin/pom.xml

@@ -51,6 +51,12 @@
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.sap.cloud.db.jdbc</groupId>
+      <artifactId>ngdbc</artifactId>
+      <version>2.3.48</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>hydrator-test</artifactId>

saphana-plugin/src/main/java/io/cdap/plugin/saphana/SapHanaPostAction.java

@@ -31,8 +31,11 @@
 @Description("Runs a SAP HANA query after a pipeline run.")
 public class SapHanaPostAction extends AbstractQueryAction {
 
+  private final SapHanaQueryActionConfig sapHanaQueryActionConfig;
+
   public SapHanaPostAction(SapHanaQueryActionConfig sapHanaQueryActionConfig) {
     super(sapHanaQueryActionConfig, false);
+    this.sapHanaQueryActionConfig = sapHanaQueryActionConfig;
   }
 
   /**

saphana-plugin/src/test/java/io/cdap/plugin/saphana/SapHanaActionTestRun.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright © 2019 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.
+ */
+
+package io.cdap.plugin.saphana;
+
+import com.google.common.collect.ImmutableMap;
+import io.cdap.cdap.etl.api.action.Action;
+import io.cdap.cdap.etl.mock.batch.MockSink;
+import io.cdap.cdap.etl.mock.batch.MockSource;
+import io.cdap.cdap.etl.proto.v2.ETLBatchConfig;
+import io.cdap.cdap.etl.proto.v2.ETLPlugin;
+import io.cdap.cdap.etl.proto.v2.ETLStage;
+import io.cdap.cdap.proto.artifact.AppRequest;
+import io.cdap.cdap.proto.id.ApplicationId;
+import io.cdap.cdap.proto.id.NamespaceId;
+import io.cdap.cdap.test.ApplicationManager;
+import io.cdap.plugin.db.batch.action.QueryConfig;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+public class SapHanaActionTestRun extends SapHanaPluginTestBase {
+
+    protected static final String ACTION_TEST_TABLE_NAME = "DB_ACTION_TEST";
+
+
+    @Before
+    public void createTables() throws Exception {
+        Connection conn = createConnection();
+        Statement statement = conn.createStatement();
+        statement.execute("CREATE TABLE " + ACTION_TEST_TABLE_NAME + "(ID INT);");
+        statement.execute("INSERT INTO " + ACTION_TEST_TABLE_NAME + " VALUES(1)");
+        statement.execute("INSERT INTO " + ACTION_TEST_TABLE_NAME + " VALUES(2)");
+    }
+
+
+    @Test
+    public void testDBAction() throws Exception {
+        ETLStage source = new ETLStage("source", MockSource.getPlugin("actionInput"));
+        ETLStage sink = new ETLStage("sink", MockSink.getPlugin("actionOutput"));
+        ETLStage action = new ETLStage("action", new ETLPlugin(
+                SapHanaConstants.PLUGIN_NAME,
+                Action.PLUGIN_TYPE,
+                ImmutableMap.<String, String>builder()
+                        .putAll(BASE_PROPS)
+                        .put(QueryConfig.QUERY, "DELETE from " + ACTION_TEST_TABLE_NAME + " where ID=1")
+                        .build(),
+                null));
+        ETLBatchConfig config = ETLBatchConfig.builder()
+                .addStage(source)
+                .addStage(sink)
+                .addStage(action)
+                .addConnection(sink.getName(), action.getName())
+                .addConnection(source.getName(), sink.getName())
+                .build();
+        AppRequest<ETLBatchConfig> appRequest = new AppRequest<>(DATAPIPELINE_ARTIFACT, config);
+        ApplicationId appId = NamespaceId.DEFAULT.app("actionTest");
+        ApplicationManager appManager = deployApplication(appId, appRequest);
+        runETLOnce(appManager, ImmutableMap.of("logical.start.time", "0"));
+
+        Connection connection = createConnection();
+        Statement statement = connection.createStatement();
+
+        ResultSet results = statement.executeQuery("select * from " + ACTION_TEST_TABLE_NAME);
+        results.next();
+        int id = results.getInt("ID");
+        Assert.assertEquals(id, 2);
+        Assert.assertNotEquals(id, 1);
+        Assert.assertFalse(results.next());
+
+    }
+
+    @After
+    public void dropTables() throws Exception {
+        Connection conn = createConnection();
+        Statement statement = conn.createStatement();
+        statement.execute("DROP TABLE " + ACTION_TEST_TABLE_NAME + ";");
+    }
+
+}

saphana-plugin/src/test/java/io/cdap/plugin/saphana/SapHanaPluginTestBase.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright © 2019 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.
+ */
+
+package io.cdap.plugin.saphana;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
+import com.sap.db.jdbc.Driver;
+import io.cdap.cdap.api.artifact.ArtifactSummary;
+import io.cdap.cdap.api.plugin.PluginClass;
+import io.cdap.cdap.datapipeline.DataPipelineApp;
+import io.cdap.cdap.proto.id.ArtifactId;
+import io.cdap.cdap.proto.id.NamespaceId;
+import io.cdap.plugin.db.ConnectionConfig;
+import io.cdap.plugin.db.DBRecord;
+import io.cdap.plugin.db.batch.DatabasePluginTestBase;
+import io.cdap.plugin.db.batch.sink.ETLDBOutputFormat;
+import io.cdap.plugin.db.batch.source.DataDrivenETLDBInputFormat;
+import org.junit.BeforeClass;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Collections;
+import java.util.Map;
+
+
+public abstract class SapHanaPluginTestBase extends DatabasePluginTestBase {
+
+
+    protected static final String JDBC_DRIVER_NAME = "sap";
+    protected static final ArtifactId DATAPIPELINE_ARTIFACT_ID = NamespaceId.DEFAULT.artifact("data-pipeline", "3.2.0");
+    protected static final ArtifactSummary DATAPIPELINE_ARTIFACT = new ArtifactSummary("data-pipeline", "3.2.0");
+    protected static final Map<String, String> BASE_PROPS = ImmutableMap.<String, String>builder()
+            .put(ConnectionConfig.HOST, System.getProperty("saphana.host", "localhost"))
+            .put(ConnectionConfig.PORT, System.getProperty("saphana.port", "39017"))
+            .put(ConnectionConfig.DATABASE, System.getProperty("sapahana.database", "SYSTEMDB"))
+            .put(ConnectionConfig.USER, System.getProperty("sapahana.user", "SYSTEM"))
+            .put(ConnectionConfig.PASSWORD, System.getProperty("sapahana.password", "SAPhxe123"))
+            .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME)
+            .build();
+    private static String connectionUrl;
+    private static Boolean setupCompleted = false;
+
+
+    @BeforeClass
+    public static void setupTest() throws Exception {
+        if (setupCompleted) {
+            return;
+        }
+        System.out.println("Setting up batch artifacts");
+        setupBatchArtifacts(DATAPIPELINE_ARTIFACT_ID, DataPipelineApp.class);
+        System.out.println("Adding plugin artifact");
+        addPluginArtifact(NamespaceId.DEFAULT.artifact(SapHanaConstants.PLUGIN_NAME, "1.0.0"),
+                DATAPIPELINE_ARTIFACT_ID, SapHanaSource.class, DBRecord.class, ETLDBOutputFormat.class,
+                DataDrivenETLDBInputFormat.class, SapHanaAction.class, SapHanaPostAction.class);
+
+        PluginClass sapHanaDriver = new PluginClass(ConnectionConfig.JDBC_PLUGIN_TYPE, JDBC_DRIVER_NAME,
+                "SapHana driver class", Driver.class.getName(), null, Collections.emptyMap());
+        addPluginArtifact(NamespaceId.DEFAULT.artifact("saphana-jdbc-connector", "1.0.0"), DATAPIPELINE_ARTIFACT_ID,
+                Sets.newHashSet(sapHanaDriver), Driver.class);
+
+        connectionUrl = "jdbc:sap://" + BASE_PROPS.get(ConnectionConfig.HOST) + ":" +
+                BASE_PROPS.get(ConnectionConfig.PORT) + "/" + BASE_PROPS.get(ConnectionConfig.DATABASE);
+        setupCompleted = true;
+    }
+
+
+    protected static Connection createConnection() throws Exception {
+        Class.forName(Driver.class.getCanonicalName());
+        return DriverManager.getConnection(connectionUrl, BASE_PROPS.get(ConnectionConfig.USER),
+                BASE_PROPS.get(ConnectionConfig.PASSWORD));
+    }
+
+
+}
+

saphana-plugin/src/test/java/io/cdap/plugin/saphana/SapHanaPluginTestSuite.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright © 2019 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.
+ */
+
+package io.cdap.plugin.saphana;
+
+import io.cdap.cdap.common.test.TestSuite;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+/**
+ * This is a test suite that runs all the tests for Database plugins.
+ */
+@RunWith(TestSuite.class)
+@Suite.SuiteClasses({
+        SapHanaActionTestRun.class,
+        SapHanaSourceTestRun.class,
+        SapHanaSinkTestRun.class,
+        SapHanaPostActionTestRun.class
+})
+public class SapHanaPluginTestSuite extends SapHanaPluginTestBase {
+
+}

saphana-plugin/src/test/java/io/cdap/plugin/saphana/SapHanaPostActionTestRun.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright © 2019 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.
+ */
+
+package io.cdap.plugin.saphana;
+
+import com.google.common.collect.ImmutableMap;
+import io.cdap.cdap.etl.api.batch.PostAction;
+import io.cdap.cdap.etl.mock.batch.MockSink;
+import io.cdap.cdap.etl.mock.batch.MockSource;
+import io.cdap.cdap.etl.proto.v2.ETLBatchConfig;
+import io.cdap.cdap.etl.proto.v2.ETLPlugin;
+import io.cdap.cdap.etl.proto.v2.ETLStage;
+import io.cdap.cdap.proto.artifact.AppRequest;
+import io.cdap.cdap.proto.id.ApplicationId;
+import io.cdap.cdap.proto.id.NamespaceId;
+import io.cdap.cdap.test.ApplicationManager;
+import io.cdap.plugin.common.batch.action.Condition;
+import io.cdap.plugin.db.batch.action.QueryActionConfig;
+import io.cdap.plugin.db.batch.action.QueryConfig;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+public class SapHanaPostActionTestRun extends SapHanaPluginTestBase {
+
+    protected static final String POST_ACTION_TEST_TABLE_NAME = "POST_ACTION_TEST";
+
+    @Before
+    public void createTables() throws Exception {
+        Connection conn = createConnection();
+        Statement statement = conn.createStatement();
+        statement.execute("CREATE TABLE " + POST_ACTION_TEST_TABLE_NAME + "(ID INT);");
+        statement.execute("INSERT INTO " + POST_ACTION_TEST_TABLE_NAME + " VALUES(1)");
+        statement.execute("INSERT INTO " + POST_ACTION_TEST_TABLE_NAME + " VALUES(2)");
+    }
+
+
+    @Test
+    public void testDBPostAction() throws Exception {
+        ETLStage source = new ETLStage("source", MockSource.getPlugin("postActionInput"));
+        ETLStage sink = new ETLStage("sink", MockSink.getPlugin("postActionOutput"));
+        ETLStage action = new ETLStage("postAction", new ETLPlugin(
+                SapHanaConstants.PLUGIN_NAME,
+                PostAction.PLUGIN_TYPE,
+                ImmutableMap.<String, String>builder()
+                        .putAll(BASE_PROPS)
+                        .put(QueryConfig.QUERY, "DELETE from " + POST_ACTION_TEST_TABLE_NAME + " where ID=1")
+                        .put(QueryActionConfig.RUN_CONDITION, Condition.SUCCESS.name())
+                        .build(),
+                null));
+
+        ETLBatchConfig config = ETLBatchConfig.builder()
+                .addStage(source)
+                .addStage(sink)
+                .addPostAction(action)
+                .addConnection(source.getName(), sink.getName())
+                .build();
+
+
+        AppRequest<ETLBatchConfig> appRequest = new AppRequest<>(DATAPIPELINE_ARTIFACT, config);
+        ApplicationId appId = NamespaceId.DEFAULT.app("postActionTest");
+        ApplicationManager appManager = deployApplication(appId, appRequest);
+        runETLOnce(appManager, ImmutableMap.of("logical.start.time", "0"));
+
+        Connection connection = createConnection();
+        Statement statement = connection.createStatement();
+
+        ResultSet results = statement.executeQuery("select * from " + POST_ACTION_TEST_TABLE_NAME);
+        results.next();
+        int id = results.getInt("ID");
+        Assert.assertEquals(id, 2);
+        Assert.assertFalse(results.next());
+        connection.close();
+    }
+
+    @After
+    public void dropTables() throws Exception {
+        Connection conn = createConnection();
+        Statement statement = conn.createStatement();
+        statement.execute("DROP TABLE " + POST_ACTION_TEST_TABLE_NAME + ";");
+    }
+}