Reverse replication low priority integration tests (#2310)

* Reverse replication low priority integration tests

* Reverse replication low it (#187)

* Add Low Level IT in CassandraToSourceDbIT

* Added Code Refectored

* Removed Unwanted Files

* Added Missing Schema

* fix session file

* fix tests

* Added CopyRight Fixes

* Adding Static VPC, SUBNET and REGION

* Added Missing params

* Added VPC Changes (#196)

* Added VCP Changes

* workerRegion instead of Region

* Refactor VPC changes

* revert VPC changes

---------

Co-authored-by: pawankashyapollion <[email protected]>
Co-authored-by: taherkl <[email protected]>

Author: 3 people

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

@@ -93,6 +93,7 @@ public class SpannerToCassandraSourceDbIT extends SpannerToSourceDbITBase {
   private static final String ALL_DATA_TYPES_TABLE = "AllDatatypeColumns";
   private static final String ALL_DATA_TYPES_CUSTOM_CONVERSION_TABLE = "AllDatatypeTransformation";
   private static final String BOUNDARY_CONVERSION_TABLE = "BoundaryConversionTestTable";
+  private static final String BOUNDARY_SIZE_TABLE = "testtable_03tpcovf16ed0klxm3v808ch3btgq0uk";
   private static final HashSet<SpannerToCassandraSourceDbIT> testInstances = new HashSet<>();
   private static PipelineLauncher.LaunchInfo jobInfo;
   public static SpannerResourceManager spannerResourceManager;
@@ -164,6 +165,36 @@ public static void cleanUp() throws IOException {
         pubsubResourceManager);
   }
 
+  @Test
+  public void testSpannerToCassandraWithMaxColumnsAndTableName()
+      throws InterruptedException, IOException {
+    assertThatPipeline(jobInfo).isRunning();
+    writeRowWithMaxColumnsNameAndTableInSpanner();
+    assertRowWithMaxColumnsInCassandra();
+  }
+
+  private void writeRowWithMaxColumnsNameAndTableInSpanner() {
+    List<Mutation> mutations = new ArrayList<>();
+    Mutation.WriteBuilder mutationBuilder =
+        Mutation.newInsertOrUpdateBuilder(BOUNDARY_SIZE_TABLE).set("id").to(1);
+    mutationBuilder.set("col_qcbf69rmxtre3b_03tpcovf16ed").to("SampleTestValue");
+
+    mutations.add(mutationBuilder.build());
+    spannerResourceManager.write(mutations);
+    LOG.info("Inserted row into Spanner using Mutations");
+  }
+
+  private void assertRowWithMaxColumnsInCassandra() {
+
+    PipelineOperator.Result result =
+        pipelineOperator()
+            .waitForCondition(
+                createConfig(jobInfo, Duration.ofMinutes(15)),
+                () -> getRowCount(BOUNDARY_SIZE_TABLE) == 1);
+    assertThatResult(result).meetsConditions();
+    LOG.info("Successfully validated columns in Cassandra");
+  }
+
   /**
    * Tests the data flow from Spanner to Cassandra.
    *

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

@@ -29,6 +29,7 @@
 import com.google.pubsub.v1.SubscriptionName;
 import java.io.IOException;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -66,6 +67,8 @@ public class SpannerToSourceDbIT extends SpannerToSourceDbITBase {
   private static final String MYSQL_SCHEMA_FILE_RESOURCE = "SpannerToSourceDbIT/mysql-schema.sql";
 
   private static final String TABLE = "Users";
+  private static final String BOUNDARY_CHECK_TABLE =
+      "testtable_03TpCoVF16ED0KLxM3v808cH3bTGQ0uK_FEXuZHbttvYZPAeGeqiO";
   private static final HashSet<SpannerToSourceDbIT> testInstances = new HashSet<>();
   private static PipelineLauncher.LaunchInfo jobInfo;
   public static SpannerResourceManager spannerResourceManager;
@@ -206,4 +209,36 @@ private void assertRowInMySQL() throws InterruptedException {
     assertThat(rows.get(0).get("name")).isEqualTo("FF");
     assertThat(rows.get(0).get("from")).isEqualTo("AA");
   }
+
+  @Test
+  public void spannerToMySQLSourceDbMaxColAndTableNameTest()
+      throws IOException, InterruptedException {
+    assertThatPipeline(jobInfo).isRunning();
+    // Write row in Spanner
+    writeMaxColRowsInSpanner();
+    // Assert events on Mysql
+    assertBoundaryRowInMySQL();
+  }
+
+  private void writeMaxColRowsInSpanner() {
+    List<Mutation> mutations = new ArrayList<>();
+    Mutation.WriteBuilder mutationBuilder =
+        Mutation.newInsertOrUpdateBuilder(BOUNDARY_CHECK_TABLE).set("id").to(1);
+    mutationBuilder
+        .set("col_qcbF69RmXTRe3B_03TpCoVF16ED0KLxM3v808cH3bTGQ0uK_FEXuZHbttvY")
+        .to("SampleTestValue");
+
+    mutations.add(mutationBuilder.build());
+    spannerResourceManager.write(mutations);
+    LOG.info("Inserted row into Spanner using Mutations");
+  }
+
+  private void assertBoundaryRowInMySQL() throws InterruptedException {
+    PipelineOperator.Result result =
+        pipelineOperator()
+            .waitForCondition(
+                createConfig(jobInfo, Duration.ofMinutes(10)),
+                () -> jdbcResourceManager.getRowCount(BOUNDARY_CHECK_TABLE) == 1);
+    assertThatResult(result).meetsConditions();
+  }
 }

v2/spanner-to-sourcedb/src/test/resources/SpannerToCassandraSourceIT/cassandra-schema.sql

@@ -216,3 +216,9 @@ CREATE TABLE EmptyStringJsonTable (
     duration_list_col LIST<TEXT>,
     frozen_ascii_set_col frozen<SET<ASCII>>
 );
+
+CREATE TABLE testtable_03tpcovf16ed0klxm3v808ch3btgq0uk (
+    id TEXT PRIMARY KEY,
+    col_qcbf69rmxtre3b_03tpcovf16ed TEXT
+);
+

v2/spanner-to-sourcedb/src/test/resources/SpannerToCassandraSourceIT/spanner-schema.sql

@@ -228,6 +228,11 @@ CREATE TABLE IF NOT EXISTS emptystringjsontable (
     frozen_ascii_set_col JSON
 ) PRIMARY KEY(varchar_column);
 
+CREATE TABLE testtable_03tpcovf16ed0klxm3v808ch3btgq0uk (
+    id STRING(100) NOT NULL,
+    col_qcbf69rmxtre3b_03tpcovf16ed STRING(100)
+) PRIMARY KEY (id);
+
 
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (

v2/spanner-to-sourcedb/src/test/resources/SpannerToSourceDbIT/mysql-schema.sql

@@ -2,4 +2,9 @@ CREATE TABLE Users (
     id INT NOT NULL,
     name VARCHAR(25),
     `from` VARCHAR(25),
+ PRIMARY KEY(id));
+
+ CREATE TABLE testtable_03TpCoVF16ED0KLxM3v808cH3bTGQ0uK_FEXuZHbttvYZPAeGeqiO(
+ id INT NOT NULL,
+ col_qcbF69RmXTRe3B_03TpCoVF16ED0KLxM3v808cH3bTGQ0uK_FEXuZHbttvY VARCHAR(25),
  PRIMARY KEY(id));