// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DateFieldConverter.java
// public class DateFieldConverter extends SinkFieldConverter {
// 
//     public DateFieldConverter() {
//         super(Date.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DecimalFieldConverter.java
// public class DecimalFieldConverter extends SinkFieldConverter {
// 
//     public enum Format {
//         DECIMAL128,         //needs MongoDB v3.4+
//         LEGACYDOUBLE        //results in double approximation
//     }
// 
//     private Format format;
// 
//     public DecimalFieldConverter() {
//         super(Decimal.schema(0));
//         this.format = Format.DECIMAL128;
//     }
// 
//     public DecimalFieldConverter(Format format) {
//         super(Decimal.schema(0));
//         this.format = format;
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
// 
//         if(data instanceof BigDecimal) {
//             if(format.equals(Format.DECIMAL128))
//                 return new BsonDecimal128(new Decimal128((BigDecimal)data));
// 
//             if(format.equals(Format.LEGACYDOUBLE))
//                 return new BsonDouble(((BigDecimal)data).doubleValue());
//         }
// 
//         throw new DataException("error: decimal conversion not possible when data is"
//                         + " of type "+data.getClass().getName() + " and format is "+format);
// 
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimeFieldConverter.java
// public class TimeFieldConverter extends SinkFieldConverter {
// 
//     public TimeFieldConverter() {
//         super(Time.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimestampFieldConverter.java
// public class TimestampFieldConverter extends SinkFieldConverter {
// 
//     public TimestampFieldConverter() {
//         super(Timestamp.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }


import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.*;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DateFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DecimalFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimeFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimestampFieldConverter;
import org.apache.kafka.connect.data.*;
import org.apache.kafka.connect.errors.DataException;
import org.bson.*;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.DynamicTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestFactory;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.time.LocalDate;
import java.time.LocalTime;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.DynamicTest.dynamicTest;
            java.util.Date.from(ZonedDateTime.of(LocalDate.of(1983,7,31), LocalTime.MIDNIGHT, ZoneOffset.systemDefault()).toInstant()),
            java.util.Date.from(ZonedDateTime.of(LocalDate.now(), LocalTime.MIDNIGHT, ZoneOffset.systemDefault()).toInstant())
        )).forEach(
                el -> tests.add(dynamicTest("conversion with "
                                + converter.getClass().getSimpleName() + " for "+el,
                        () -> assertEquals(el.toInstant().getEpochSecond()*1000,
                                                ((BsonDateTime)converter.toBson(el)).getValue())
                ))
        );

        tests.add(dynamicTest("optional type conversions", () -> {
            Schema valueOptionalDefault = Date.builder().optional().defaultValue(
                java.util.Date.from(ZonedDateTime.of(LocalDate.of(1970,1,1), LocalTime.MIDNIGHT, ZoneOffset.systemDefault()).toInstant())
            );
            assertAll("checks",
                    () -> assertThrows(DataException.class, () -> converter.toBson(null, Date.SCHEMA)),
                    () -> assertEquals(new BsonNull(), converter.toBson(null, Date.builder().optional())),
                    () -> assertEquals(((java.util.Date)valueOptionalDefault.defaultValue()).toInstant().getEpochSecond()*1000,
                            ((BsonDateTime)converter.toBson(null, valueOptionalDefault)).getValue())
            );
        }));

        return tests;

    }

    @TestFactory
    @DisplayName("tests for logical type time field conversions")
    public List<DynamicTest> testTimeFieldConverter() {

// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DateFieldConverter.java
// public class DateFieldConverter extends SinkFieldConverter {
// 
//     public DateFieldConverter() {
//         super(Date.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DecimalFieldConverter.java
// public class DecimalFieldConverter extends SinkFieldConverter {
// 
//     public enum Format {
//         DECIMAL128,         //needs MongoDB v3.4+
//         LEGACYDOUBLE        //results in double approximation
//     }
// 
//     private Format format;
// 
//     public DecimalFieldConverter() {
//         super(Decimal.schema(0));
//         this.format = Format.DECIMAL128;
//     }
// 
//     public DecimalFieldConverter(Format format) {
//         super(Decimal.schema(0));
//         this.format = format;
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
// 
//         if(data instanceof BigDecimal) {
//             if(format.equals(Format.DECIMAL128))
//                 return new BsonDecimal128(new Decimal128((BigDecimal)data));
// 
//             if(format.equals(Format.LEGACYDOUBLE))
//                 return new BsonDouble(((BigDecimal)data).doubleValue());
//         }
// 
//         throw new DataException("error: decimal conversion not possible when data is"
//                         + " of type "+data.getClass().getName() + " and format is "+format);
// 
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimeFieldConverter.java
// public class TimeFieldConverter extends SinkFieldConverter {
// 
//     public TimeFieldConverter() {
//         super(Time.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimestampFieldConverter.java
// public class TimestampFieldConverter extends SinkFieldConverter {
// 
//     public TimestampFieldConverter() {
//         super(Timestamp.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }


// Path: src/test/java/at/grahsl/kafka/connect/mongodb/converter/SinkFieldConverterTest.java
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.*;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DateFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DecimalFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimeFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimestampFieldConverter;
import org.apache.kafka.connect.data.*;
import org.apache.kafka.connect.errors.DataException;
import org.bson.*;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.DynamicTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestFactory;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.time.LocalDate;
import java.time.LocalTime;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.DynamicTest.dynamicTest;

            java.util.Date.from(ZonedDateTime.of(LocalDate.of(1983,7,31), LocalTime.MIDNIGHT, ZoneOffset.systemDefault()).toInstant()),
            java.util.Date.from(ZonedDateTime.of(LocalDate.now(), LocalTime.MIDNIGHT, ZoneOffset.systemDefault()).toInstant())
        )).forEach(
                el -> tests.add(dynamicTest("conversion with "
                                + converter.getClass().getSimpleName() + " for "+el,
                        () -> assertEquals(el.toInstant().getEpochSecond()*1000,
                                                ((BsonDateTime)converter.toBson(el)).getValue())
                ))
        );

        tests.add(dynamicTest("optional type conversions", () -> {
            Schema valueOptionalDefault = Date.builder().optional().defaultValue(
                java.util.Date.from(ZonedDateTime.of(LocalDate.of(1970,1,1), LocalTime.MIDNIGHT, ZoneOffset.systemDefault()).toInstant())
            );
            assertAll("checks",
                    () -> assertThrows(DataException.class, () -> converter.toBson(null, Date.SCHEMA)),
                    () -> assertEquals(new BsonNull(), converter.toBson(null, Date.builder().optional())),
                    () -> assertEquals(((java.util.Date)valueOptionalDefault.defaultValue()).toInstant().getEpochSecond()*1000,
                            ((BsonDateTime)converter.toBson(null, valueOptionalDefault)).getValue())
            );
        }));

        return tests;

    }

    @TestFactory
    @DisplayName("tests for logical type time field conversions")
    public List<DynamicTest> testTimeFieldConverter() {

        SinkFieldConverter converter = new TimeFieldConverter();