http202 add more unit tests for coverage

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

Author: davidradl

src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java

@@ -271,20 +271,37 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
 
     private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
         List<RowData> result = new ArrayList<>();
-        responseBodyDecoder.deserialize(rawBytes, new Collector<>() {
-            @Override
-            public void collect(RowData record) {
-                result.add(record);
-            }
-
-            @Override
-            public void close() {
-                // No-op
-            }
-        });
+        responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
         return result;
     }
 
+    @VisibleForTesting
+    Collector<RowData> createRowDataCollector(List<RowData> result) {
+        return new RowDataCollector(result);
+    }
+
+    /**
+     * A simple collector implementation that adds RowData records to a list.
+     */
+    @VisibleForTesting
+    static class RowDataCollector implements Collector<RowData> {
+        private final List<RowData> result;
+
+        RowDataCollector(List<RowData> result) {
+            this.result = result;
+        }
+
+        @Override
+        public void collect(RowData record) {
+            result.add(record);
+        }
+
+        @Override
+        public void close() {
+            // No-op - nothing to clean up
+        }
+    }
+
     private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
         List<JsonNode> rawObjects =
             objectMapper.readValue(rawBytes, new TypeReference<>() {

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

@@ -1,6 +1,5 @@
 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;
@@ -228,42 +227,12 @@ public void shouldBuildClientWithHeaders() throws ConfigurationException {
     }
 
     @Test
-    public void shouldCallCloseOnCollectorInDeserializeSingleValue() throws Exception {
+    public void shouldCollectRowDataInCollector() throws ConfigurationException {
         // GIVEN
-        // Track if close() and collect() were called
-        final int[] collectCount = {0};
-        final boolean[] closeCalled = {false};
-
-        // Create a mock DeserializationSchema that calls both collect() and close()
-        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 {
-                // Simulate adding multiple records
-                out.collect(GenericRowData.of(StringData.fromString("test1")));
-                out.collect(GenericRowData.of(StringData.fromString("test2")));
-                // Call close() to trigger the code path we want to test
-                out.close();
-            }
-
-            @Override
-            public boolean isEndOfStream(RowData nextElement) {
-                return false;
-            }
-
-            @Override
-            public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProducedType() {
-                return null;
-            }
-        };
-
+        List<RowData> result = new ArrayList<>();
         JavaNetHttpPollingClient client = new JavaNetHttpPollingClient(
             httpClient,
-            mockDecoder,
+            decoder,
             options,
             new GetRequestFactory(
                 new GenericGetQueryCreator(lookupRow),
@@ -272,56 +241,53 @@ public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProduced
             )
         );
 
-        // WHEN
-        // Call deserializeSingleValue through reflection to test it directly
-        java.lang.reflect.Method method = JavaNetHttpPollingClient.class
-            .getDeclaredMethod("deserializeSingleValue", byte[].class);
-        method.setAccessible(true);
+        Collector<RowData> collector = client.createRowDataCollector(result);
 
-        byte[] testData = "{\"test\":\"value\"}".getBytes();
-        List<RowData> result = (List<RowData>) method.invoke(client, testData);
+        RowData row1 = GenericRowData.of(StringData.fromString("test1"));
+        RowData row2 = GenericRowData.of(StringData.fromString("test2"));
+
+        // WHEN
+        collector.collect(row1);
+        collector.collect(row2);
 
         // THEN
-        // Verify that the method executed successfully and returned results
-        assertThat(result).isNotNull();
         assertThat(result).hasSize(2);
-        assertThat(result.get(0).getString(0).toString()).isEqualTo("test1");
-        assertThat(result.get(1).getString(0).toString()).isEqualTo("test2");
-
-        // Note: The close() method in the anonymous Collector is a no-op,
-        // but this test ensures it's called and doesn't throw exceptions
+        assertThat(result.get(0)).isEqualTo(row1);
+        assertThat(result.get(1)).isEqualTo(row2);
     }
 
     @Test
-    public void shouldHandleEmptyResultInDeserializeSingleValue() throws Exception {
+    public void shouldCallCloseOnRowDataCollectorWithoutException() throws ConfigurationException {
         // GIVEN
-        // Create a mock DeserializationSchema that doesn't add any records but calls close()
-        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 {
-                // Don't collect anything, just close
-                out.close();
-            }
-
-            @Override
-            public boolean isEndOfStream(RowData nextElement) {
-                return false;
-            }
-
-            @Override
-            public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProducedType() {
-                return null;
-            }
-        };
+        List<RowData> result = new ArrayList<>();
+        JavaNetHttpPollingClient client = new JavaNetHttpPollingClient(
+            httpClient,
+            decoder,
+            options,
+            new GetRequestFactory(
+                new GenericGetQueryCreator(lookupRow),
+                headerPreprocessor,
+                options
+            )
+        );
+
+        Collector<RowData> collector = client.createRowDataCollector(result);
+        collector.collect(GenericRowData.of(StringData.fromString("test")));
+
+        // WHEN - close should not throw any exception
+        collector.close();
+
+        // THEN
+        assertThat(result).hasSize(1);
+    }
 
+    @Test
+    public void shouldHandleEmptyCollectionInRowDataCollector() throws ConfigurationException {
+        // GIVEN
+        List<RowData> result = new ArrayList<>();
         JavaNetHttpPollingClient client = new JavaNetHttpPollingClient(
             httpClient,
-            mockDecoder,
+            decoder,
             options,
             new GetRequestFactory(
                 new GenericGetQueryCreator(lookupRow),
@@ -330,16 +296,12 @@ public org.apache.flink.api.common.typeinfo.TypeInformation<RowData> getProduced
             )
         );
 
-        // WHEN
-        java.lang.reflect.Method method = JavaNetHttpPollingClient.class
-            .getDeclaredMethod("deserializeSingleValue", byte[].class);
-        method.setAccessible(true);
+        Collector<RowData> collector = client.createRowDataCollector(result);
 
-        byte[] testData = "{}".getBytes();
-        List<RowData> result = (List<RowData>) method.invoke(client, testData);
+        // WHEN - close without collecting anything
+        collector.close();
 
         // THEN
-        assertThat(result).isNotNull();
         assertThat(result).isEmpty();
     }