Add mapper test

snowflakedb · Dec 4, 2024 · 9c9182b · 9c9182b
1 parent 4bb5888
commit 9c9182b
Showing 1 changed file with 33 additions and 11 deletions.
diff --git a/...fka/connector/internal/streaming/schemaevolution/iceberg/IcebergColumnTypeMapperTest.java b/...fka/connector/internal/streaming/schemaevolution/iceberg/IcebergColumnTypeMapperTest.java
@@ -1,15 +1,19 @@
 package com.snowflake.kafka.connector.internal.streaming.schemaevolution.iceberg;
 
+import static com.snowflake.kafka.connector.internal.SnowflakeErrors.ERROR_5026;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException;
 import java.util.stream.Stream;
 import org.apache.kafka.connect.data.Date;
 import org.apache.kafka.connect.data.Decimal;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Time;
 import org.apache.kafka.connect.data.Timestamp;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -29,7 +33,26 @@ void shouldMapKafkaTypeToSnowflakeColumnType(
   @ParameterizedTest()
   @MethodSource("jsonNodeTypesToMap")
   void shouldMapJsonNodeTypeToKafkaType(JsonNode value, Schema.Type expectedKafkaType) {
-    assertThat(mapper.mapJsonNodeTypeToKafkaType(value)).isEqualTo(expectedKafkaType);
+    assertThat(mapper.mapJsonNodeTypeToKafkaType("test", value)).isEqualTo(expectedKafkaType);
+  }
+
+  @ParameterizedTest()
+  @MethodSource("jsonNodeValuesToThrowException")
+  void shouldThrowExceptionWhenMappingEmptyOrNullNode(JsonNode value) {
+    assertThatThrownBy(() -> mapper.mapJsonNodeTypeToKafkaType("test", value))
+        .isInstanceOf(SnowflakeKafkaConnectorException.class)
+        .hasMessageContaining("'test' field value is null or empty")
+        .matches(
+            e -> ((SnowflakeKafkaConnectorException) e).getCode().equals(ERROR_5026.getCode()));
+  }
+
+  @Test
+  void shouldThrowExceptionForNullValue() {
+    assertThatThrownBy(() -> mapper.mapJsonNodeTypeToKafkaType("test", null))
+        .isInstanceOf(SnowflakeKafkaConnectorException.class)
+        .hasMessageContaining("'test' field value is null or empty")
+        .matches(
+            e -> ((SnowflakeKafkaConnectorException) e).getCode().equals(ERROR_5026.getCode()));
   }
 
   private static Stream<Arguments> kafkaTypesToMap() {
@@ -52,16 +75,8 @@ private static Stream<Arguments> kafkaTypesToMap() {
         Arguments.of(Schema.Type.STRUCT, null, "OBJECT"));
   }
 
-  private static Stream<Arguments> kafkaTypesToThrowException() {
-    return Stream.of(
-        Arguments.of(Schema.Type.ARRAY),
-        Arguments.of(Schema.Type.MAP),
-        Arguments.of(Schema.Type.STRUCT));
-  }
-
   private static Stream<Arguments> jsonNodeTypesToMap() {
     return Stream.of(
-        Arguments.of(JsonNodeFactory.instance.nullNode(), Schema.Type.STRING),
         Arguments.of(JsonNodeFactory.instance.numberNode((short) 1), Schema.Type.INT64),
         Arguments.of(JsonNodeFactory.instance.numberNode(1), Schema.Type.INT64),
         Arguments.of(JsonNodeFactory.instance.numberNode(1L), Schema.Type.INT64),
@@ -70,7 +85,14 @@ private static Stream<Arguments> jsonNodeTypesToMap() {
         Arguments.of(JsonNodeFactory.instance.booleanNode(true), Schema.Type.BOOLEAN),
         Arguments.of(JsonNodeFactory.instance.textNode("text"), Schema.Type.STRING),
         Arguments.of(JsonNodeFactory.instance.binaryNode(new byte[] {1, 2, 3}), Schema.Type.BYTES),
-        Arguments.of(JsonNodeFactory.instance.arrayNode(), Schema.Type.ARRAY),
-        Arguments.of(JsonNodeFactory.instance.objectNode(), Schema.Type.STRUCT));
+        Arguments.of(JsonNodeFactory.instance.arrayNode().add(1), Schema.Type.ARRAY),
+        Arguments.of(JsonNodeFactory.instance.objectNode().put("test", 1), Schema.Type.STRUCT));
+  }
+
+  private static Stream<Arguments> jsonNodeValuesToThrowException() {
+    return Stream.of(
+        Arguments.of(JsonNodeFactory.instance.nullNode()),
+        Arguments.of(JsonNodeFactory.instance.arrayNode()),
+        Arguments.of(JsonNodeFactory.instance.objectNode()));
   }
 }