🪟 🚥 Add tests for verify sync mode options (#16549)

add tests for checking sync mode options

airbyte-webapp-e2e-tests/cypress.json

@@ -12,7 +12,7 @@
   "db": {
     "user": "postgres",
     "host": "localhost",
-    "database": "airbyte_ci",
+    "database": "airbyte_ci_source",
     "password": "secret_password",
     "port":5433
   }

airbyte-webapp-e2e-tests/cypress/commands/connection.ts

@@ -1,5 +1,5 @@
 import { submitButtonClick } from "./common";
-import { createLocalJsonDestination } from "./destination";
+import { createLocalJsonDestination, createPostgresDestination } from "./destination";
 import { createPokeApiSource, createPostgresSource } from "./source";
 import { openAddSource } from "pages/destinationPage";
 import { selectSchedule, setupDestinationNamespaceSourceFormat, enterConnectionName } from "pages/replicationPage";
@@ -12,6 +12,7 @@ export const createTestConnection = (sourceName: string, destinationName: string
     case sourceName.includes("PokeAPI"):
       createPokeApiSource(sourceName, "luxray");
       break;
+
     case sourceName.includes("Postgres"):
       createPostgresSource(sourceName);
       break;
@@ -19,7 +20,17 @@ export const createTestConnection = (sourceName: string, destinationName: string
       createPostgresSource(sourceName);
   }
 
-  createLocalJsonDestination(destinationName, "/local");
+  switch (true) {
+    case destinationName.includes("Postgres"):
+      createPostgresDestination(destinationName);
+      break;
+    case destinationName.includes("JSON"):
+      createLocalJsonDestination(destinationName);
+      break;
+    default:
+      createLocalJsonDestination(destinationName);
+  }
+
   cy.wait(5000);
 
   openAddSource();
@@ -33,5 +44,5 @@ export const createTestConnection = (sourceName: string, destinationName: string
   setupDestinationNamespaceSourceFormat();
   submitButtonClick();
 
-  cy.wait("@createConnection");
+  cy.wait("@createConnection", { requestTimeout: 10000 });
 };

airbyte-webapp-e2e-tests/cypress/commands/connector.ts

@@ -8,6 +8,7 @@ import {
   enterUsername,
   enterPassword,
   enterPokemonName,
+  enterSchema,
 } from "pages/createConnectorPage";
 
 export const fillPostgresForm = (
@@ -16,18 +17,18 @@ export const fillPostgresForm = (
   port: string,
   database: string,
   username: string,
-  password: string
+  password: string,
+  schema: string
 ) => {
   cy.intercept("/api/v1/source_definition_specifications/get").as("getSourceSpecifications");
 
   selectServiceType("Postgres");
 
-  cy.wait("@getSourceSpecifications");
-
   enterName(name);
   enterHost(host);
   enterPort(port);
   enterDatabase(database);
+  enterSchema(schema);
   enterUsername(username);
   enterPassword(password);
 };
@@ -37,8 +38,6 @@ export const fillPokeAPIForm = (name: string, pokeName: string) => {
 
   selectServiceType("PokeAPI");
 
-  cy.wait("@getSourceSpecifications");
-
   enterName(name);
   enterPokemonName(pokeName);
 };

airbyte-webapp-e2e-tests/cypress/commands/destination.ts

@@ -1,8 +1,8 @@
 import { deleteEntity, openSettingForm, submitButtonClick, updateField } from "./common";
-import { fillLocalJsonForm } from "./connector";
+import { fillLocalJsonForm, fillPostgresForm } from "./connector";
 import { goToDestinationPage, openNewDestinationForm } from "pages/destinationPage";
 
-export const createLocalJsonDestination = (name: string, destinationPath: string) => {
+export const createLocalJsonDestination = (name: string, destinationPath: string = "/local") => {
   cy.intercept("/api/v1/scheduler/destinations/check_connection").as("checkDestinationConnection");
   cy.intercept("/api/v1/destinations/create").as("createDestination");
 
@@ -11,8 +11,28 @@ export const createLocalJsonDestination = (name: string, destinationPath: string
   fillLocalJsonForm(name, destinationPath);
   submitButtonClick();
 
-  cy.wait(3000);
-  cy.wait("@checkDestinationConnection");
+  cy.wait("@checkDestinationConnection", { requestTimeout: 8000 });
+  cy.wait("@createDestination");
+};
+
+export const createPostgresDestination = (
+  name: string,
+  host: string = "localhost",
+  port: string = "5434",
+  database: string = "airbyte_ci_destination",
+  username: string = "postgres",
+  password: string = "secret_password",
+  schema: string = ""
+) => {
+  cy.intercept("/api/v1/scheduler/destinations/check_connection").as("checkDestinationConnection");
+  cy.intercept("/api/v1/destinations/create").as("createDestination");
+
+  goToDestinationPage();
+  openNewDestinationForm();
+  fillPostgresForm(name, host, port, database, username, password, schema);
+  submitButtonClick();
+
+  cy.wait("@checkDestinationConnection", { requestTimeout: 8000 });
   cy.wait("@createDestination");
 };
 

airbyte-webapp-e2e-tests/cypress/commands/source.ts

@@ -5,20 +5,21 @@ import { fillPostgresForm, fillPokeAPIForm } from "./connector";
 export const createPostgresSource = (
   name: string,
   host: string = "localhost",
-  port: string = "{selectAll}{del}5433",
-  database: string = "airbyte_ci",
+  port: string = "5433",
+  database: string = "airbyte_ci_source",
   username: string = "postgres",
-  password: string = "secret_password"
+  password: string = "secret_password",
+  schema: string = ""
 ) => {
   cy.intercept("/api/v1/scheduler/sources/check_connection").as("checkSourceUpdateConnection");
   cy.intercept("/api/v1/sources/create").as("createSource");
 
   goToSourcePage();
   openNewSourceForm();
-  fillPostgresForm(name, host, port, database, username, password);
+  fillPostgresForm(name, host, port, database, username, password, schema);
   submitButtonClick();
 
-  cy.wait("@checkSourceUpdateConnection");
+  cy.wait("@checkSourceUpdateConnection", { requestTimeout: 10000 });
   cy.wait("@createSource");
 };
 

airbyte-webapp-e2e-tests/cypress/commands/workspaces.ts

@@ -8,7 +8,7 @@ export const initialSetupCompleted = (completed = true) => {
     });
   });
 
-  cy.on('uncaught:exception', (err, runnable) => {
+  cy.on("uncaught:exception", (err, runnable) => {
     return false;
   });
 };

airbyte-webapp-e2e-tests/cypress/integration/connection.spec.ts

@@ -9,11 +9,19 @@ import {
   fillOutDestinationPrefix,
   goToReplicationTab,
   setupDestinationNamespaceCustomFormat,
-  selectFullAppendSyncMode,
   checkSuccessResult,
   refreshSourceSchemaBtnClick,
   resetModalSaveBtnClick,
   toggleStreamEnabledState,
+  searchStream,
+  selectCursorField,
+  checkCursorField,
+  selectSyncMode,
+  setupDestinationNamespaceDefaultFormat,
+  checkPrimaryKey,
+  isPrimaryKeyNonExist,
+  selectPrimaryKeyField,
+  checkPreFilledPrimaryKeyField
 } from "pages/replicationPage";
 import { openSourceDestinationFromGrid, goToSourcePage } from "pages/sourcePage";
 import { goToSettingsPage } from "pages/settingsConnectionPage";
@@ -27,6 +35,7 @@ import {
   toggleStreamWithChangesAccordion,
 } from "../pages/modals/catalogDiffModal";
 import { updateSchemaModalConfirmBtnClick } from "../pages/modals/updateSchemaModal";
+import { update, ceil } from "cypress/types/lodash";
 
 describe("Connection main actions", () => {
   beforeEach(() => {
@@ -75,6 +84,140 @@ describe("Connection main actions", () => {
     deleteDestination(destName);
   });
 
+  it("Connection sync mode Incremental Append", () => {
+    const sourceName = appendRandomString("Test connection Postgres source cypress");
+    const destName = appendRandomString("Test connection Postgres destination cypress");
+
+    cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
+
+    populateDBSource();
+
+    createTestConnection(sourceName, destName);
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    searchStream("users");
+    selectSyncMode("Incremental", "Append");
+    selectCursorField("col1");
+
+    submitButtonClick();
+    confirmStreamConfigurationChangedPopup();
+
+    cy.wait("@updateConnection", { timeout: 5000 }).then((interception) => {
+      assert.isNotNull(interception.response?.statusCode, "200");
+    });
+
+    checkSuccessResult();
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    searchStream("users");
+    checkCursorField("col1");
+
+    deleteSource(sourceName);
+    deleteDestination(destName);
+    cleanDBSource();
+  });
+
+  it("Connection sync mode Incremental Deduped History - PK is defined", () => {
+    const sourceName = appendRandomString("Test connection Postgres source cypress");
+    const destName = appendRandomString("Test connection Postgres destination cypress");
+
+    cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
+
+    populateDBSource();
+    createTestConnection(sourceName, destName);
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    searchStream("users");
+    selectSyncMode("Incremental", "Deduped + history");
+    selectCursorField("col1");
+    checkPreFilledPrimaryKeyField("id");
+
+    submitButtonClick();
+    confirmStreamConfigurationChangedPopup();
+
+    cy.wait("@updateConnection", { timeout: 5000 }).then((interception) => {
+      assert.isNotNull(interception.response?.statusCode, "200");
+    });
+
+    checkSuccessResult();
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    searchStream("users");
+
+    checkCursorField("col1");
+    checkPreFilledPrimaryKeyField("id");
+
+    deleteSource(sourceName);
+    deleteDestination(destName);
+    cleanDBSource();
+  });
+
+  it("Connection sync mode Incremental Deduped History - PK is not defined", () => {
+    const sourceName = appendRandomString("Test connection Postgres source cypress");
+    const destName = appendRandomString("Test connection Postgres destination cypress");
+
+    cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
+
+    populateDBSource();
+    createTestConnection(sourceName, destName);
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    searchStream("cities");
+    selectSyncMode("Incremental", "Deduped + history");
+    selectCursorField("city");
+    isPrimaryKeyNonExist();
+    selectPrimaryKeyField("city_code");
+
+    submitButtonClick();
+    confirmStreamConfigurationChangedPopup();
+
+    cy.wait("@updateConnection", { timeout: 5000 }).then((interception) => {
+      assert.isNotNull(interception.response?.statusCode, "200");
+    });
+
+    checkSuccessResult();
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    searchStream("cities");
+
+    checkCursorField("city");
+    checkPrimaryKey("city_code");
+
+    deleteSource(sourceName);
+    deleteDestination(destName);
+    cleanDBSource();
+  });
+
   it("Update connection (pokeAPI)", () => {
     cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
 
@@ -85,14 +228,14 @@ describe("Connection main actions", () => {
 
     goToSourcePage();
     openSourceDestinationFromGrid(sourceName);
-    openSourceDestinationFromGrid("Test update connection Local JSON destination cypress");
+    openSourceDestinationFromGrid(destName);
 
     goToReplicationTab();
 
     selectSchedule("Every hour");
     fillOutDestinationPrefix("auto_test");
     setupDestinationNamespaceCustomFormat("_test");
-    selectFullAppendSyncMode();
+    selectSyncMode("Full refresh", "Append");
 
     const prefix = "auto_test";
     fillOutDestinationPrefix(prefix);
@@ -281,4 +424,120 @@ describe("Connection main actions", () => {
     deleteSource(sourceName);
     deleteDestination(destName);
   });
+
+  it("Saving a connection's destination namespace with 'Custom format' option", () => {
+    cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
+
+    const sourceName = appendRandomString("Test update connection PokeAPI source cypress");
+    const destName = appendRandomString("Test update connection Local JSON destination cypress");
+
+    createTestConnection(sourceName, destName);
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    const namespace = "_DestinationNamespaceCustomFormat";
+    setupDestinationNamespaceCustomFormat(namespace);
+
+    // Ensures the DestinationNamespace is applied to the streams
+    assert(cy.get(`[title*="${namespace}"]`));
+
+    submitButtonClick();
+
+    cy.wait("@updateConnection").then((interception) => {
+      assert.isNotNull(interception.response?.statusCode, "200");
+      expect(interception.request.method).to.eq("POST");
+      expect(interception.request)
+        .property("body")
+        .to.contain({
+          name: sourceName + " <> " + destName + "Connection name",
+          namespaceDefinition: "customformat",
+          namespaceFormat: "${SOURCE_NAMESPACE}_DestinationNamespaceCustomFormat",
+          status: "active",
+        });
+
+      const streamToUpdate = interception.request.body.syncCatalog.streams[0];
+
+      expect(streamToUpdate.stream).to.contain({
+        name: "pokemon",
+      });
+    });
+    checkSuccessResult();
+
+    deleteSource(sourceName);
+    deleteDestination(destName);
+  });
+
+  it("Saving a connection's destination namespace with 'Mirror source structure' option", () => {
+    cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
+
+    const sourceName = appendRandomString("Test update connection PokeAPI source cypress");
+    const destName = appendRandomString("Test update connection Local JSON destination cypress");
+
+    createTestConnection(sourceName, destName);
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    const namespace = "<source schema>";
+
+    // Ensures the DestinationNamespace is applied to the streams
+    assert(cy.get(`[title*="${namespace}"]`));
+
+    deleteSource(sourceName);
+    deleteDestination(destName);
+  });
+
+  it("Saving a connection's destination namespace with 'Destination default' option", () => {
+    cy.intercept("/api/v1/web_backend/connections/update").as("updateConnection");
+
+    const sourceName = appendRandomString("Test update connection PokeAPI source cypress");
+    const destName = appendRandomString("Test update connection Local JSON destination cypress");
+
+    createTestConnection(sourceName, destName);
+
+    goToSourcePage();
+    openSourceDestinationFromGrid(sourceName);
+    openSourceDestinationFromGrid(destName);
+
+    goToReplicationTab();
+
+    setupDestinationNamespaceDefaultFormat();
+
+    const namespace = "<destination schema>";
+
+    // Ensures the DestinationNamespace is applied to the streams
+    assert(cy.get(`[title*="${namespace}"]`));
+
+    submitButtonClick();
+
+    cy.wait("@updateConnection").then((interception) => {
+      assert.isNotNull(interception.response?.statusCode, "200");
+      expect(interception.request.method).to.eq("POST");
+      expect(interception.request)
+        .property("body")
+        .to.contain({
+          name: sourceName + " <> " + destName + "Connection name",
+          namespaceDefinition: "destination",
+          namespaceFormat: "${SOURCE_NAMESPACE}",
+          status: "active",
+        });
+
+      const streamToUpdate = interception.request.body.syncCatalog.streams[0];
+
+      expect(streamToUpdate.stream).to.contain({
+        name: "pokemon",
+      });
+    });
+    checkSuccessResult();
+
+    deleteSource(sourceName);
+    deleteDestination(destName);
+  });
 });

airbyte-webapp-e2e-tests/cypress/integration/destination.spec.ts

@@ -16,11 +16,7 @@ describe("Destination main actions", () => {
   it("Update destination", () => {
     const destName = appendRandomString("Test destination cypress for update");
     createLocalJsonDestination(destName, "/local");
-    updateDestination(
-      destName,
-      "connectionConfiguration.destination_path",
-      "/local/my-json"
-    );
+    updateDestination(destName, "connectionConfiguration.destination_path", "/local/my-json");
 
     cy.get("div[data-id='success-result']").should("exist");
     cy.get("input[value='/local/my-json']").should("exist");

airbyte-webapp-e2e-tests/cypress/pages/createConnectorPage.ts

@@ -6,6 +6,7 @@ const databaseInput = "input[name='connectionConfiguration.database']";
 const usernameInput = "input[name='connectionConfiguration.username']";
 const passwordInput = "input[name='connectionConfiguration.password']";
 const pokemonNameInput = "input[name='connectionConfiguration.pokemon_name']";
+const schemaInput = "[data-testid='tag-input'] input";
 const destinationPathInput = "input[name='connectionConfiguration.destination_path']";
 
 export const selectServiceType = (type: string) => {
@@ -22,7 +23,7 @@ export const enterHost = (host: string) => {
 };
 
 export const enterPort = (port: string) => {
-  cy.get(portInput).type(port);
+  cy.get(portInput).type("{selectAll}{del}").type(port);
 };
 
 export const enterDatabase = (database: string) => {
@@ -44,3 +45,17 @@ export const enterPokemonName = (pokeName: string) => {
 export const enterDestinationPath = (destinationPath: string) => {
   cy.get(destinationPathInput).type(destinationPath);
 };
+
+export const enterSchema = (value: string) => {
+  if (!value) {
+    return;
+  }
+  cy.get(schemaInput).first().type(value, { force: true }).type("{enter}", { force: true });
+};
+
+export const removeSchema = (value: string = "Remove public") => {
+  if (!value) {
+    return;
+  }
+  cy.get(`[aria-label*="${value}"]`).click();
+};

airbyte-webapp-e2e-tests/cypress/pages/replicationPage.ts

@@ -4,14 +4,21 @@ const destinationPrefix = "input[data-testid='prefixInput']";
 const replicationTab = "div[data-id='replication-step']";
 const destinationNamespace = "div[data-testid='namespaceDefinition']";
 const destinationNamespaceCustom = "div[data-testid='namespaceDefinition-customformat']";
+const destinationNamespaceDefault = "div[data-testid='namespaceDefinition-destination']";
 const destinationNamespaceSource = "div[data-testid='namespaceDefinition-source']";
 const destinationNamespaceCustomInput = "input[data-testid='input']";
 const syncModeDropdown = "div[data-testid='syncSettingsDropdown'] input";
+const cursorFieldDropdown = "button[class^='PathPopoutButton_button']";
+const cursorFieldText = "[class^='PathPopoutButton_button__']";
+const primaryKeyText = "[class^='PathPopoutButton_button__']";
+const preFilledPrimaryKeyText = "div[class^='PathPopout_text']";
+const primaryKeyDropdown = "button[class^='PathPopoutButton_button']";
 const successResult = "div[data-id='success-result']";
 const saveStreamChangesButton = "button[data-testid='resetModal-save']";
 const connectionNameInput = "input[data-testid='connectionName']";
 const refreshSourceSchemaButton = "button[data-testid='refresh-source-schema-btn']";
 const streamSyncEnabledSwitch = (streamName: string) => `[data-testid='${streamName}-stream-sync-switch']`;
+const streamNameInput = "input[data-testid='input']";
 
 export const goToReplicationTab = () => {
   cy.get(replicationTab).click();
@@ -48,18 +55,51 @@ export const refreshSourceSchemaBtnClick = () => {
   cy.get(refreshSourceSchemaButton).click();
 };
 
-
-
 export const resetModalSaveBtnClick = () => {
   cy.get("[data-testid='resetModal-save']").click();
 };
 
-export const selectFullAppendSyncMode = () => {
+export const setupDestinationNamespaceDefaultFormat = () => {
+  cy.get(destinationNamespace).click();
+  cy.get(destinationNamespaceDefault).click();
+};
+
+export const selectSyncMode = (source: string, dest: string) => {
   cy.get(syncModeDropdown).first().click({ force: true });
 
-  cy.get(`.react-select__menu`)
-    .contains("Append") // it would be nice to select for "Full refresh" is there too
-    .click();
+  cy.get(`.react-select__option`).contains(`Source:${source}|Dest:${dest}`).click();
+};
+
+export const selectCursorField = (value: string) => {
+  cy.get(cursorFieldDropdown).first().click({ force: true });
+
+  cy.get(`.react-select__option`).contains(value).click();
+};
+
+export const checkCursorField = (expectedValue: string) => {
+  cy.get(cursorFieldText).first().contains(expectedValue);
+};
+
+export const checkPrimaryKey = (expectedValue: string) => {
+  cy.get(primaryKeyText).last().contains(expectedValue);
+};
+
+export const checkPreFilledPrimaryKeyField = (expectedValue: string) => {
+  cy.get(preFilledPrimaryKeyText).contains(expectedValue);
+};
+
+export const isPrimaryKeyNonExist = () => {
+  cy.get(preFilledPrimaryKeyText).should("not.exist");
+};
+
+export const selectPrimaryKeyField = (value: string) => {
+  cy.get(primaryKeyDropdown).last().click({ force: true });
+
+  cy.get(`.react-select__option`).contains(value).click();
+};
+
+export const searchStream = (value: string) => {
+  cy.get(streamNameInput).type(value);
 };
 
 export const checkSuccessResult = () => {

airbyte-webapp-e2e-tests/package.json

@@ -6,7 +6,9 @@
     "cypress:open": "cypress open",
     "cypress:ci": "CYPRESS_BASE_URL=http://localhost:8000 cypress run",
     "cypress:ci:record": "CYPRESS_BASE_URL=http://localhost:8000 cypress run --record --key $CYPRESS_KEY",
-    "createdb": "docker run --rm -d -p 5433:5432 -e POSTGRES_PASSWORD=secret_password -e POSTGRES_DB=airbyte_ci --name airbyte_ci_pg postgres"
+    "createdbsource": "docker run --rm -d -p 5433:5432 -e POSTGRES_PASSWORD=secret_password -e POSTGRES_DB=airbyte_ci_source --name airbyte_ci_pg_source postgres",
+    "createdbdestination": "docker run --rm -d -p 5434:5432 -e POSTGRES_PASSWORD=secret_password -e POSTGRES_DB=airbyte_ci_destination --name airbyte_ci_pg_destination postgres"
+
   },
   "eslintConfig": {
     "env": {

tools/bin/e2e_test.sh

@@ -19,7 +19,9 @@ VERSION=dev BASIC_AUTH_USERNAME="" BASIC_AUTH_PASSWORD="" TRACKING_STRATEGY=logg
 # Uncomment for debugging. Warning, this is verbose.
 # trap 'echo "docker-compose logs:" && docker-compose logs -t --tail 1000 && docker-compose down && docker stop airbyte_ci_pg' EXIT
 
-docker run --rm -d -p 5433:5432 -e POSTGRES_PASSWORD=secret_password -e POSTGRES_DB=airbyte_ci --name airbyte_ci_pg postgres
+docker run --rm -d -p 5433:5432 -e POSTGRES_PASSWORD=secret_password -e POSTGRES_DB=airbyte_ci_source --name airbyte_ci_pg_source postgres
+docker run --rm -d -p 5434:5432 -e POSTGRES_PASSWORD=secret_password -e POSTGRES_DB=airbyte_ci_destination --name airbyte_ci_pg_destination postgres
+
 echo "Waiting for health API to be available..."
 # Retry loading the health API of the server to check that the server is fully available
 until $(curl --output /dev/null --fail --silent --max-time 5 --head localhost:8001/api/v1/health); do