Destination acceptence test : improve comparison methods

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

@@ -7,7 +7,6 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -29,6 +28,8 @@
 import io.airbyte.config.StandardCheckConnectionOutput.Status;
 import io.airbyte.config.WorkerDestinationConfig;
 import io.airbyte.integrations.destination.NamingConventionTransformer;
+import io.airbyte.integrations.standardtest.destination.comparator.BasicTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import io.airbyte.protocol.models.AirbyteCatalog;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.AirbyteMessage.Type;
@@ -63,10 +64,8 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
-import java.util.Iterator;
 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;
@@ -99,10 +98,12 @@ public abstract class DestinationAcceptanceTest {
   private TestDestinationEnv testEnv;
 
   private Path jobRoot;
-  protected Path localRoot;
   private ProcessFactory processFactory;
   private WorkerConfigs workerConfigs;
 
+  protected Path localRoot;
+  protected TestDataComparator testDataComparator = getTestDataComparator();
+
   /**
    * Name of the docker image that the tests will run against.
    *
@@ -300,10 +301,13 @@ protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv test
    */
   protected abstract void tearDown(TestDestinationEnv testEnv) throws Exception;
 
+  /**
+   * @deprecated This method is moved to the AdvancedTestDataComparator. Please move your destination
+   *             implementation of the method to your comparator implementation.
+   */
+  @Deprecated
   protected List<String> resolveIdentifier(final String identifier) {
-    final List<String> result = new ArrayList<>();
-    result.add(identifier);
-    return result;
+    return List.of(identifier);
   }
 
   @BeforeEach
@@ -1115,8 +1119,7 @@ protected void retrieveRawRecordsAndAssertSameMessages(final AirbyteCatalog cata
       final String schema = stream.getNamespace() != null ? stream.getNamespace() : defaultSchema;
       final List<AirbyteRecordMessage> msgList = retrieveRecords(testEnv, streamName, schema, stream.getJsonSchema())
           .stream()
-          .map(data -> new AirbyteRecordMessage().withStream(streamName).withNamespace(schema).withData(data))
-          .collect(Collectors.toList());
+          .map(data -> new AirbyteRecordMessage().withStream(streamName).withNamespace(schema).withData(data)).toList();
       actualMessages.addAll(msgList);
     }
 
@@ -1140,44 +1143,7 @@ protected void assertSameMessages(final List<AirbyteMessage> expected,
         .map(AirbyteRecordMessage::getData)
         .collect(Collectors.toList());
 
-    assertSameData(expectedProcessed, actualProcessed);
-  }
-
-  private void assertSameData(final List<JsonNode> expected, final List<JsonNode> actual) {
-    LOGGER.info("Expected data {}", expected);
-    LOGGER.info("Actual data   {}", actual);
-    assertEquals(expected.size(), actual.size());
-    final Iterator<JsonNode> expectedIterator = expected.iterator();
-    final Iterator<JsonNode> actualIterator = actual.iterator();
-    while (expectedIterator.hasNext() && actualIterator.hasNext()) {
-      final JsonNode expectedData = expectedIterator.next();
-      final JsonNode actualData = actualIterator.next();
-      final Iterator<Entry<String, JsonNode>> expectedDataIterator = expectedData.fields();
-      LOGGER.info("Expected row {}", expectedData);
-      LOGGER.info("Actual row   {}", actualData);
-      assertEquals(expectedData.size(), actualData.size(), "Unequal row size");
-      while (expectedDataIterator.hasNext()) {
-        final Entry<String, JsonNode> expectedEntry = expectedDataIterator.next();
-        final JsonNode expectedValue = expectedEntry.getValue();
-        JsonNode actualValue = null;
-        String key = expectedEntry.getKey();
-        for (final String tmpKey : resolveIdentifier(expectedEntry.getKey())) {
-          actualValue = actualData.get(tmpKey);
-          if (actualValue != null) {
-            key = tmpKey;
-            break;
-          }
-        }
-        LOGGER.info("For {} Expected {} vs Actual {}", key, expectedValue, actualValue);
-        assertTrue(actualData.has(key));
-        assertSameValue(expectedValue, actualValue);
-      }
-    }
-  }
-
-  // Allows subclasses to implement custom comparison asserts
-  protected void assertSameValue(final JsonNode expectedValue, final JsonNode actualValue) {
-    assertEquals(expectedValue, actualValue);
+    testDataComparator.assertSameData(expectedProcessed, actualProcessed);
   }
 
   protected List<AirbyteRecordMessage> retrieveNormalizedRecords(final AirbyteCatalog catalog, final String defaultSchema) throws Exception {
@@ -1188,8 +1154,7 @@ protected List<AirbyteRecordMessage> retrieveNormalizedRecords(final AirbyteCata
 
       final List<AirbyteRecordMessage> msgList = retrieveNormalizedRecords(testEnv, streamName, defaultSchema)
           .stream()
-          .map(data -> new AirbyteRecordMessage().withStream(streamName).withData(data))
-          .collect(Collectors.toList());
+          .map(data -> new AirbyteRecordMessage().withStream(streamName).withData(data)).toList();
       actualMessages.addAll(msgList);
     }
     return actualMessages;
@@ -1389,6 +1354,10 @@ public void testStressPerformance() throws Exception {
           + "\n"
           + "Praesent finibus scelerisque elit, accumsan condimentum risus mattis vitae. Donec tristique hendrerit facilisis. Curabitur metus purus, venenatis non elementum id, finibus eu augue. Quisque posuere rhoncus ligula, et vehicula erat pulvinar at. Pellentesque vel quam vel lectus tincidunt congue quis id sapien. Ut efficitur mauris vitae pretium iaculis. Aliquam consectetur iaculis nisi vitae laoreet. Integer vel odio quis diam mattis tempor eget nec est. Donec iaculis facilisis neque, at dictum magna vestibulum ut. Sed malesuada non nunc ac consequat. Maecenas tempus lectus a nisl congue, ac venenatis diam viverra. Nam ac justo id nulla iaculis lobortis in eu ligula. Vivamus et ligula id sapien efficitur aliquet. Curabitur est justo, tempus vitae mollis quis, tincidunt vitae felis. Vestibulum molestie laoreet justo, nec mollis purus vulputate at.";
 
+  protected TestDataComparator getTestDataComparator() {
+    return new BasicTestDataComparator(this::resolveIdentifier);
+  }
+
   protected boolean supportBasicDataTypeTest() {
     return false;
   }

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

@@ -0,0 +1,186 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.standardtest.destination.comparator;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AdvancedTestDataComparator implements TestDataComparator {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(AdvancedTestDataComparator.class);
+
+  public static final String AIRBYTE_DATE_FORMAT = "yyyy-MM-dd";
+  public static final String AIRBYTE_DATETIME_FORMAT = "yyyy-MM-dd'T'HH:mm:ss";
+  public static final String AIRBYTE_DATETIME_WITH_TZ_FORMAT = "yyyy-MM-dd'T'HH:mm:ssXXX";
+
+  @Override
+  public void assertSameData(List<JsonNode> expected, List<JsonNode> actual) {
+    LOGGER.info("Expected data {}", expected);
+    LOGGER.info("Actual data   {}", actual);
+    assertEquals(expected.size(), actual.size());
+    final Iterator<JsonNode> expectedIterator = expected.iterator();
+    final Iterator<JsonNode> actualIterator = actual.iterator();
+    while (expectedIterator.hasNext() && actualIterator.hasNext()) {
+      compareObjects(expectedIterator.next(), actualIterator.next());
+    }
+  }
+
+  protected List<String> resolveIdentifier(final String identifier) {
+    return List.of(identifier);
+  }
+
+  protected void compareObjects(final JsonNode expectedObject, final JsonNode actualObject) {
+    if (!areBothEmpty(expectedObject, actualObject)) {
+      LOGGER.info("Expected Object : {}", expectedObject);
+      LOGGER.info("Actual Object   : {}", actualObject);
+      final Iterator<Map.Entry<String, JsonNode>> expectedDataIterator = expectedObject.fields();
+      while (expectedDataIterator.hasNext()) {
+        final Map.Entry<String, JsonNode> expectedEntry = expectedDataIterator.next();
+        final JsonNode expectedValue = expectedEntry.getValue();
+        String key = expectedEntry.getKey();
+        JsonNode actualValue = ComparatorUtils.getActualValueByExpectedKey(key, actualObject, this::resolveIdentifier);
+        LOGGER.info("For {} Expected {} vs Actual {}", key, expectedValue, actualValue);
+        assertSameValue(expectedValue, actualValue);
+      }
+    } else {
+      LOGGER.info("Both rows are empty.");
+    }
+  }
+
+  private boolean isJsonNodeEmpty(final JsonNode jsonNode) {
+    return jsonNode.isEmpty() || (jsonNode.size() == 1 && jsonNode.iterator().next().asText().isEmpty());
+  }
+
+  private boolean areBothEmpty(final JsonNode expectedData, final JsonNode actualData) {
+    return isJsonNodeEmpty(expectedData) && isJsonNodeEmpty(actualData);
+  }
+
+  // Allows subclasses to implement custom comparison asserts
+  protected void assertSameValue(final JsonNode expectedValue, final JsonNode actualValue) {
+    LOGGER.info("assertSameValue : {} vs {}", expectedValue, actualValue);
+
+    assertTrue(compareJsonNodes(expectedValue, actualValue), "Expected value " + expectedValue + " vs Actual value " + actualValue);
+  }
+
+  protected boolean compareJsonNodes(final JsonNode expectedValue, final JsonNode actualValue) {
+    if (expectedValue == null || actualValue == null) {
+      return expectedValue == null && actualValue == null;
+    } else if (expectedValue.isNumber() || expectedValue.isDouble() || expectedValue.isFloat()) {
+      return compareNumericValues(expectedValue.asText(), actualValue.asText());
+    } else if (expectedValue.isBoolean()) {
+      return compareBooleanValues(expectedValue.asText(), actualValue.asText());
+    } else if (isDateTimeWithTzValue(expectedValue.asText())) {
+      return compareDateTimeWithTzValues(expectedValue.asText(), actualValue.asText());
+    } else if (isDateTimeValue(expectedValue.asText())) {
+      return compareDateTimeValues(expectedValue.asText(), actualValue.asText());
+    } else if (isDateValue(expectedValue.asText())) {
+      return compareDateValues(expectedValue.asText(), actualValue.asText());
+    } else if (expectedValue.isArray() && actualValue.isArray()) {
+      return compareArrays(expectedValue, actualValue);
+    } else if (expectedValue.isObject() && actualValue.isObject()) {
+      compareObjects(expectedValue, actualValue);
+      return true;
+    } else {
+      return compareString(expectedValue, actualValue);
+    }
+  }
+
+  protected boolean compareString(final JsonNode expectedValue, final JsonNode actualValue) {
+    return expectedValue.asText().equals(actualValue.asText());
+  }
+
+  private List<JsonNode> getArrayList(final JsonNode jsonArray) {
+    List<JsonNode> result = new ArrayList<>();
+    jsonArray.elements().forEachRemaining(result::add);
+    return result;
+  }
+
+  protected boolean compareArrays(final JsonNode expectedArray, final JsonNode actualArray) {
+    var expectedList = getArrayList(expectedArray);
+    var actualList = getArrayList(actualArray);
+
+    if (expectedList.size() != actualList.size()) {
+      return false;
+    } else {
+      for (JsonNode expectedNode : expectedList) {
+        var sameActualNode = actualList.stream().filter(actualNode -> compareJsonNodes(expectedNode, actualNode)).findFirst();
+        if (sameActualNode.isPresent()) {
+          actualList.remove(sameActualNode.get());
+        } else {
+          return false;
+        }
+      }
+      return true;
+    }
+  }
+
+  protected boolean compareBooleanValues(final String firstBooleanValue, final String secondBooleanValue) {
+    return Boolean.parseBoolean(firstBooleanValue) == Boolean.parseBoolean(secondBooleanValue);
+  }
+
+  protected boolean compareNumericValues(final String firstNumericValue, final String secondNumericValue) {
+    double firstValue = Double.parseDouble(firstNumericValue);
+    double secondValue = Double.parseDouble(secondNumericValue);
+
+    return firstValue == secondValue;
+  }
+
+  protected DateTimeFormatter getAirbyteDateTimeWithTzFormatter() {
+    return DateTimeFormatter.ofPattern(AIRBYTE_DATETIME_WITH_TZ_FORMAT);
+  }
+
+  protected boolean isDateTimeWithTzValue(final String value) {
+    return value.matches(".+[+-]\\d{2}:\\d{2}");
+  }
+
+  protected ZonedDateTime parseDestinationDateWithTz(final String destinationValue) {
+    return ZonedDateTime.parse(destinationValue, DateTimeFormatter.ofPattern(AIRBYTE_DATETIME_WITH_TZ_FORMAT)).withZoneSameInstant(ZoneOffset.UTC);
+  }
+
+  protected boolean compareDateTimeWithTzValues(final String airbyteMessageValue, final String destinationValue) {
+    try {
+      ZonedDateTime airbyteDate = ZonedDateTime.parse(airbyteMessageValue, getAirbyteDateTimeWithTzFormatter()).withZoneSameInstant(ZoneOffset.UTC);
+      ZonedDateTime destinationDate = parseDestinationDateWithTz(destinationValue);
+      return airbyteDate.equals(destinationDate);
+    } catch (DateTimeParseException e) {
+      LOGGER.warn("Fail to convert values to ZonedDateTime. Try to compare as text. Airbyte value({}), Destination value ({}). Exception: {}",
+          airbyteMessageValue, destinationValue, e);
+      return compareTextValues(airbyteMessageValue, destinationValue);
+    }
+  }
+
+  protected boolean isDateTimeValue(final String value) {
+    return value.matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}");
+  }
+
+  protected boolean compareDateTimeValues(final String airbyteMessageValue, final String destinationValue) {
+    return compareTextValues(airbyteMessageValue, destinationValue);
+  }
+
+  protected boolean isDateValue(final String value) {
+    return value.matches("\\d{4}-\\d{2}-\\d{2}");
+  }
+
+  protected boolean compareDateValues(final String airbyteMessageValue, final String destinationValue) {
+    return compareTextValues(airbyteMessageValue, destinationValue);
+  }
+
+  protected boolean compareTextValues(final String firstValue, final String secondValue) {
+    return firstValue.equals(secondValue);
+  }
+
+}

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.standardtest.destination.comparator;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BasicTestDataComparator implements TestDataComparator {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(BasicTestDataComparator.class);
+
+  private final Function<String, List<String>> nameResolver;
+
+  public BasicTestDataComparator(Function<String, List<String>> nameResolver) {
+    this.nameResolver = nameResolver;
+  }
+
+  @Override
+  public void assertSameData(List<JsonNode> expected, List<JsonNode> actual) {
+    LOGGER.info("Expected data {}", expected);
+    LOGGER.info("Actual data   {}", actual);
+    assertEquals(expected.size(), actual.size());
+    final Iterator<JsonNode> expectedIterator = expected.iterator();
+    final Iterator<JsonNode> actualIterator = actual.iterator();
+    while (expectedIterator.hasNext() && actualIterator.hasNext()) {
+      final JsonNode expectedData = expectedIterator.next();
+      final JsonNode actualData = actualIterator.next();
+      final Iterator<Map.Entry<String, JsonNode>> expectedDataIterator = expectedData.fields();
+      LOGGER.info("Expected row {}", expectedData);
+      LOGGER.info("Actual row   {}", actualData);
+      assertEquals(expectedData.size(), actualData.size(), "Unequal row size");
+      while (expectedDataIterator.hasNext()) {
+        final Map.Entry<String, JsonNode> expectedEntry = expectedDataIterator.next();
+        final JsonNode expectedValue = expectedEntry.getValue();
+        String key = expectedEntry.getKey();
+        JsonNode actualValue = ComparatorUtils.getActualValueByExpectedKey(key, actualData, nameResolver);
+        LOGGER.info("For {} Expected {} vs Actual {}", key, expectedValue, actualValue);
+        assertSameValue(expectedValue, actualValue);
+      }
+    }
+  }
+
+  // Allows subclasses to implement custom comparison asserts
+  protected void assertSameValue(final JsonNode expectedValue, final JsonNode actualValue) {
+    assertEquals(expectedValue, actualValue);
+  }
+
+}