DBZ-720 Reworking SnapshotDatatypesIT to test types captured during initial snapshotting

debezium-connector-oracle/src/test/java/io/debezium/connector/oracle/AbstractOracleDatatypesTest.java

@@ -151,16 +151,17 @@ protected static void createTables() throws SQLException {
         }
     }
 
-    List<String> getAllTables() {
+    protected List<String> getAllTables() {
         return Arrays.asList(ALL_TABLES);
     }
 
+    protected abstract boolean insertRecordsDuringTest();
+
     private static void streamTable(String table) throws SQLException {
         connection.execute("GRANT SELECT ON " + table + " to " +  TestHelper.CONNECTOR_USER);
         connection.execute("ALTER TABLE " + table + " ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS");
     }
 
-
     @AfterClass
     public static void closeConnection() throws SQLException {
         if (connection != null) {
@@ -171,8 +172,10 @@ public static void closeConnection() throws SQLException {
     @Test
     public void stringTypes() throws Exception {
         int expectedRecordCount = 0;
-        connection.execute("INSERT INTO debezium.type_string VALUES (1, 'v\u010d2', 'nv\u010d2', 'c', 'n\u010d')");
+
-        connection.execute("COMMIT");
+        if (insertRecordsDuringTest()) {
+            insertStringTypes();
+        }
 
         Testing.debug("Inserted");
         expectedRecordCount++;
@@ -183,7 +186,13 @@ public void stringTypes() throws Exception {
         assertThat(testTableRecords).hasSize(expectedRecordCount);
 
         // insert
+        if (insertRecordsDuringTest()) {
             VerifyRecord.isValidInsert(testTableRecords.get(0));
+        }
+        else {
+            VerifyRecord.isValidRead(testTableRecords.get(0));
+        }
+
         Struct after = (Struct) ((Struct)testTableRecords.get(0).value()).get("after");
         assertRecord(after, EXPECTED_STRING);
     }
@@ -191,8 +200,10 @@ public void stringTypes() throws Exception {
     @Test
     public void fpTypes() throws Exception {
         int expectedRecordCount = 0;
-        connection.execute("INSERT INTO debezium.type_fp VALUES (1, 1.1, 2.22, 3.33, 4.4444, 5.555, 6.66, 77.323)");
+
-        connection.execute("COMMIT");
+        if (insertRecordsDuringTest()) {
+            insertFpTypes();
+        }
 
         Testing.debug("Inserted");
         expectedRecordCount++;
@@ -203,7 +214,13 @@ public void fpTypes() throws Exception {
         assertThat(testTableRecords).hasSize(expectedRecordCount);
 
         // insert
+        if (insertRecordsDuringTest()) {
             VerifyRecord.isValidInsert(testTableRecords.get(0));
+        }
+        else {
+            VerifyRecord.isValidRead(testTableRecords.get(0));
+        }
+
         Struct after = (Struct) ((Struct)testTableRecords.get(0).value()).get("after");
         assertRecord(after, EXPECTED_FP);
     }
@@ -211,8 +228,10 @@ public void fpTypes() throws Exception {
     @Test
     public void intTypes() throws Exception {
         int expectedRecordCount = 0;
-        connection.execute("INSERT INTO debezium.type_int VALUES (1, 1, 22, 333)");
+
-        connection.execute("COMMIT");
+        if (insertRecordsDuringTest()) {
+            insertIntTypes();
+        }
 
         Testing.debug("Inserted");
         expectedRecordCount++;
@@ -223,7 +242,13 @@ public void intTypes() throws Exception {
         assertThat(testTableRecords).hasSize(expectedRecordCount);
 
         // insert
+        if (insertRecordsDuringTest()) {
             VerifyRecord.isValidInsert(testTableRecords.get(0));
+        }
+        else {
+            VerifyRecord.isValidRead(testTableRecords.get(0));
+        }
+
         Struct after = (Struct) ((Struct)testTableRecords.get(0).value()).get("after");
         assertRecord(after, EXPECTED_INT);
     }
@@ -231,16 +256,10 @@ public void intTypes() throws Exception {
     @Test
     public void timeTypes() throws Exception {
         int expectedRecordCount = 0;
-        connection.execute("INSERT INTO debezium.type_time VALUES ("
+
-                + "1"
+        if (insertRecordsDuringTest()) {
-                + ", TO_DATE('27-MAR-2018', 'dd-MON-yyyy')"
+            insertTimeTypes();
-                + ", TO_TIMESTAMP('27-MAR-2018 12:34:56.00789', 'dd-MON-yyyy HH24:MI:SS.FF5')"
+        }
-                + ", TO_TIMESTAMP_TZ('27-MAR-2018 01:34:56.00789 -11:00', 'dd-MON-yyyy HH24:MI:SS.FF5 TZH:TZM')"
-                + ", TO_TIMESTAMP_TZ('27-MAR-2018 01:34:56.00789', 'dd-MON-yyyy HH24:MI:SS.FF5')"
-                + ", INTERVAL '-3-6' YEAR TO MONTH"
-                + ", INTERVAL '-1 2:3:4.56' DAY TO SECOND"
-                + ")");
-        connection.execute("COMMIT");
 
         Testing.debug("Inserted");
         expectedRecordCount++;
@@ -251,11 +270,45 @@ public void timeTypes() throws Exception {
         assertThat(testTableRecords).hasSize(expectedRecordCount);
 
         // insert
+        if (insertRecordsDuringTest()) {
             VerifyRecord.isValidInsert(testTableRecords.get(0));
+        }
+        else {
+            VerifyRecord.isValidRead(testTableRecords.get(0));
+        }
+
         Struct after = (Struct) ((Struct)testTableRecords.get(0).value()).get("after");
         assertRecord(after, EXPECTED_TIME);
     }
 
+    protected static void insertStringTypes() throws SQLException {
+        connection.execute("INSERT INTO debezium.type_string VALUES (1, 'v\u010d2', 'nv\u010d2', 'c', 'n\u010d')");
+        connection.execute("COMMIT");
+    }
+
+    protected static void insertFpTypes() throws SQLException {
+        connection.execute("INSERT INTO debezium.type_fp VALUES (1, 1.1, 2.22, 3.33, 4.4444, 5.555, 6.66, 77.323)");
+        connection.execute("COMMIT");
+    }
+
+    protected static void insertIntTypes() throws SQLException {
+        connection.execute("INSERT INTO debezium.type_int VALUES (1, 1, 22, 333)");
+        connection.execute("COMMIT");
+    }
+
+    protected static void insertTimeTypes() throws SQLException {
+        connection.execute("INSERT INTO debezium.type_time VALUES ("
+                + "1"
+                + ", TO_DATE('27-MAR-2018', 'dd-MON-yyyy')"
+                + ", TO_TIMESTAMP('27-MAR-2018 12:34:56.00789', 'dd-MON-yyyy HH24:MI:SS.FF5')"
+                + ", TO_TIMESTAMP_TZ('27-MAR-2018 01:34:56.00789 -11:00', 'dd-MON-yyyy HH24:MI:SS.FF5 TZH:TZM')"
+                + ", TO_TIMESTAMP_TZ('27-MAR-2018 01:34:56.00789', 'dd-MON-yyyy HH24:MI:SS.FF5')"
+                + ", INTERVAL '-3-6' YEAR TO MONTH"
+                + ", INTERVAL '-1 2:3:4.56' DAY TO SECOND"
+                + ")");
+        connection.execute("COMMIT");
+    }
+
     private void assertRecord(Struct record, List<SchemaAndValueField> expected) {
         expected.forEach(schemaAndValueField -> schemaAndValueField.assertFor(record));
     }

debezium-connector-oracle/src/test/java/io/debezium/connector/oracle/SnapshotDatatypesIT.java

@@ -7,26 +7,33 @@
 
 import java.sql.SQLException;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.rules.TestName;
 
 import io.debezium.config.Configuration;
-import io.debezium.connector.oracle.OracleConnectorConfig.SnapshotMode;
 import io.debezium.connector.oracle.util.TestHelper;
 import io.debezium.util.Testing;
 
 /**
- * Integration test to verify different Oracle datatypes.
+ * Integration test to verify different Oracle datatypes as captured during initial snapshotting.
  *
  * @author Jiri Pechanec
  */
 public class SnapshotDatatypesIT extends AbstractOracleDatatypesTest {
 
+    @Rule public TestName name = new TestName();
+
     @BeforeClass
     public static void beforeClass() throws SQLException {
         createTables();
+
+        insertStringTypes();
+        insertFpTypes();
+        insertIntTypes();
+        insertTimeTypes();
     }
 
     @Before
@@ -36,14 +43,8 @@ public void before() throws Exception {
         Testing.Debug.enable();
         Testing.Files.delete(TestHelper.DB_HISTORY_PATH);
 
-        String whitelistedTables = getAllTables().stream()
-                .map(t -> "ORCLPDB1." + t)
-                .map(t -> t.replaceAll("\\.", "\\\\."))
-                .collect(Collectors.joining(","));
-
         Configuration config = TestHelper.defaultConfig()
-                .with(OracleConnectorConfig.TABLE_WHITELIST, whitelistedTables)
+                .with(OracleConnectorConfig.TABLE_WHITELIST, getTableWhitelist())
-                .with(OracleConnectorConfig.SNAPSHOT_MODE, SnapshotMode.INITIAL_SCHEMA_ONLY)
                 .build();
 
         start(OracleConnector.class, config);
@@ -51,4 +52,24 @@ public void before() throws Exception {
 
         Thread.sleep(2000);
     }
+
+    private String getTableWhitelist() {
+        switch(name.getMethodName()) {
+            case "stringTypes":
+                return "ORCLPDB1.debezium.type_string";
+            case "fpTypes":
+                return "ORCLPDB1.debezium.type_fp";
+            case "intTypes":
+                return "ORCLPDB1.debezium.type_int";
+            case "timeTypes":
+                return "ORCLPDB1.debezium.type_time";
+            default:
+                throw new IllegalArgumentException("Unexpected test method: " + name.getMethodName());
+        }
+    }
+
+    @Override
+    protected boolean insertRecordsDuringTest() {
+        return false;
+    }
 }

debezium-connector-oracle/src/test/java/io/debezium/connector/oracle/StreamingDatatypesIT.java

@@ -16,7 +16,7 @@
 import io.debezium.util.Testing;
 
 /**
- * Integration test to verify different Oracle datatypes.
+ * Integration test to verify different Oracle datatypes as captured during streaming.
  *
  * @author Jiri Pechanec
  */
@@ -47,4 +47,9 @@ public void before() throws Exception {
         Thread.sleep(2000);
         createTables();
     }
+
+    @Override
+    protected boolean insertRecordsDuringTest() {
+        return true;
+    }
 }