ESP-98ESP-98_SinkParameters - add unit tests

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Author: kristoffSC

src/main/java/com/getindata/connectors/http/internal/config/ConfigException.java

@@ -5,6 +5,10 @@ public class ConfigException extends RuntimeException {
 
     private static final long serialVersionUID = 1L;
 
+    public ConfigException(String message) {
+        super(message);
+    }
+
     public ConfigException(String message, Throwable t) {
         super(message, t);
     }

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClient.java

@@ -49,7 +49,7 @@ public JavaNetSinkHttpClient(Properties properties) {
             .flatMap(entry -> {
                 String originalKey = entry.getKey();
                 // TODO EXP-98 extract this to utils and add tests. Wrap with try/catch in Utils
-                String newKey = originalKey.substring(originalKey.lastIndexOf(".") + 1);
+                String newKey = ConfigUtils.extractPropertyLastElement(originalKey);
 
                 return Stream.of(newKey, entry.getValue());
             }).toArray(String[]::new);

src/main/java/com/getindata/connectors/http/internal/utils/ConfigUtils.java

@@ -6,24 +6,35 @@
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import org.apache.flink.util.StringUtils;
 
 import com.getindata.connectors.http.internal.config.ConfigException;
 
-// TODO EXP-98 add Javadoc and tests
 @NoArgsConstructor(access = AccessLevel.NONE)
 public final class ConfigUtils {
 
+    private static final String PROPERTY_NAME_DELIMITER = ".";
+
+    /**
+     * Convert properties that name starts with given {@code keyPrefix} to Map.
+     * Values for this property will be cast to {@code valueClazz} type.
+     *
+     * @param properties properties to extract keys from.
+     * @param keyPrefix prefix used to match property name with.
+     * @param valueClazz type to cast property values to.
+     * @return Map of propertyName to propertyValue.
+     */
     public static <T> Map<String, T> propertiesToMap(
             Properties properties,
             String keyPrefix,
-            Class<T> clazz) {
+            Class<T> valueClazz) {
 
         Map<String, T> map = new HashMap<>();
         for (Map.Entry<Object, Object> entry : properties.entrySet()) {
             if (entry.getKey() instanceof String) {
                 String key = (String) entry.getKey();
                 if (key.startsWith(keyPrefix)) {
-                    tryAddToConfigMap(properties, clazz, map, key);
+                    tryAddToConfigMap(properties, valueClazz, map, key);
                 }
             } else {
                 throw new ConfigException(
@@ -35,6 +46,38 @@ public static <T> Map<String, T> propertiesToMap(
         return map;
     }
 
+    /**
+     * A utility method to extract last element from property name. This method assumes property to
+     * be in format as <b>{@code this.is.my.property.name}</b>, using "dot" as a delimiter. For this
+     * example the returned value would be <b>{@code name}</b>.
+     *
+     * @param propertyKay Property name to extract the last element from.
+     * @return property last element or the property name if {@code propertyKey} parameter had no
+     * dot delimiter.
+     * @throws ConfigException when invalid property such as null, empty, blank, ended with dot was
+     *                         used.
+     */
+    public static String extractPropertyLastElement(String propertyKay) {
+        if (StringUtils.isNullOrWhitespaceOnly(propertyKay)) {
+            throw new ConfigException("Provided a property name that is null, empty or blank.");
+        }
+
+        if (!propertyKay.contains(PROPERTY_NAME_DELIMITER)) {
+            return propertyKay;
+        }
+
+        int delimiterLastIndex = propertyKay.lastIndexOf(PROPERTY_NAME_DELIMITER);
+        if (delimiterLastIndex == propertyKay.length() - 1) {
+            throw new ConfigException(
+                String.format(
+                    "Invalid property - %s. Property name should not end with property delimiter.",
+                    propertyKay)
+            );
+        }
+
+        return propertyKay.substring(delimiterLastIndex + 1);
+    }
+
     private static <T> void tryAddToConfigMap(
             Properties properties,
             Class<T> clazz, Map<String, T> map,

src/test/java/com/getindata/connectors/http/internal/utils/ConfigUtilsTest.java

@@ -0,0 +1,84 @@
+package com.getindata.connectors.http.internal.utils;
+
+import java.util.Map;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.getindata.connectors.http.internal.config.ConfigException;
+
+class ConfigUtilsTest {
+
+    @Test
+    public void shouldExtractPropertiesToMap() {
+        Properties properties = new Properties();
+        properties.setProperty("property", "val1");
+        properties.setProperty("my.property", "val2");
+        properties.setProperty("my.super.property", "val3");
+        properties.setProperty("my.property.detail", "val4");
+        properties.setProperty("my.property.extra", "val5");
+        properties.setProperty("another.my.property.extra", "val6");
+
+        Map<String, String> mappedProperties =
+            ConfigUtils.propertiesToMap(properties, "my.property", String.class);
+
+        assertThat(mappedProperties).hasSize(3);
+        assertThat(mappedProperties)
+            .containsExactlyEntriesOf(
+                Map.of(
+                    "my.property", "val2",
+                    "my.property.detail", "val4",
+                    "my.property.extra", "val5"
+                ));
+    }
+
+    @Test
+    public void shouldConvertNoProperty() {
+        Properties properties = new Properties();
+        properties.setProperty("property", "val1");
+        properties.setProperty("my.property", "val2");
+        properties.setProperty("my.super.property", "val3");
+
+        Map<String, String> mappedProperties =
+            ConfigUtils.propertiesToMap(properties, "my.custom", String.class);
+        assertThat(mappedProperties).isEmpty();
+    }
+
+    @Test
+    public void shouldHandleInvalidPropertyType() {
+
+        Properties properties = new Properties();
+        properties.put("a.property", 1);
+
+        // Should ignore "invalid" property since does not match the prefix
+        Map<String, String> mappedProperties =
+            ConfigUtils.propertiesToMap(properties, "my.custom", String.class);
+        assertThat(mappedProperties).isEmpty();
+
+        // should throw on invalid value, when name matches the prefix.
+        assertThatThrownBy(
+            () -> ConfigUtils.propertiesToMap(properties, "a.property", String.class))
+            .isInstanceOf(ConfigException.class);
+    }
+
+    @ParameterizedTest(name = "Property full name - {0}")
+    @ValueSource(strings = {"property", "my.property", "my.super.property", ".my.super.property"})
+    public void shouldGetPropertyName(String fullPropertyName) {
+
+        String propertyLastElement = ConfigUtils.extractPropertyLastElement(fullPropertyName);
+        assertThat(propertyLastElement).isEqualTo("property");
+    }
+
+    @ParameterizedTest(name = "Property full name - {0}")
+    @ValueSource(strings = {"", " ", "my.super.property.", ".", "..."})
+    public void shouldThrowOnInvalidProperty(String invalidProperty) {
+
+        assertThatThrownBy(
+            () -> ConfigUtils.extractPropertyLastElement(invalidProperty))
+            .isInstanceOf(ConfigException.class);
+    }
+}