ElasticSearch, GCS, Keen, Kafka, Kinesis destinations : Enable DAT tests

airbyte-integrations/connectors/destination-elasticsearch/src/test-integration/java/io/airbyte/integrations/destination/elasticsearch/ElasticsearchDestinationAcceptanceTest.java

@@ -7,6 +7,8 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.io.IOException;
 import java.time.Duration;
 import java.util.List;
@@ -63,6 +65,26 @@ protected boolean supportsNormalization() {
     return false;
   }
 
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new AdvancedTestDataComparator();
+  }
+
   @Override
   protected JsonNode getConfig() {
     var configJson = mapper.createObjectNode();

airbyte-integrations/connectors/destination-gcs/src/test-integration/java/io/airbyte/integrations/destination/gcs/GcsAvroDestinationAcceptanceTest.java

@@ -13,6 +13,7 @@
 import io.airbyte.integrations.destination.s3.avro.AvroConstants;
 import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
 import io.airbyte.integrations.destination.s3.util.AvroRecordHelper;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.util.LinkedList;
 import java.util.List;
 import org.apache.avro.file.DataFileReader;
@@ -35,6 +36,11 @@ protected JsonNode getFormatConfig() {
         + "}");
   }
 
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new GcsAvroTestDataComparator();
+  }
+
   @Override
   protected List<JsonNode> retrieveRecords(final TestDestinationEnv testEnv,
                                            final String streamName,

airbyte-integrations/connectors/destination-gcs/src/test-integration/java/io/airbyte/integrations/destination/gcs/GcsAvroTestDataComparator.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.gcs;
+
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import java.time.*;
+import java.time.format.DateTimeFormatter;
+
+public class GcsAvroTestDataComparator extends AdvancedTestDataComparator {
+
+  @Override
+  protected boolean compareDateValues(String expectedValue, String actualValue) {
+    var destinationDate = LocalDate.ofEpochDay(Long.parseLong(actualValue));
+    var expectedDate = LocalDate.parse(expectedValue, DateTimeFormatter.ofPattern(AIRBYTE_DATE_FORMAT));
+    return expectedDate.equals(destinationDate);
+  }
+
+  private Instant getInstantFromEpoch(String epochValue) {
+    return Instant.ofEpochMilli(Long.parseLong(epochValue.replaceAll("000$", "")));
+  }
+
+  @Override
+  protected ZonedDateTime parseDestinationDateWithTz(String destinationValue) {
+    return ZonedDateTime.ofInstant(getInstantFromEpoch(destinationValue), ZoneOffset.UTC);
+  }
+
+  @Override
+  protected boolean compareDateTimeValues(String airbyteMessageValue, String destinationValue) {
+    var format = DateTimeFormatter.ofPattern(AIRBYTE_DATETIME_FORMAT);
+    LocalDateTime dateTime = LocalDateTime.ofInstant(getInstantFromEpoch(destinationValue), ZoneOffset.UTC);
+    return super.compareDateTimeValues(airbyteMessageValue, format.format(dateTime));
+  }
+
+}

airbyte-integrations/connectors/destination-gcs/src/test-integration/java/io/airbyte/integrations/destination/gcs/GcsCsvDestinationAcceptanceTest.java

@@ -77,12 +77,22 @@ private static JsonNode getJsonNode(final Map<String, String> input, final Map<S
         case "boolean" -> json.put(key, Boolean.valueOf(value));
         case "integer" -> json.put(key, Integer.valueOf(value));
         case "number" -> json.put(key, Double.valueOf(value));
+        case "" -> addNoTypeValue(json, key, value);
         default -> json.put(key, value);
       }
     }
     return json;
   }
 
+  private static void addNoTypeValue(ObjectNode json, String key, String value) {
+    if (value != null && (value.matches("^\\[.*\\]$")) || value.matches("^\\{.*\\}$")) {
+      var newNode = Jsons.deserialize(value);
+      json.set(key, newNode);
+    } else {
+      json.put(key, value);
+    }
+  }
+
   @Override
   protected List<JsonNode> retrieveRecords(final TestDestinationEnv testEnv,
                                            final String streamName,

airbyte-integrations/connectors/destination-gcs/src/test-integration/java/io/airbyte/integrations/destination/gcs/GcsDestinationAcceptanceTest.java

@@ -21,6 +21,8 @@
 import io.airbyte.integrations.destination.s3.S3FormatConfig;
 import io.airbyte.integrations.destination.s3.S3StorageOperations;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.nio.file.Path;
 import java.util.Comparator;
 import java.util.LinkedList;
@@ -85,6 +87,26 @@ protected String getDefaultSchema(final JsonNode config) {
     return null;
   }
 
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new AdvancedTestDataComparator();
+  }
+
   @Override
   protected JsonNode getFailCheckConfig() {
     final JsonNode baseJson = getBaseConfigJson();

airbyte-integrations/connectors/destination-gcs/src/test-integration/java/io/airbyte/integrations/destination/gcs/GcsParquetDestinationAcceptanceTest.java

@@ -14,6 +14,7 @@
 import io.airbyte.integrations.destination.s3.avro.AvroConstants;
 import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
 import io.airbyte.integrations.destination.s3.util.AvroRecordHelper;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -38,6 +39,11 @@ protected JsonNode getFormatConfig() {
         "compression_codec", "GZIP"));
   }
 
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new GcsAvroTestDataComparator();
+  }
+
   @Override
   protected List<JsonNode> retrieveRecords(final TestDestinationEnv testEnv,
                                            final String streamName,

airbyte-integrations/connectors/destination-kafka/src/test-integration/java/io/airbyte/integrations/destination/kafka/KafkaDestinationAcceptanceTest.java

@@ -14,6 +14,8 @@
 import io.airbyte.integrations.destination.NamingConventionTransformer;
 import io.airbyte.integrations.destination.StandardNameTransformer;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -114,6 +116,26 @@ protected String getDefaultSchema(final JsonNode config) {
     return "";
   }
 
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new AdvancedTestDataComparator();
+  }
+
   @Override
   protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv testEnv, final String streamName, final String namespace) {
     return retrieveRecords(testEnv, streamName, namespace, null);

airbyte-integrations/connectors/destination-keen/src/test-integration/java/io/airbyte/integrations/destination/keen/KeenDestinationTest.java

@@ -14,6 +14,8 @@
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import java.nio.file.Files;
@@ -45,6 +47,26 @@ protected JsonNode getConfig() throws Exception {
     return configJson;
   }
 
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new AdvancedTestDataComparator();
+  }
+
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
   @Override
   protected JsonNode getFailCheckConfig() throws Exception {
     ((ObjectNode) configJson).put(CONFIG_PROJECT_ID, "fake");

airbyte-integrations/connectors/destination-kinesis/src/test-integration/java/io/airbyte/integrations/destination/kinesis/KinesisDestinationAcceptanceTest.java

@@ -7,6 +7,8 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.util.Comparator;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -42,6 +44,26 @@ protected void setup(TestDestinationEnv testEnv) {
     kinesisNameTransformer = new KinesisNameTransformer();
   }
 
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new AdvancedTestDataComparator();
+  }
+
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
   @Override
   protected void tearDown(TestDestinationEnv testEnv) {
     kinesisStream.deleteAllStreams();