Add namespace test in destination acceptance test

airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/DataArgumentsProvider.java

@@ -20,6 +20,8 @@ public class DataArgumentsProvider implements ArgumentsProvider {
       new CatalogMessageTestConfigPair("exchange_rate_catalog.json", "exchange_rate_messages.txt");
   public static final CatalogMessageTestConfigPair EDGE_CASE_CONFIG =
       new CatalogMessageTestConfigPair("edge_case_catalog.json", "edge_case_messages.txt");
+  public static final CatalogMessageTestConfigPair NAMESPACE_CONFIG =
+      new CatalogMessageTestConfigPair("namespace_catalog.json", "namespace_messages.txt");
 
   @Override
   public Stream<? extends Arguments> provideArguments(final ExtensionContext context) {

airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/DestinationAcceptanceTest.java

@@ -19,6 +19,7 @@
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.lang.Exceptions;
 import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.commons.util.MoreIterators;
 import io.airbyte.commons.util.MoreLists;
 import io.airbyte.config.EnvConfigs;
 import io.airbyte.config.JobGetSpecConfig;
@@ -27,6 +28,7 @@
 import io.airbyte.config.StandardCheckConnectionOutput;
 import io.airbyte.config.StandardCheckConnectionOutput.Status;
 import io.airbyte.config.WorkerDestinationConfig;
+import io.airbyte.integrations.destination.NamingConventionTransformer;
 import io.airbyte.protocol.models.AirbyteCatalog;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.AirbyteMessage.Type;
@@ -64,16 +66,21 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
 import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.joda.time.DateTime;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
 import org.junit.jupiter.params.provider.ArgumentsSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -915,6 +922,55 @@ void testSyncWriteSameTableNameDifferentNamespace() throws Exception {
     retrieveRawRecordsAndAssertSameMessages(catalog, allMessages, defaultSchema);
   }
 
+  public static class NamespaceTestCaseProvider implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(final ExtensionContext context) throws Exception {
+      final JsonNode testCases =
+          Jsons.deserialize(MoreResources.readResource("namespace_test_cases.json"));
+      return MoreIterators.toList(testCases.elements()).stream()
+          .filter(testCase -> testCase.get("enabled").asBoolean())
+          .map(testCase -> Arguments.of(
+              testCase.get("id").asText(),
+              testCase.get("namespace").asText(),
+              testCase.get("normalized").asText()));
+    }
+
+  }
+
+  @ParameterizedTest
+  @ArgumentsSource(NamespaceTestCaseProvider.class)
+  public void testNamespaces(final String testCaseId, final String namespace, final String normalizedNamespace) throws Exception {
+    final Optional<NamingConventionTransformer> nameTransformer = getNameTransformer();
+    nameTransformer.ifPresent(namingConventionTransformer -> assertNamespaceNormalization(testCaseId, normalizedNamespace,
+        namingConventionTransformer.getIdentifier(namespace)));
+
+    if (!implementsNamespaces() || !supportNamespaceTest()) {
+      return;
+    }
+
+    final AirbyteCatalog catalog = Jsons.deserialize(
+        MoreResources.readResource(DataArgumentsProvider.NAMESPACE_CONFIG.catalogFile), AirbyteCatalog.class);
+    catalog.getStreams().forEach(stream -> stream.setNamespace(namespace));
+    final ConfiguredAirbyteCatalog configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog);
+
+    final List<AirbyteMessage> messages = MoreResources.readResource(DataArgumentsProvider.NAMESPACE_CONFIG.messageFile).lines()
+        .map(record -> Jsons.deserialize(record, AirbyteMessage.class)).collect(Collectors.toList());
+    messages.forEach(
+        message -> {
+          if (message.getRecord() != null) {
+            message.getRecord().setNamespace(namespace);
+          }
+        });
+
+    final JsonNode config = getConfig();
+    try {
+      runSyncAndVerifyStateOutput(config, messages, configuredCatalog, false);
+    } catch (final Exception e) {
+      throw new IOException(String.format("[%s] Destination failed to sync data to namespace %s", testCaseId, namespace), e);
+    }
+  }
+
   /**
    * In order to launch a source on Kubernetes in a pod, we need to be able to wrap the entrypoint.
    * The source connector must specify its entrypoint in the AIRBYTE_ENTRYPOINT variable. This test
@@ -933,6 +989,32 @@ public void testEntrypointEnvVar() throws Exception {
     assertFalse(entrypoint.isBlank());
   }
 
+  /**
+   * Whether the destination should be tested against different namespaces.
+   */
+  protected boolean supportNamespaceTest() {
+    return false;
+  }
+
+  /**
+   * Set up the name transformer used by a destination to test it against a variety of namespaces.
+   */
+  protected Optional<NamingConventionTransformer> getNameTransformer() {
+    return Optional.empty();
+  }
+
+  /**
+   * Override this method if the normalized namespace is different from the default one. E.g. S3 does
+   * not allow a name starting with a number. So it should change the expected normalized namespace
+   * when testCaseId = "s3a-1". Find the testCaseId in "namespace_test_cases.json".
+   */
+  protected void assertNamespaceNormalization(final String testCaseId,
+                                              final String expectedNormalizedNamespace,
+                                              final String actualNormalizedNamespace) {
+    assertEquals(expectedNormalizedNamespace, actualNormalizedNamespace,
+        String.format("Test case %s failed; if this is expected, please override assertNamespaceNormalization", testCaseId));
+  }
+
   private ConnectorSpecification runSpec() throws WorkerException {
     return new DefaultGetSpecWorker(
         workerConfigs, new AirbyteIntegrationLauncher(JOB_ID, JOB_ATTEMPT, getImageName(), processFactory, null))

airbyte-integrations/bases/standard-destination-test/src/main/resources/namespace_catalog.json

@@ -0,0 +1,14 @@
+{
+  "streams": [
+    {
+      "name": "data_stream",
+      "json_schema": {
+        "properties": {
+          "field1": {
+            "type": "boolean"
+          }
+        }
+      }
+    }
+  ]
+}

airbyte-integrations/bases/standard-destination-test/src/main/resources/namespace_messages.txt

@@ -0,0 +1 @@
+{"type": "RECORD", "record": {"stream": "data_stream", "emitted_at": 1602637589000, "data": { "field1" : true }}}

airbyte-integrations/bases/standard-destination-test/src/main/resources/namespace_test_cases.json

@@ -0,0 +1,51 @@
+[
+  {
+    "id": "s1-1",
+    "description": "namespace are converted to lowercase",
+    "namespace": "NAMESPACE",
+    "enabled": true,
+    "normalized": "namespace"
+  },
+  {
+    "id": "s2-1",
+    "description": "namespace allows alphabets, numbers, and underscore",
+    "namespace": "Dest_1001_Namespace",
+    "enabled": true,
+    "normalized": "dest_1001_namespace"
+  },
+  {
+    "id": "s2a-1",
+    "description": "namespace romanization",
+    "namespace": "namespace_with_spécial_character",
+    "enabled": true,
+    "normalized": "namespace_with_special_character"
+  },
+  {
+    "id": "s2a-2",
+    "description": "namespace romanization (japanese)",
+    "namespace": "namespace_こんにちは",
+    "enabled": false,
+    "normalized": "namespace_konnichiwa"
+  },
+  {
+    "id": "s3a-1",
+    "description": "namespace starting with a number",
+    "namespace": "99namespace",
+    "enabled": true,
+    "normalized": "99namespace"
+  },
+  {
+    "id": "s3b-1",
+    "description": "long namespace (300 characters)",
+    "namespace": "a_300_characters_looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo_namespace",
+    "enabled": false,
+    "normalized": ""
+  },
+  {
+    "id": "s3c-1",
+    "description": "reserved word (for future testing)",
+    "namespace": "select",
+    "enabled": false,
+    "normalized": ""
+  }
+]

airbyte-integrations/connectors/destination-bigquery/src/test-integration/java/io/airbyte/integrations/destination/bigquery/BigQueryGcsDestinationAcceptanceTest.java

@@ -8,11 +8,15 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.string.Strings;
+import io.airbyte.integrations.destination.NamingConventionTransformer;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Optional;
 
 public class BigQueryGcsDestinationAcceptanceTest extends BigQueryDestinationAcceptanceTest {
 
+  private static final NamingConventionTransformer NAME_TRANSFORMER = new BigQuerySQLNameTransformer();
+
   private static final Path CREDENTIALS_PATH = Path.of("secrets/credentials.json");
 
   @Override
@@ -60,4 +64,14 @@ protected void setup(final TestDestinationEnv testEnv) throws Exception {
     setupBigQuery(bigqueryConfigFromSecretFile);
   }
 
+  @Override
+  protected boolean supportNamespaceTest() {
+    return true;
+  }
+
+  @Override
+  protected Optional<NamingConventionTransformer> getNameTransformer() {
+    return Optional.of(NAME_TRANSFORMER);
+  }
+
 }

airbyte-integrations/connectors/destination-snowflake/src/test-integration/java/io/airbyte/integrations/destination/snowflake/SnowflakeInternalStagingDestinatiomAcceptanceTest.java → airbyte-integrations/connectors/destination-snowflake/src/test-integration/java/io/airbyte/integrations/destination/snowflake/SnowflakeInternalStagingDestinationAcceptanceTest.java

@@ -8,9 +8,13 @@
 import com.google.common.base.Preconditions;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.destination.NamingConventionTransformer;
 import java.nio.file.Path;
+import java.util.Optional;
 
-public class SnowflakeInternalStagingDestinatiomAcceptanceTest extends SnowflakeInsertDestinationAcceptanceTest {
+public class SnowflakeInternalStagingDestinationAcceptanceTest extends SnowflakeInsertDestinationAcceptanceTest {
+
+  private static final NamingConventionTransformer NAME_TRANSFORMER = new SnowflakeSQLNameTransformer();
 
   public JsonNode getStaticConfig() {
     final JsonNode internalStagingConfig = Jsons.deserialize(IOs.readFile(Path.of("secrets/internal_staging_config.json")));
@@ -19,4 +23,14 @@ public JsonNode getStaticConfig() {
     return internalStagingConfig;
   }
 
+  @Override
+  protected boolean supportNamespaceTest() {
+    return true;
+  }
+
+  @Override
+  protected Optional<NamingConventionTransformer> getNameTransformer() {
+    return Optional.of(NAME_TRANSFORMER);
+  }
+
 }