content: /*
   * Copyright 2018 Confluent Inc.
   *
   * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations under the License.
   */
  
  package io.confluent.connect.jdbc.dialect;
  
  import org.apache.kafka.connect.data.Date;
  import org.apache.kafka.connect.data.Decimal;
  import org.apache.kafka.connect.data.Schema;
  import org.apache.kafka.connect.data.Schema.Type;
  import org.apache.kafka.connect.data.Time;
  import org.apache.kafka.connect.data.Timestamp;
  import org.junit.Test;
  
  import java.util.List;
  
  import io.confluent.connect.jdbc.util.QuoteMethod;
  import io.confluent.connect.jdbc.util.TableId;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  
  public class Db2DatabaseDialectTest extends BaseDialectTest<Db2DatabaseDialect> {
  
    @Override
    protected Db2DatabaseDialect createDialect() {
      return new Db2DatabaseDialect(sourceConfigWithUrl("jdbc:db2://something"));
    }
  
    @Test
    public void shouldMapPrimitiveSchemaTypeToSqlTypes() {
      assertPrimitiveMapping(Type.INT8, "SMALLINT");
      assertPrimitiveMapping(Type.INT16, "SMALLINT");
      assertPrimitiveMapping(Type.INT32, "INTEGER");
      assertPrimitiveMapping(Type.INT64, "BIGINT");
      assertPrimitiveMapping(Type.FLOAT32, "FLOAT");
      assertPrimitiveMapping(Type.FLOAT64, "DOUBLE");
      assertPrimitiveMapping(Type.BOOLEAN, "SMALLINT");
      assertPrimitiveMapping(Type.BYTES, "BLOB(64000)");
      assertPrimitiveMapping(Type.STRING, "VARCHAR(32672)");
    }
  
    @Test
    public void shouldMapDecimalSchemaTypeToDecimalSqlType() {
      assertDecimalMapping(0, "DECIMAL(31,0)");
      assertDecimalMapping(5, "DECIMAL(31,5)");
    }
  
    @Test
    public void shouldMapDataTypes() {
      verifyDataTypeMapping("SMALLINT", Schema.INT8_SCHEMA);
      verifyDataTypeMapping("SMALLINT", Schema.INT16_SCHEMA);
      verifyDataTypeMapping("INTEGER", Schema.INT32_SCHEMA);
      verifyDataTypeMapping("BIGINT", Schema.INT64_SCHEMA);
      verifyDataTypeMapping("FLOAT", Schema.FLOAT32_SCHEMA);
      verifyDataTypeMapping("DOUBLE", Schema.FLOAT64_SCHEMA);
      verifyDataTypeMapping("SMALLINT", Schema.BOOLEAN_SCHEMA);
      verifyDataTypeMapping("VARCHAR(32672)", Schema.STRING_SCHEMA);
      verifyDataTypeMapping("BLOB(64000)", Schema.BYTES_SCHEMA);
      verifyDataTypeMapping("DECIMAL(31,0)", Decimal.schema(0));
      verifyDataTypeMapping("DECIMAL(31,2)", Decimal.schema(2));
      verifyDataTypeMapping("DATE", Date.SCHEMA);
      verifyDataTypeMapping("TIME", Time.SCHEMA);
      verifyDataTypeMapping("TIMESTAMP", Timestamp.SCHEMA);
    }
  
    @Test
    public void shouldMapDateSchemaTypeToDateSqlType() {
      assertDateMapping("DATE");
    }
  
    @Test
    public void shouldMapTimeSchemaTypeToTimeSqlType() {
      assertTimeMapping("TIME");
    }
  
    @Test
    public void shouldMapTimestampSchemaTypeToTimestampSqlType() {
      assertTimestampMapping("TIMESTAMP");
    }
  
    @Test
    public void shouldBuildCreateQueryStatement() {
      String expected =
          "CREATE TABLE \"myTable\" (\n" + "\"c1\" INTEGER NOT NULL,\n" + "\"c2\" BIGINT NOT NULL,\n"
          + "\"c3\" VARCHAR(32672) NOT NULL,\n" + "\"c4\" VARCHAR(32672) NULL,\n"
          + "\"c5\" DATE DEFAULT '2001-03-15',\n" + "\"c6\" TIME DEFAULT '00:00:00.000',\n"
          + "\"c7\" TIMESTAMP DEFAULT '2001-03-15 00:00:00.000',\n" + "\"c8\" DECIMAL(31,4) NULL,\n"
          + "\"c9\" SMALLINT DEFAULT 1,\n"
          + "PRIMARY KEY(\"c1\"))";
      String sql = dialect.buildCreateTableStatement(tableId, sinkRecordFields);
      assertEquals(expected, sql);
    }
  
    @Test
    public void shouldBuildCreateQueryStatementWithNoIdentifierQuoting() {
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      String expected =
          "CREATE TABLE myTable (\n"
          + "c1 INTEGER NOT NULL,\n"
          + "c2 BIGINT NOT NULL,\n"
          + "c3 VARCHAR(32672) NOT NULL,\n"
          + "c4 VARCHAR(32672) NULL,\n"
          + "c5 DATE DEFAULT '2001-03-15',\n"
          + "c6 TIME DEFAULT '00:00:00.000',\n"
          + "c7 TIMESTAMP DEFAULT '2001-03-15 00:00:00.000',\n"
          + "c8 DECIMAL(31,4) NULL,\n"
          + "c9 SMALLINT DEFAULT 1,\n"
          + "PRIMARY KEY(c1))";
      String sql = dialect.buildCreateTableStatement(tableId, sinkRecordFields);
      assertEquals(expected, sql);
    }
  
    @Test
    public void shouldBuildAlterTableStatement() {
      List<String> statements = dialect.buildAlterTable(tableId, sinkRecordFields);
      String[] sql = {"ALTER TABLE \"myTable\" \n"
                      + "ADD \"c1\" INTEGER NOT NULL,\n"
                      + "ADD \"c2\" BIGINT NOT NULL,\n"
                      + "ADD \"c3\" VARCHAR(32672) NOT NULL,\n"
                      + "ADD \"c4\" VARCHAR(32672) NULL,\n"
                      + "ADD \"c5\" DATE DEFAULT '2001-03-15',\n"
                      + "ADD \"c6\" TIME DEFAULT '00:00:00.000',\n"
                      + "ADD \"c7\" TIMESTAMP DEFAULT '2001-03-15 00:00:00.000',\n"
                      + "ADD \"c8\" DECIMAL(31,4) NULL,\n"
                      + "ADD \"c9\" SMALLINT DEFAULT 1"};
      assertStatements(sql, statements);
    }
  
    @Test
    public void shouldBuildAlterTableStatementWithNoIdentifierQuoting() {
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      List<String> statements = dialect.buildAlterTable(tableId, sinkRecordFields);
      String[] sql = {"ALTER TABLE myTable \n"
                      + "ADD c1 INTEGER NOT NULL,\n"
                      + "ADD c2 BIGINT NOT NULL,\n"
                      + "ADD c3 VARCHAR(32672) NOT NULL,\n"
                      + "ADD c4 VARCHAR(32672) NULL,\n"
                      + "ADD c5 DATE DEFAULT '2001-03-15',\n"
                      + "ADD c6 TIME DEFAULT '00:00:00.000',\n"
                      + "ADD c7 TIMESTAMP DEFAULT '2001-03-15 00:00:00.000',\n"
                      + "ADD c8 DECIMAL(31,4) NULL,\n"
                      + "ADD c9 SMALLINT DEFAULT 1"};
      assertStatements(sql, statements);
    }
  
    @Test
    public void createOneColNoPk() {
      verifyCreateOneColNoPk(
          "CREATE TABLE \"myTable\" (" + System.lineSeparator() + "\"col1\" INTEGER NOT NULL)");
  
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      verifyCreateOneColNoPk(
          "CREATE TABLE myTable (" + System.lineSeparator() + "col1 INTEGER NOT NULL)");
    }
  
    @Test
    public void createOneColOnePk() {
      verifyCreateOneColOnePk(
          "CREATE TABLE \"myTable\" (" + System.lineSeparator() + "\"pk1\" INTEGER NOT NULL," + System
              .lineSeparator() + "PRIMARY KEY(\"pk1\"))");
  
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      verifyCreateOneColOnePk(
          "CREATE TABLE myTable (" + System.lineSeparator() + "pk1 INTEGER NOT NULL," + System
              .lineSeparator() + "PRIMARY KEY(pk1))");
    }
  
    @Test
    public void createThreeColTwoPk() {
      verifyCreateThreeColTwoPk(
          "CREATE TABLE \"myTable\" (" + System.lineSeparator() + "\"pk1\" INTEGER NOT NULL," + System
              .lineSeparator() + "\"pk2\" INTEGER NOT NULL," + System.lineSeparator()
          + "\"col1\" INTEGER NOT NULL," + System.lineSeparator() + "PRIMARY KEY(\"pk1\",\"pk2\"))");
  
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      verifyCreateThreeColTwoPk(
          "CREATE TABLE myTable (" + System.lineSeparator() + "pk1 INTEGER NOT NULL," + System
              .lineSeparator() + "pk2 INTEGER NOT NULL," + System.lineSeparator()
          + "col1 INTEGER NOT NULL," + System.lineSeparator() + "PRIMARY KEY(pk1,pk2))");
    }
  
    @Test
    public void alterAddOneCol() {
      verifyAlterAddOneCol("ALTER TABLE \"myTable\" ADD \"newcol1\" INTEGER NULL");
    }
  
    @Test
    public void alterAddTwoCol() {
      verifyAlterAddTwoCols(
          "ALTER TABLE \"myTable\" " + System.lineSeparator() + "ADD \"newcol1\" INTEGER NULL,"
          + System.lineSeparator() + "ADD \"newcol2\" INTEGER DEFAULT 42");
    }
  
    @Test
    public void insert() {
      TableId customers = tableId("customers");
      String expected = "INSERT INTO \"customers\"(\"age\",\"firstName\",\"lastName\") VALUES(?,?,?)";
      String sql = dialect.buildInsertStatement(customers, columns(customers),
                                                columns(customers, "age", "firstName", "lastName")
      );
      assertEquals(expected, sql);
    }
  
    @Test
    public void update() {
      TableId customers = tableId("customers");
      String expected =
          "UPDATE \"customers\" SET \"age\" = ?, \"firstName\" = ?, \"lastName\" = ? WHERE "
          + "\"id\" = ?";
      String sql = dialect.buildUpdateStatement(customers, columns(customers, "id"),
                                                columns(customers, "age", "firstName", "lastName")
      );
      assertEquals(expected, sql);
  
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      expected =
          "UPDATE customers SET age = ?, firstName = ?, lastName = ? WHERE "
          + "id = ?";
      sql = dialect.buildUpdateStatement(customers, columns(customers, "id"),
          columns(customers, "age", "firstName", "lastName")
      );
      assertEquals(expected, sql);
    }
  
    @Test
    public void shouldBuildUpsertStatement() {
      String expected = "merge into \"myTable\" using (values(?, ?, ?, ?, ?, ?)) "
                        + "as DAT(\"id1\", \"id2\", \"columnA\", \"columnB\", \"columnC\", "
                        + "\"columnD\") "
                        + "on \"myTable\".\"id1\"=DAT.\"id1\" and \"myTable\".\"id2\"=DAT.\"id2\" "
                        + "when matched then update set "
                        + "\"myTable\".\"columnA\"=DAT.\"columnA\", "
                        + "\"myTable\".\"columnB\"=DAT.\"columnB\", "
                        + "\"myTable\".\"columnC\"=DAT.\"columnC\", "
                        + "\"myTable\".\"columnD\"=DAT.\"columnD\" " + "when not matched then "
                        + "insert(\"myTable\".\"columnA\",\"myTable\".\"columnB\",\"myTable\""
                        + ".\"columnC\","
                        + "\"myTable\".\"columnD\",\"myTable\".\"id1\",\"myTable\"" + ".\"id2\") "
                        + "values(DAT.\"columnA\",DAT.\"columnB\",DAT.\"columnC\",DAT.\"columnD\","
                        + "DAT.\"id1\","
                        + "DAT.\"id2\")";
      String sql = dialect.buildUpsertQueryStatement(tableId, pkColumns, columnsAtoD);
      assertEquals(expected, sql);
    }
  
    @Test
    public void shouldBuildUpsertStatementWithNoIdentifierQuoting() {
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      String expected = "merge into myTable using (values(?, ?, ?, ?, ?, ?)) "
                        + "as DAT(id1, id2, columnA, columnB, columnC, "
                        + "columnD) "
                        + "on myTable.id1=DAT.id1 and myTable.id2=DAT.id2 "
                        + "when matched then update set "
                        + "myTable.columnA=DAT.columnA, "
                        + "myTable.columnB=DAT.columnB, "
                        + "myTable.columnC=DAT.columnC, "
                        + "myTable.columnD=DAT.columnD " + "when not matched then "
                        + "insert(myTable.columnA,myTable.columnB,myTable"
                        + ".columnC,"
                        + "myTable.columnD,myTable.id1,myTable.id2) "
                        + "values(DAT.columnA,DAT.columnB,DAT.columnC,DAT.columnD,"
                        + "DAT.id1,"
                        + "DAT.id2)";
      String sql = dialect.buildUpsertQueryStatement(tableId, pkColumns, columnsAtoD);
      assertEquals(expected, sql);
    }
  
    @Test
    public void upsert() {
      TableId actor = tableId("actor");
      String expected = "merge into \"actor\" using (values(?, ?, ?, ?)) as DAT(\"actor_id\", "
                        + "\"first_name\", \"last_name\", \"score\") on \"actor\".\"actor_id\"=DAT"
                        + ".\"actor_id\" when matched then update set \"actor\".\"first_name\"=DAT"
                        + ".\"first_name\", \"actor\".\"last_name\"=DAT.\"last_name\", "
                        + "\"actor\".\"score\"=DAT.\"score\" when not matched then insert(\"actor\""
                        + ".\"first_name\",\"actor\".\"last_name\",\"actor\".\"score\",\"actor\""
                        + ".\"actor_id\") values(DAT.\"first_name\",DAT.\"last_name\",DAT"
                        + ".\"score\",DAT.\"actor_id\")";
      String sql = dialect.buildUpsertQueryStatement(actor, columns(actor, "actor_id"),
                                                     columns(actor, "first_name", "last_name",
                                                             "score"
                                                     )
      );
      assertEquals(expected, sql);
    }
  
    @Test
    public void upsertOnlyKeyCols() {
      TableId actor = tableId("actor");
      String expected = "merge into \"actor\" using (values(?)) as DAT(\"actor_id\") on \"actor\""
                        + ".\"actor_id\"=DAT.\"actor_id\" when not matched then insert(\"actor\""
                        + ".\"actor_id\") values(DAT.\"actor_id\")";
      String sql = dialect.buildUpsertQueryStatement(
          actor, columns(actor, "actor_id"), columns(actor));
      assertEquals(expected, sql);
  
      quoteIdentfiiers = QuoteMethod.NEVER;
      dialect = createDialect();
  
      expected = "merge into actor using (values(?)) as DAT(actor_id) on actor"
                 + ".actor_id=DAT.actor_id when not matched then insert(actor"
                 + ".actor_id) values(DAT.actor_id)";
      sql = dialect.buildUpsertQueryStatement(
          actor, columns(actor, "actor_id"), columns(actor));
      assertEquals(expected, sql);
    }
  
    @Test
    public void shouldSanitizeUrlWithoutCredentialsInProperties() {
      assertSanitizedUrl(
          "jdbc:db2://sysmvs1.stl.ibm.com:5021/STLEC1:user=dbadm;other=dbadm;"
          + "traceLevel=all",
          "jdbc:db2://sysmvs1.stl.ibm.com:5021/STLEC1:user=dbadm;other=dbadm;"
          + "traceLevel=all"
      );
    }
  
    @Test
    public void shouldSanitizeUrlWithCredentialsInUrlProperties() {
      assertSanitizedUrl(
          "jdbc:db2://sysmvs1.stl.ibm.com:5021/STLEC1:user=dbadm;password=dbadm;"
          + "traceLevel=all",
          "jdbc:db2://sysmvs1.stl.ibm.com:5021/STLEC1:user=dbadm;password=****;"
          + "traceLevel=all"
      );
      assertSanitizedUrl(
          "jdbc:db2://sysmvs1.stl.ibm.com:5021/STLEC1:password=dbadm;user=dbadm;"
          + "traceLevel=all",
          "jdbc:db2://sysmvs1.stl.ibm.com:5021/STLEC1:password=****;user=dbadm;"
          + "traceLevel=all"
      );
    }
  
    @Test
    public void testCurrentTimestampDatabaseQuery() {
      assertFalse(dialect.currentTimestampDatabaseQuery().contains(";"));
    }
  
    @Test
    public void testCheckConnectionQuery() {
      assertFalse(dialect.checkConnectionQuery().contains(";"));
    }
  }
  