Add Oracle Reverse ITs

.github/workflows/spanner-pr.yml

@@ -166,6 +166,8 @@ jobs:
       id: setup-env
       uses: ./.github/actions/setup-env
     - name: Run Integration Smoke Tests
+      env:
+        JAVA_TOOL_OPTIONS: "-Doracle.net.disableOob=true"
       run: |
         ./cicd/run-it-smoke-tests \
         --modules-to-build="SPANNER" \
@@ -209,6 +211,7 @@ jobs:
     - name: Run Integration Tests
       env:
         SPECIFIC_TEST: ${{ github.event.inputs.specific_test }}
+        JAVA_TOOL_OPTIONS: "-Doracle.net.disableOob=true"
       run: |
         ./cicd/run-it-tests \
         --modules-to-build="SPANNER" \
@@ -258,6 +261,8 @@ jobs:
       id: setup-env
       uses: ./.github/actions/setup-env
     - name: Run Load Tests
+      env:
+        JAVA_TOOL_OPTIONS: "-Doracle.net.disableOob=true"
       run: |
         ./cicd/run-load-tests \
         --modules-to-build="SPANNER" \

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbITBase.java

@@ -54,6 +54,9 @@
 public abstract class SpannerToSourceDbITBase extends TemplateTestBase {
 
   private static final Logger LOG = LoggerFactory.getLogger(SpannerToSourceDbITBase.class);
+  protected String testUsername;
+  protected String testUsernameShardA;
+  protected String testUsernameShardB;
 
   protected SpannerResourceManager setUpSpannerResourceManager() {
     return SpannerResourceManager.builder("rr-main-" + testName, PROJECT, REGION)
@@ -166,6 +169,21 @@ private Shard createShardConfig(JDBCResourceManager jdbcResourceManager, String
     shard.setLogicalShardId(shardId);
     shard.setUser(jdbcResourceManager.getUsername());
     shard.setPassword(jdbcResourceManager.getPassword());
+    if (shardId.equals("Shard1") && testUsername != null) {
+      shard.setNamespace(testUsername);
+      shard.setUser(testUsername);
+      shard.setPassword("password");
+    } else if (shardId.equals("shardA") && testUsernameShardA != null) {
+      shard.setNamespace(testUsernameShardA);
+      shard.setUser(testUsernameShardA);
+      shard.setPassword("password");
+    } else if (shardId.equals("shardB") && testUsernameShardB != null) {
+      shard.setNamespace(testUsernameShardB);
+      shard.setUser(testUsernameShardB);
+      shard.setPassword("password");
+    } else if (jdbcResourceManager instanceof org.apache.beam.it.jdbc.OracleResourceManager) {
+      shard.setNamespace(jdbcResourceManager.getUsername().toUpperCase());
+    }
     if (jdbcResourceManager instanceof org.apache.beam.it.jdbc.PostgresResourceManager pgRm) {
       shard.setHost(pgRm.getHost());
       shard.setPort(String.valueOf(pgRm.getPort()));
@@ -174,7 +192,14 @@ private Shard createShardConfig(JDBCResourceManager jdbcResourceManager, String
       shard.setHost(mySqlRm.getHost());
       shard.setPort(String.valueOf(mySqlRm.getPort()));
       shard.setDbName(mySqlRm.getDatabaseName());
+
+    } else if (jdbcResourceManager
+        instanceof org.apache.beam.it.jdbc.OracleResourceManager oracleRm) {
+      shard.setHost(oracleRm.getHost());
+      shard.setPort(String.valueOf(oracleRm.getPort()));
+      shard.setDbName(oracleRm.getDatabaseName());
     } else {
+
       throw new IllegalArgumentException("Unsupported JDBC resource manager type");
     }
     return shard;
@@ -299,7 +324,8 @@ public PipelineLauncher.LaunchInfo launchDataflowJob(
                             && !Objects.equals(
                                 sourceType,
                                 com.google.cloud.teleport.v2.templates.constants.Constants
-                                    .SOURCE_POSTGRESQL))
+                                    .SOURCE_POSTGRESQL)
+                            && !Objects.equals(sourceType, "oracle"))
                         ? "input/cassandra-config.conf"
                         : "input/shard.json",
                     gcsResourceManager));
@@ -561,4 +587,104 @@ protected void createMySQLTableWithNColumns(
       throw new RuntimeException("Error executing DDL statement: " + ddl, e);
     }
   }
