http202 add more unit tests for coverage

Signed-off-by: davidradl <[email protected]>

Author: davidradl

src/test/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClientTest.java

@@ -1,5 +1,6 @@
 package com.getindata.connectors.http.internal.table.lookup;
 
+import java.io.IOException;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
@@ -305,4 +306,187 @@ public void shouldHandleEmptyCollectionInRowDataCollector() throws Configuration
         assertThat(result).isEmpty();
     }
 
+    @Test
+    public void shouldDeserializeArrayWithValidObjects() throws Exception {
+        // GIVEN
+        DeserializationSchema<RowData> mockDecoder = new DeserializationSchema<RowData>() {
+            @Override
+            public RowData deserialize(byte[] message) throws IOException {
+                return null;
+            }
+
+            @Override
+            public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
+                String msg = new String(message);
+                if (msg.contains("value1")) {
+                    out.collect(GenericRowData.of(StringData.fromString("row1")));
+                } else if (msg.contains("value2")) {
+                    out.collect(GenericRowData.of(StringData.fromString("row2")));
+                }
+                out.close();
+            }
+
+            @Override
+            public boolean isEndOfStream(RowData nextElement) {
+                return false;
+            }
+
+            @Override
+            public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProducedType() {
+                return null;
+            }
+        };
+
+        Properties properties = new Properties();
+        properties.setProperty(HttpConnectorConfigConstants.RESULT_TYPE, "array");
+
+        HttpLookupConfig lookupConfig = HttpLookupConfig.builder()
+            .url(BASE_URL)
+            .properties(properties)
+            .build();
+
+        JavaNetHttpPollingClient client = new JavaNetHttpPollingClient(
+            httpClient,
+            mockDecoder,
+            lookupConfig,
+            new GetRequestFactory(
+                new GenericGetQueryCreator(lookupRow),
+                headerPreprocessor,
+                lookupConfig
+            )
+        );
+
+        // WHEN
+        java.lang.reflect.Method method = JavaNetHttpPollingClient.class
+            .getDeclaredMethod("deserializeArray", byte[].class);
+        method.setAccessible(true);
+
+        String jsonArray = "[{\"key\":\"value1\"},{\"key\":\"value2\"}]";
+        List<RowData> result = (List<RowData>) method.invoke(client, jsonArray.getBytes());
+
+        // THEN
+        assertThat(result).isNotNull();
+        assertThat(result).hasSize(2);
+    }
+
+    @Test
+    public void shouldHandleNullNodesInArray() throws Exception {
+        // GIVEN
+        DeserializationSchema<RowData> mockDecoder = new DeserializationSchema<RowData>() {
+            @Override
+            public RowData deserialize(byte[] message) throws IOException {
+                return null;
+            }
+
+            @Override
+            public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
+                out.collect(GenericRowData.of(StringData.fromString("valid")));
+                out.close();
+            }
+
+            @Override
+            public boolean isEndOfStream(RowData nextElement) {
+                return false;
+            }
+
+            @Override
+            public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProducedType() {
+                return null;
+            }
+        };
+
+        Properties properties = new Properties();
+        properties.setProperty(HttpConnectorConfigConstants.RESULT_TYPE, "array");
+
+        HttpLookupConfig lookupConfig = HttpLookupConfig.builder()
+            .url(BASE_URL)
+            .properties(properties)
+            .build();
+
+        JavaNetHttpPollingClient client = new JavaNetHttpPollingClient(
+            httpClient,
+            mockDecoder,
+            lookupConfig,
+            new GetRequestFactory(
+                new GenericGetQueryCreator(lookupRow),
+                headerPreprocessor,
+                lookupConfig
+            )
+        );
+
+        // WHEN
+        java.lang.reflect.Method method = JavaNetHttpPollingClient.class
+            .getDeclaredMethod("deserializeArray", byte[].class);
+        method.setAccessible(true);
+
+        String jsonArray = "[{\"key\":\"value1\"},null,{\"key\":\"value2\"}]";
+        List<RowData> result = (List<RowData>) method.invoke(client, jsonArray.getBytes());
+
+        // THEN - null nodes should be skipped
+        assertThat(result).isNotNull();
+        assertThat(result).hasSize(2);
+    }
+
+    @Test
+    public void shouldHandleEmptyDeserializationInArray() throws Exception {
+        // GIVEN
+        DeserializationSchema<RowData> mockDecoder = new DeserializationSchema<RowData>() {
+            @Override
+            public RowData deserialize(byte[] message) throws IOException {
+                return null;
+            }
+
+            @Override
+            public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
+                String msg = new String(message);
+                // Only collect for specific messages, return empty for others
+                if (msg.contains("\"status\":\"valid\"")) {
+                    out.collect(GenericRowData.of(StringData.fromString("data")));
+                }
+                // Don't collect anything for other messages
+                out.close();
+            }
+
+            @Override
+            public boolean isEndOfStream(RowData nextElement) {
+                return false;
+            }
+
+            @Override
+            public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProducedType() {
+                return null;
+            }
+        };
+
+        Properties properties = new Properties();
+        properties.setProperty(HttpConnectorConfigConstants.RESULT_TYPE, "array");
+
+        HttpLookupConfig lookupConfig = HttpLookupConfig.builder()
+            .url(BASE_URL)
+            .properties(properties)
+            .build();
+
+        JavaNetHttpPollingClient client = new JavaNetHttpPollingClient(
+            httpClient,
+            mockDecoder,
+            lookupConfig,
+            new GetRequestFactory(
+                new GenericGetQueryCreator(lookupRow),
+                headerPreprocessor,
+                lookupConfig
+            )
+        );
+
+        // WHEN
+        java.lang.reflect.Method method = JavaNetHttpPollingClient.class
+            .getDeclaredMethod("deserializeArray", byte[].class);
+        method.setAccessible(true);
+
+        String jsonArray = "[{\"status\":\"invalid\"},{\"status\":\"valid\"},{\"status\":\"invalid\"}]";
+        List<RowData> result = (List<RowData>) method.invoke(client, jsonArray.getBytes());
+
+        // THEN - only valid deserialization should be included
+        assertThat(result).isNotNull();
+        assertThat(result).hasSize(1);
+    }
 }