ESP-98ESP-98_SinkParameters - add unit tests

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

Author: kristoffSC

pom.xml

@@ -83,6 +83,7 @@ under the License.
     <wiremock.version>2.27.2</wiremock.version>
     <jacoco.plugin.version>0.8.7</jacoco.plugin.version>
     <maven.shade.plugin.version>3.1.1</maven.shade.plugin.version>
+    <mockito-inline.version>4.6.1</mockito-inline.version>
   </properties>
 
   <repositories>
@@ -264,6 +265,13 @@ under the License.
       <version>${wiremock.version}</version>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-inline</artifactId>
+      <version>${mockito-inline.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

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

@@ -1,6 +1,9 @@
 package com.getindata.connectors.http.internal.config;
 
-// TODO EXP-98 add Javadoc
+/**
+ * A Runtime exception throw when there is any issue with configuration properties for Http
+ * Connector.
+ */
 public class ConfigException extends RuntimeException {
 
     private static final long serialVersionUID = 1L;

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

@@ -2,15 +2,28 @@
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import lombok.experimental.UtilityClass;
 
-// TODO EXP-98 add Javadoc
+/**
+ * A dictionary class containing properties or properties prefixes for Http connector.
+ */
+@UtilityClass
 @NoArgsConstructor(access = AccessLevel.NONE)
 public final class HttpConnectorConfigConstants {
 
+    /**
+     * A property prefix for http connector.
+     */
     public static final String GID_CONNECTOR_HTTP = "gid.connector.http.";
 
+    /**
+     * A property prefix for http connector header properties
+     */
     public static final String SINK_HEADER_PREFIX = GID_CONNECTOR_HTTP + "sink.header.";
 
+    /**
+     * A property for Content-Type HTTP header.
+     */
     public static final String CONTENT_TYPE_HEADER = SINK_HEADER_PREFIX + "Content-Type";
 
 }

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

@@ -8,14 +8,15 @@
 import java.net.http.HttpRequest.Builder;
 import java.net.http.HttpResponse;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.annotation.VisibleForTesting;
 
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
@@ -42,22 +43,14 @@ public JavaNetSinkHttpClient(Properties properties) {
         Map<String, String> headerMap =
             ConfigUtils.propertiesToMap(properties, SINK_HEADER_PREFIX, String.class);
 
-        // TODO EXP-98 add tests
-        headersAndValues = headerMap
-            .entrySet()
-            .stream()
-            .flatMap(entry -> {
-                String originalKey = entry.getKey();
-                // TODO EXP-98 extract this to utils and add tests. Wrap with try/catch in Utils
-                String newKey = ConfigUtils.extractPropertyLastElement(originalKey);
-
-                return Stream.of(newKey, entry.getValue());
-            }).toArray(String[]::new);
+        // TODO ESP-98 add tests
+        headersAndValues = ConfigUtils.flatMapToHeaderArray(headerMap);
     }
 
     @Override
     public CompletableFuture<SinkHttpClientResponse> putRequests(
-            List<HttpSinkRequestEntry> requestEntries, String endpointUrl) {
+            List<HttpSinkRequestEntry> requestEntries,
+            String endpointUrl) {
         return submitRequests(requestEntries, endpointUrl).thenApply(
             this::prepareSinkHttpClientResponse);
     }
@@ -78,8 +71,8 @@ private HttpRequest buildHttpRequest(HttpSinkRequestEntry requestEntry, URI endp
     }
 
     private CompletableFuture<List<JavaNetHttpResponseWrapper>> submitRequests(
-        List<HttpSinkRequestEntry> requestEntries, String endpointUrl
-    ) {
+            List<HttpSinkRequestEntry> requestEntries,
+            String endpointUrl) {
         var endpointUri = URI.create(endpointUrl);
         var responseFutures = new ArrayList<CompletableFuture<JavaNetHttpResponseWrapper>>();
 
@@ -101,7 +94,7 @@ private CompletableFuture<List<JavaNetHttpResponseWrapper>> submitRequests(
     }
 
     private SinkHttpClientResponse prepareSinkHttpClientResponse(
-        List<JavaNetHttpResponseWrapper> responses) {
+            List<JavaNetHttpResponseWrapper> responses) {
         var successfulResponses = new ArrayList<HttpSinkRequestEntry>();
         var failedResponses = new ArrayList<HttpSinkRequestEntry>();
 
@@ -117,4 +110,9 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse(
 
         return new SinkHttpClientResponse(successfulResponses, failedResponses);
     }
+
+    @VisibleForTesting
+    String[] getHeadersAndValues() {
+        return Arrays.copyOf(headersAndValues, headersAndValues.length);
+    }
 }

src/main/java/com/getindata/connectors/http/internal/table/sink/HttpDynamicSink.java

@@ -115,7 +115,7 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
         var insertMethod = tableOptions.get(INSERT_METHOD);
         var contentType = getContentTypeFromFormat(tableOptions.get(FactoryUtil.FORMAT));
 
-        // TODO EXP-98 add headers to DDL and add tests for this
+        // TODO ESP-98 add headers to DDL and add tests for this
         HttpSinkBuilder<RowData> builder = HttpSink
             .<RowData>builder()
             .setEndpointUrl(tableOptions.get(URL))

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

@@ -3,6 +3,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Stream;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
@@ -78,6 +79,31 @@ public static String extractPropertyLastElement(String propertyKay) {
         return propertyKay.substring(delimiterLastIndex + 1);
     }
 
+    /**
+     * Flat map a given Map of header name and header value map to an array containing both header
+     * names and values. For example, header map of
+     * <pre>
+     *     Map.of(
+     *     header1, val1,
+     *     header2, val2
+     *     )
+     * </pre>
+     * will be converter to an array of:
+     * <pre>
+     *      String[] headers = {"header1", "val1", "header2", "val2"};
+     * </pre>
+     */
+    public static String[] flatMapToHeaderArray(Map<String, String> headerMap) {
+        return headerMap
+            .entrySet()
+            .stream()
+            .flatMap(entry -> {
+                String originalKey = entry.getKey();
+                String newKey = ConfigUtils.extractPropertyLastElement(originalKey);
+                return Stream.of(newKey, entry.getValue());
+            }).toArray(String[]::new);
+    }
+
     private static <T> void tryAddToConfigMap(
             Properties properties,
             Class<T> clazz, Map<String, T> map,

src/test/java/com/getindata/connectors/http/TestHelper.java

@@ -3,10 +3,13 @@
 import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.Objects;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import org.junit.jupiter.api.Assertions;
+import static org.assertj.core.api.Assertions.assertThat;
 
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public final class TestHelper {
@@ -22,4 +25,25 @@ public static String readTestFile(String pathToFile) {
         }
     }
 
+    public static void assertPropertyArray(
+            String[] headerArray,
+            String propertyName,
+            String expectedValue) {
+        // important thing is that we have property followed by its value.
+        for (int i = 0; i < headerArray.length; i++) {
+            if (headerArray[i].equals(propertyName)) {
+                assertThat(headerArray[i + 1])
+                    .withFailMessage("Property Array does not contain property name, value pairs.")
+                    .isEqualTo(expectedValue);
+                return;
+            }
+        }
+        Assertions.fail(
+            String.format(
+                "Missing property name [%s] in header array %s.",
+                propertyName,
+                Arrays.toString(headerArray))
+        );
+    }
+
 }

src/test/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClientTest.java

@@ -0,0 +1,90 @@
+package com.getindata.connectors.http.internal.sink.httpclient;
+
+import java.net.http.HttpClient;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
+import static com.getindata.connectors.http.TestHelper.assertPropertyArray;
+
+@ExtendWith(MockitoExtension.class)
+class JavaNetSinkHttpClientTest {
+
+    private static MockedStatic<HttpClient> httpClientStaticMock;
+
+    @Mock
+    private HttpClient.Builder httpClientBuilder;
+
+    @BeforeAll
+    public static void beforeAll() {
+        httpClientStaticMock = Mockito.mockStatic(HttpClient.class);
+    }
+
+    @AfterAll
+    public static void afterAll() {
+        if (httpClientStaticMock != null) {
+            httpClientStaticMock.close();
+        }
+    }
+
+    @BeforeEach
+    public void setUp() {
+        httpClientStaticMock.when(HttpClient::newBuilder).thenReturn(httpClientBuilder);
+        Mockito
+            .when(httpClientBuilder.followRedirects(Mockito.any()))
+            .thenReturn(httpClientBuilder);
+    }
+
+    @Test
+    public void shouldBuildClientWithNoHeader() {
+
+        JavaNetSinkHttpClient client = new JavaNetSinkHttpClient(new Properties());
+        assertThat(client.getHeadersAndValues()).isEmpty();
+    }
+
+    @Test
+    public void shouldBuildClientWithHeaders() {
+
+        // GIVEN
+        Properties properties = new Properties();
+        properties.setProperty("property", "val1");
+        properties.setProperty("my.property", "val2");
+        properties.setProperty(
+            HttpConnectorConfigConstants.SINK_HEADER_PREFIX + "Origin",
+            "https://developer.mozilla.org")
+        ;
+        properties.setProperty(
+            HttpConnectorConfigConstants.SINK_HEADER_PREFIX + "Cache-Control",
+            "no-cache, no-store, max-age=0, must-revalidate"
+        );
+        properties.setProperty(
+            HttpConnectorConfigConstants.SINK_HEADER_PREFIX + "Access-Control-Allow-Origin",
+            "*"
+        );
+
+        // WHEN
+        JavaNetSinkHttpClient client = new JavaNetSinkHttpClient(properties);
+        String[] headersAndValues = client.getHeadersAndValues();
+        assertThat(headersAndValues).hasSize(6);
+
+        // THEN
+        // assert that we have property followed by its value.
+        assertPropertyArray(headersAndValues, "Origin", "https://developer.mozilla.org");
+        assertPropertyArray(
+            headersAndValues,
+            "Cache-Control", "no-cache, no-store, max-age=0, must-revalidate"
+        );
+        assertPropertyArray(headersAndValues, "Access-Control-Allow-Origin", "*");
+    }
+
+}

src/test/java/com/getindata/connectors/http/internal/table/sink/HttpDynamicSinkTest.java

@@ -9,13 +9,31 @@
 import org.apache.flink.table.types.AtomicDataType;
 import org.apache.flink.table.types.logical.BooleanType;
 import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
+import com.getindata.connectors.http.internal.table.sink.HttpDynamicSink.HttpDynamicTableSinkBuilder;
 import static com.getindata.connectors.http.internal.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD;
 import static com.getindata.connectors.http.internal.table.sink.HttpDynamicSinkConnectorOptions.URL;
 
 public class HttpDynamicSinkTest {
 
+    @Test
+    public void testAsSummaryString() {
+        var mockFormat = new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
+
+        HttpDynamicSink dynamicSink = new HttpDynamicTableSinkBuilder()
+            .setTableOptions(new Configuration())
+            .setConsumedDataType(
+                new AtomicDataType(new BooleanType(false)))
+            .setEncodingFormat(mockFormat)
+            .setFormatContentTypeMap(Map.of())
+            .build();
+
+        assertThat(dynamicSink.asSummaryString()).isEqualTo("HttpSink");
+    }
+
     @Test
     public void copyEqualityTest() {
         var mockFormat = new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());