content: /*
   * Copyright 2019 Aiven Oy and jdbc-connector-for-apache-kafka project contributors
   * Copyright 2017 Confluent Inc.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package io.aiven.connect.jdbc.source;
  
  import java.math.BigDecimal;
  import java.time.ZoneOffset;
  import java.util.Arrays;
  import java.util.List;
  import java.util.TimeZone;
  
  import org.apache.kafka.connect.data.Decimal;
  import org.apache.kafka.connect.data.Schema;
  import org.apache.kafka.connect.data.SchemaBuilder;
  import org.apache.kafka.connect.data.Struct;
  import org.apache.kafka.connect.data.Timestamp;
  import org.apache.kafka.connect.errors.ConnectException;
  
  import io.aiven.connect.jdbc.util.ColumnId;
  import io.aiven.connect.jdbc.util.TableId;
  
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  
  import static org.assertj.core.api.Assertions.assertThat;
  import static org.assertj.core.api.Assertions.assertThatThrownBy;
  
  public class TimestampIncrementingCriteriaTest {
  
      private static final TableId TABLE_ID = new TableId(null, null, "myTable");
      private static final ColumnId INCREMENTING_COLUMN = new ColumnId(TABLE_ID, "id");
      private static final ColumnId TS1_COLUMN = new ColumnId(TABLE_ID, "ts1");
      private static final ColumnId TS2_COLUMN = new ColumnId(TABLE_ID, "ts2");
      private static final List<ColumnId> TS_COLUMNS = Arrays.asList(TS1_COLUMN, TS2_COLUMN);
  
      private TimestampIncrementingCriteria criteria;
      private TimestampIncrementingCriteria criteriaInc;
      private TimestampIncrementingCriteria criteriaTs;
      private TimestampIncrementingCriteria criteriaIncTs;
      private Schema schema;
      private Struct record;
      private final TimeZone utcTimeZone = TimeZone.getTimeZone(ZoneOffset.UTC);
  
      @BeforeEach
      public void beforeEach() {
          criteria = new TimestampIncrementingCriteria(null, null, utcTimeZone);
          criteriaInc = new TimestampIncrementingCriteria(INCREMENTING_COLUMN, null, utcTimeZone);
          criteriaTs = new TimestampIncrementingCriteria(null, TS_COLUMNS, utcTimeZone);
          criteriaIncTs = new TimestampIncrementingCriteria(INCREMENTING_COLUMN, TS_COLUMNS, utcTimeZone);
      }
  
      protected void assertExtractedOffset(final long expected, final Schema schema, final Struct record) {
          final TimestampIncrementingCriteria criteria;
          if (schema.field(INCREMENTING_COLUMN.name()) != null) {
              if (schema.field(TS1_COLUMN.name()) != null) {
                  criteria = criteriaIncTs;
              } else {
                  criteria = criteriaInc;
              }
          } else if (schema.field(TS1_COLUMN.name()) != null) {
              criteria = criteriaTs;
          } else {
              criteria = this.criteria;
          }
          final TimestampIncrementingOffset offset = criteria.extractValues(schema, record, null);
          assertThat(offset.getIncrementingOffset()).isEqualTo(expected);
      }
  
      @Test
      public void extractIntOffset() {
          schema = SchemaBuilder.struct().field("id", SchemaBuilder.INT32_SCHEMA).build();
          record = new Struct(schema).put("id", 42);
          assertExtractedOffset(42L, schema, record);
      }
  
      @Test
      public void extractLongOffset() {
          schema = SchemaBuilder.struct().field("id", SchemaBuilder.INT64_SCHEMA).build();
          record = new Struct(schema).put("id", 42L);
          assertExtractedOffset(42L, schema, record);
      }
  
      @Test
      public void extractDecimalOffset() {
          final Schema decimalSchema = Decimal.schema(0);
          schema = SchemaBuilder.struct().field("id", decimalSchema).build();
          record = new Struct(schema).put("id", new BigDecimal(42));
          assertExtractedOffset(42L, schema, record);
      }
  
      @Test
      public void extractTooLargeDecimalOffset() {
          final Schema decimalSchema = Decimal.schema(0);
          schema = SchemaBuilder.struct().field("id", decimalSchema).build();
          record = new Struct(schema).put(
              "id",
              new BigDecimal(Long.MAX_VALUE).add(new BigDecimal(1)));
          assertThatThrownBy(() -> assertExtractedOffset(42L, schema, record))
              .isInstanceOf(ConnectException.class);
      }
  
      @Test
      public void extractFractionalDecimalOffset() {
          final Schema decimalSchema = Decimal.schema(2);
          schema = SchemaBuilder.struct().field("id", decimalSchema).build();
          record = new Struct(schema).put("id", new BigDecimal("42.42"));
          assertThatThrownBy(() -> assertExtractedOffset(42L, schema, record))
              .isInstanceOf(ConnectException.class);
      }
  
      @Test
      public void extractWithIncColumn() {
          schema = SchemaBuilder.struct()
              .field("id", SchemaBuilder.INT32_SCHEMA)
              .field(TS1_COLUMN.name(), Timestamp.SCHEMA)
              .field(TS2_COLUMN.name(), Timestamp.SCHEMA)
              .build();
          record = new Struct(schema).put("id", 42);
          assertExtractedOffset(42L, schema, record);
  
      }
  }
  