code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.destination.keen;
  
  import static java.util.Map.entry;
  
  import com.fasterxml.jackson.core.JsonProcessingException;
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.google.common.collect.ImmutableMap;
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.commons.resources.MoreResources;
  import io.airbyte.protocol.models.AirbyteCatalog;
  import io.airbyte.protocol.models.AirbyteMessage;
  import io.airbyte.protocol.models.AirbyteRecordMessage;
  import io.airbyte.protocol.models.AirbyteStream;
  import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
  import io.airbyte.protocol.models.ConfiguredAirbyteStream;
  import java.io.IOException;
  import java.util.List;
  import java.util.Map;
  import java.util.stream.Collectors;
  import org.junit.jupiter.api.Assertions;
  import org.junit.jupiter.api.Test;
  
  public class KeenTimestampServiceTest {
  
    private final ObjectMapper objectMapper = new ObjectMapper();
  
    @Test
    void shouldInitializeCursorFieldsFromCatalog() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("cursors_catalog.json");
  
      final Map<String, List<String>> expectedCursorFieldsMap = Map.ofEntries(
          entry("StringTypeStream1", List.of("property1")),
          entry("StringTypeStream2", List.of("property1")),
          entry("StringTypeStream3", List.of("property1")),
          entry("NumberTypeStream1", List.of("property1")),
          entry("NumberTypeStream2", List.of("property1")),
          entry("ArrayTypeStream1", List.of("property1")),
          entry("ArrayTypeStream2", List.of("property1")),
          entry("ArrayTypeStream3", List.of("property1")),
          entry("NestedCursorStream", List.of("property1", "inside")));
  
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final Map<String, List<String>> cursorFieldMap = keenTimestampService.getStreamCursorFields();
      Assertions.assertEquals(expectedCursorFieldsMap, cursorFieldMap);
    }
  
    @Test
    void shouldInjectTimestampWhenCursorIsValidString() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("string_cursor_catalog.json");
  
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog, "1999/12/15 14:44 utc");
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp("\"1999/12/15 14:44 utc\"", "1999-12-15T14:44:00Z");
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
    }
  
    @Test
    void shouldInjectNumberTimestampWhenTimestampIsSeconds() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("number_cursor_catalog.json");
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final int secondsCursor = 1628080068;
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog, secondsCursor);
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp(secondsCursor, "2021-08-04T12:27:48Z");
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
    }
  
    @Test
    void shouldInjectNumberTimestampWhenTimestampIsMillis() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("number_cursor_catalog.json");
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final long millisCursor = 1628081113151L;
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog, millisCursor);
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp(millisCursor, "2021-08-04T12:45:13.151Z");
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
    }
  
    @Test
    void shouldInjectEmittedAtWhenCursorNumberValueIsTooLow() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("number_cursor_catalog.json");
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final int notUnixTimestampCursor = 250_000;
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog, notUnixTimestampCursor);
  
      // 2020-10-14T01:09:49.200Z is hardcoded emitted at
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp(notUnixTimestampCursor, "2020-10-14T01:09:49.200Z");
  
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
    }
  
    @Test
    void shouldInjectEmittedAtWhenCursorIsUnparsableAndRemoveFieldFromMap() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("string_cursor_catalog.json");
  
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final Map<String, List<String>> cursorFieldMap = keenTimestampService.getStreamCursorFields();
      Assertions.assertEquals(cursorFieldMap.size(), 1);
  
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog, "some_text");
  
      // 2020-10-14T01:09:49.200Z is hardcoded emitted at
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp("\"some_text\"", "2020-10-14T01:09:49.200Z");
  
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
      Assertions.assertEquals(cursorFieldMap.size(), 0);
    }
  
    @Test
    void shouldInjectEmittedAtWhenCursorIsValidAndInferenceIsDisabled() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("number_cursor_catalog.json");
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, false);
  
      final int secondsCursor = 1628080068;
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog, secondsCursor);
  
      // 2020-10-14T01:09:49.200Z is hardcoded emitted at
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp(secondsCursor, "2020-10-14T01:09:49.200Z");
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
    }
  
    @Test
    void shouldInjectTimestampWhenCursorIsNestedField() throws IOException {
      final ConfiguredAirbyteCatalog configuredCatalog = readConfiguredCatalogFromFile("nested_cursor_catalog.json");
      final KeenTimestampService keenTimestampService = new KeenTimestampService(configuredCatalog, true);
  
      final int secondsCursor = 1628080068;
      final AirbyteMessage message = buildMessageWithCursorValue(configuredCatalog,
          ImmutableMap.builder().put("nestedProperty", secondsCursor).build());
  
      final String nestedJson = String.format("{\"nestedProperty\": %s}", secondsCursor);
  
      final JsonNode expectedJson = buildExpectedJsonWithTimestamp(nestedJson, "2021-08-04T12:27:48Z");
      final JsonNode jsonNode = keenTimestampService.injectTimestamp(message.getRecord());
  
      Assertions.assertEquals(jsonNode, expectedJson);
    }
  
    private <T> AirbyteMessage buildMessageWithCursorValue(final ConfiguredAirbyteCatalog configuredCatalog, final T cursorValue) {
      return new AirbyteMessage()
          .withType(AirbyteMessage.Type.RECORD)
          .withRecord(new AirbyteRecordMessage()
              .withStream(configuredCatalog.getStreams().get(0).getStream().getName())
              .withEmittedAt(1602637789200L)
              .withData(Jsons.jsonNode(ImmutableMap.builder()
                  .put("cursorProperty", cursorValue)
                  .put("otherProperty", "something")
                  .build())));
    }
  
    private <T> JsonNode buildExpectedJsonWithTimestamp(final T value, final String parsedTimestamp)
        throws JsonProcessingException {
      return objectMapper.readTree(
          String.format(
              "{" +
                  "\"cursorProperty\": %s," +
                  "\"otherProperty\": \"something\"," +
                  "\"keen\" : { \"timestamp\": \"%s\"}" +
                  "}",
              value, parsedTimestamp));
    }
  
    private ConfiguredAirbyteCatalog readConfiguredCatalogFromFile(final String fileName)
        throws IOException {
      final AirbyteCatalog catalog = Jsons.deserialize(MoreResources.readResource(fileName), AirbyteCatalog.class);
      return new ConfiguredAirbyteCatalog()
          .withStreams(catalog.getStreams()
              .stream()
              .map(this::toConfiguredStreamWithCursors)
              .collect(Collectors.toList()));
    }
  
    public ConfiguredAirbyteStream toConfiguredStreamWithCursors(final AirbyteStream stream) {
      return new ConfiguredAirbyteStream()
          .withStream(stream)
          .withCursorField(stream.getDefaultCursorField());
    }
  
  }
  