content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You 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 org.apache.camel.kafkaconnector.transforms;
  
  import java.nio.charset.Charset;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Map;
  
  import io.netty.buffer.Unpooled;
  import org.apache.camel.component.netty.http.NettyChannelBufferStreamCache;
  import org.apache.kafka.common.config.ConfigException;
  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.source.SourceRecord;
  import org.apache.kafka.connect.transforms.Transformation;
  import org.junit.jupiter.api.Test;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  public class CamelTypeConverterTransformTest {
  
      @Test
      public void testIfItConvertsConnectRecordCorrectly() {
          final SourceRecord connectRecord = new SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "topic", Schema.STRING_SCHEMA, "1234", Schema.STRING_SCHEMA, "TRUE");
  
          final Map<String, Object> propsForKeySmt = new HashMap<>();
          propsForKeySmt.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG, Integer.class.getName());
  
          final Map<String, Object> propsForValueSmt = new HashMap<>();
          propsForValueSmt.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG, "java.lang.Boolean");
  
          final Transformation<SourceRecord> transformationKey = new CamelTypeConverterTransform.Key<>();
          final Transformation<SourceRecord> transformationValue = new CamelTypeConverterTransform.Value<>();
  
          transformationKey.configure(propsForKeySmt);
          transformationValue.configure(propsForValueSmt);
  
          final SourceRecord transformedKeySourceRecord = transformationKey.apply(connectRecord);
          final SourceRecord transformedValueSourceRecord = transformationValue.apply(connectRecord);
  
          assertEquals(1234, transformedKeySourceRecord.key());
          assertEquals(Schema.INT32_SCHEMA, transformedKeySourceRecord.keySchema());
  
          assertEquals(true, transformedValueSourceRecord.value());
          assertEquals(Schema.BOOLEAN_SCHEMA, transformedValueSourceRecord.valueSchema());
      }
  
      @Test
      public void testIfItConvertsNettyCorrectly() {
          final String testMessage = "testMessage";
          NettyChannelBufferStreamCache nettyTestValue = new NettyChannelBufferStreamCache(Unpooled.wrappedBuffer(testMessage.getBytes(Charset.defaultCharset())));
  
          final SourceRecord connectRecord = new SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "topic", Schema.STRING_SCHEMA, "1234", Schema.BYTES_SCHEMA, nettyTestValue);
  
          final Map<String, Object> propsForValueSmt = new HashMap<>();
          propsForValueSmt.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG, "java.lang.String");
  
          final Transformation<SourceRecord> transformationValue = new CamelTypeConverterTransform.Value<>();
  
          transformationValue.configure(propsForValueSmt);
  
          final SourceRecord transformedValueSourceRecord = transformationValue.apply(connectRecord);
  
          assertEquals(java.lang.String.class, transformedValueSourceRecord.value().getClass());
          assertEquals(Schema.STRING_SCHEMA, transformedValueSourceRecord.valueSchema());
          assertEquals(testMessage, transformedValueSourceRecord.value());
      }
  
      @Test
      public void testIfHandlesTypeConvertersFromCamelComponents() {
          // we know we have a type converter from struct to map in dbz component, so we use this for testing
          final Schema schema = SchemaBuilder.struct()
                  .field("id", Schema.INT32_SCHEMA)
                  .field("name", Schema.STRING_SCHEMA)
                  .field("valid", Schema.BOOLEAN_SCHEMA)
                  .field("extra", Schema.STRING_SCHEMA)
                  .build();
  
          final Struct value = new Struct(schema);
          value.put("id", 12);
          value.put("name", "test-name");
          value.put("valid", true);
  
          final SourceRecord connectRecord = new SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "topic", Schema.STRING_SCHEMA, "1234", schema, value);
  
          final Map<String, Object> props = new HashMap<>();
          props.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG, Map.class.getName());
  
          final Transformation<SourceRecord> transformationValue = new CamelTypeConverterTransform.Value<>();
  
          transformationValue.configure(props);
  
          final SourceRecord transformedValueSourceRecord = transformationValue.apply(connectRecord);
  
          // assert
          assertNotNull(transformedValueSourceRecord);
  
          final Map<String, Object> outputValue = (Map<String, Object>) transformedValueSourceRecord.value();
  
          assertEquals(12, outputValue.get("id"));
          assertEquals("test-name", outputValue.get("name"));
          assertNull(outputValue.get("extra"));
          assertTrue((boolean)outputValue.get("valid"));
          assertEquals(Schema.Type.MAP, transformedValueSourceRecord.valueSchema().type());
      }
  
      @Test
      public void testIfItCanHandleEmptyKeyProps() {
          final Transformation<SourceRecord> transformationKey = new CamelTypeConverterTransform.Key<>();
  
          final Map<String, Object> props = new HashMap<>();
          props.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG, Map.class.getName());
  
          assertThrows(ConfigException.class, () -> transformationKey.configure(Collections.emptyMap()));
      }
  
  }