ElasticSearch, GCS, Keen, Kafka, Kinesis destinations : Enable DAT te…

…sts (#11971) * add Boolean, Number, DateTimeWithTZ compare methods * Improve value comparison * Compare inherit objects element by element * Move comparison methods to a new class not to overload the test class. Basic - old implementation which provides backward compatibility of this PR. Advanced - comparator required for destinations with enabled DAT tests. * format * Move common method to ComparatorUtils. + Review update * format * review * review * mark resolveIdentifier method as deprecated * enable DAT tests for destination-Elasticsearch, destination-gcs, destination-kafka, destination-keen, destination-kinesis * remove size objects validation. We iterate expected elements and compare it with actualObject elements. Expected might have empty elements and they are equal missing element in the actual Object. Also, actualObject might contain additional elements which is not an exception. * fix Csv actual data read. + use proper comparator for different file types * format
airbytehq · Apr 19, 2022 · 57d059f · 57d059f
1 parent 3113ef4
commit 57d059f
Show file tree

Hide file tree

Showing 9 changed files with 168 additions and 0 deletions.
diff --git a/...irbyte/integrations/destination/elasticsearch/ElasticsearchDestinationAcceptanceTest.java b/...irbyte/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();

diff --git a/...ration/java/io/airbyte/integrations/destination/gcs/GcsAvroDestinationAcceptanceTest.java b/...ration/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,

diff --git a/...t-integration/java/io/airbyte/integrations/destination/gcs/GcsAvroTestDataComparator.java b/...t-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));
+  }
+
+}
diff --git a/...gration/java/io/airbyte/integrations/destination/gcs/GcsCsvDestinationAcceptanceTest.java b/...gration/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,

diff --git a/...ntegration/java/io/airbyte/integrations/destination/gcs/GcsDestinationAcceptanceTest.java b/...ntegration/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();

diff --git a/...ion/java/io/airbyte/integrations/destination/gcs/GcsParquetDestinationAcceptanceTest.java b/...ion/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,

diff --git a/...ration/java/io/airbyte/integrations/destination/kafka/KafkaDestinationAcceptanceTest.java b/...ration/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);

diff --git a/...c/test-integration/java/io/airbyte/integrations/destination/keen/KeenDestinationTest.java b/...c/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");

diff --git a/...on/java/io/airbyte/integrations/destination/kinesis/KinesisDestinationAcceptanceTest.java b/...on/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();