+
+  protected static String setupOracleIsolatedUser(
+      org.apache.beam.it.jdbc.JDBCResourceManager jdbcResourceManager) {
+    String username =
+        "REV_"
+            + java.util.UUID.randomUUID().toString().replace("-", "").substring(0, 8).toUpperCase();
+    LOG.info("Creating isolated Oracle user: {}", username);
+    jdbcResourceManager.runSQLUpdate("CREATE USER " + username + " IDENTIFIED BY password");
+    jdbcResourceManager.runSQLUpdate("GRANT ALL PRIVILEGES TO " + username);
+    jdbcResourceManager.runSQLUpdate("GRANT UNLIMITED TABLESPACE TO " + username);
+    jdbcResourceManager.runSQLUpdate("GRANT DBA TO " + username);
+    return username;
+  }
+
+  public static long runIsolatedGetRowCount(
+      org.apache.beam.it.jdbc.JDBCResourceManager manager, String testUsername, String tableName) {
+    String fullTableName = testUsername + "." + tableName;
+    return manager.getRowCount(fullTableName);
+  }
+
+  public static java.util.List<java.util.Map<String, Object>> runIsolatedReadTable(
+      org.apache.beam.it.jdbc.JDBCResourceManager manager, String testUsername, String tableName) {
+    String fullTableName = testUsername + "." + tableName;
+    return manager.readTable(fullTableName);
+  }
+
+  public static java.util.List<java.util.Map<String, Object>> runIsolatedSQLQuery(
+      org.apache.beam.it.jdbc.JDBCResourceManager jdbcResourceManager,
+      String testUsername,
+      String query) {
+    try (java.sql.Connection connection =
+            java.sql.DriverManager.getConnection(
+                jdbcResourceManager.getUri(), testUsername, "password");
+        java.sql.Statement stmt = connection.createStatement()) {
+      if (!"SYSTEM".equalsIgnoreCase(testUsername)
+          && jdbcResourceManager instanceof org.apache.beam.it.jdbc.OracleResourceManager) {
+        stmt.execute("ALTER SESSION SET CURRENT_SCHEMA = " + testUsername);
+      }
+      java.util.List<java.util.Map<String, Object>> result = new java.util.ArrayList<>();
+      try (java.sql.ResultSet rs = stmt.executeQuery(query)) {
+        java.sql.ResultSetMetaData md = rs.getMetaData();
+        int columns = md.getColumnCount();
+        while (rs.next()) {
+          java.util.Map<String, Object> row = new java.util.HashMap<>(columns);
+          for (int i = 1; i <= columns; ++i) {
+            row.put(md.getColumnName(i).toLowerCase(), rs.getObject(i));
+          }
+          result.add(row);
+        }
+      }
+      return result;
+    } catch (Exception e) {
+      throw new RuntimeException("Error running isolated query", e);
+    }
+  }
+
+  protected void createOracleSchema(
+      org.apache.beam.it.jdbc.OracleResourceManager jdbcResourceManager,
+      String mySqlSchemaFile,
+      String targetUsername)
+      throws java.io.IOException {
+    String ddl =
+        String.join(
+            " ",
+            com.google.common.io.Resources.readLines(
+                com.google.common.io.Resources.getResource(mySqlSchemaFile),
+                java.nio.charset.StandardCharsets.UTF_8));
+    ddl = ddl.replaceAll("\r\n", " ").replaceAll("\n", " ");
+    String[] ddls = ddl.split(";");
+    try (java.sql.Connection connection =
+            java.sql.DriverManager.getConnection(
+                jdbcResourceManager.getUri(), targetUsername, "password");
+        java.sql.Statement stmt = connection.createStatement()) {
+      if (!"SYSTEM".equalsIgnoreCase(targetUsername)) {
+        stmt.execute("ALTER SESSION SET CURRENT_SCHEMA = " + targetUsername);
+      }
+      for (String d : ddls) {
+        if (!d.trim().isEmpty() && !d.trim().toUpperCase().startsWith("SELECT")) {
+          try {
+            stmt.executeUpdate(d);
+          } catch (Exception e) {
+            throw new RuntimeException("Failed to execute schema DDL: " + d, e);
+          }
+        }
+      }
+    } catch (Exception e) {
+      throw new RuntimeException("failed creating isolated oracle schema", e);
+    }
+  }
+
+  protected void createOracleTableWithNColumns(
+      org.apache.beam.it.jdbc.OracleResourceManager jdbcResourceManager,
+      String arg1,
+      int arg2,
+      String arg3) {}
+
+  @org.junit.After
+  public void clearIsolatedUser() {
+    testUsername = null;
+  }
 }