main - add test for passing properties to format used by lookup query creator.

Signed-off-by: Krzysztof Chmielewski <krzysiek.chmielewski@gmail.com>

Author: kristoffSC

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

@@ -0,0 +1,46 @@
+package com.getindata.connectors.http.internal.table.lookup.querycreators;
+
+import java.util.Collections;
+import java.util.Set;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory.Context;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.factories.TestFormatFactory.EncodingFormatMock;
+
+public class CustomFormatFactory implements SerializationFormatFactory {
+
+    public static final String IDENTIFIER = "query-creator-test-format";
+    public static final String REQUIRED_OPTION = "required-option-one";
+    static boolean requiredOptionsWereUsed = false;
+
+    @Override
+    public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
+        Context context,
+        ReadableConfig readableConfig) {
+        FactoryUtil.validateFactoryOptions(this, readableConfig);
+        return new EncodingFormatMock(",");
+    }
+
+    @Override
+    public String factoryIdentifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public Set<ConfigOption<?>> requiredOptions() {
+        requiredOptionsWereUsed = true;
+        return Set.of(ConfigOptions.key(REQUIRED_OPTION).stringType().noDefaultValue());
+    }
+
+    @Override
+    public Set<ConfigOption<?>> optionalOptions() {
+        return Collections.emptySet();
+    }
+}

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

@@ -0,0 +1,61 @@
+package com.getindata.connectors.http.internal.table.lookup.querycreators;
+
+import java.util.List;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.getindata.connectors.http.internal.table.lookup.LookupRow;
+import com.getindata.connectors.http.internal.table.lookup.RowDataSingleValueLookupSchemaEntry;
+import static com.getindata.connectors.http.internal.table.lookup.HttpLookupTableSourceFactory.row;
+
+class GenericJsonQueryCreatorFactoryTest {
+
+    private Configuration config;
+    private LookupRow lookupRow;
+
+    @BeforeEach
+    public void setUp() {
+        this.config = new Configuration();
+        this.lookupRow = new LookupRow();
+        lookupRow = new LookupRow()
+            .addLookupEntry(
+                new RowDataSingleValueLookupSchemaEntry(
+                    "key1",
+                    RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)
+                ));
+
+        lookupRow.setLookupPhysicalRowDataType(
+            row(List.of(
+                DataTypes.FIELD("key1", DataTypes.STRING())
+            )));
+
+        CustomFormatFactory.requiredOptionsWereUsed = false;
+    }
+
+    @Test
+    public void shouldPassPropertiesToQueryCreatorFormat() {
+        assertThat(CustomFormatFactory.requiredOptionsWereUsed)
+            .withFailMessage(
+                "CustomFormatFactory was not cleared, "
+                    + "make sure `CustomFormatFactory.requiredOptionsWereUsed = false` "
+                    + "was called before this test execution.")
+            .isFalse();
+
+        this.config.setString("lookup-request.format", CustomFormatFactory.IDENTIFIER);
+        this.config.setString(
+            String.format("lookup-request.format.%s.%s", CustomFormatFactory.IDENTIFIER,
+                CustomFormatFactory.REQUIRED_OPTION), "optionValue");
+
+        new GenericJsonQueryCreatorFactory().createLookupQueryCreator(
+            config,
+            lookupRow
+        );
+
+        assertThat(CustomFormatFactory.requiredOptionsWereUsed).isTrue();
+    }
+}

src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory

@@ -1 +1,2 @@
-com.getindata.connectors.http.TestPostRequestCallbackFactory
+com.getindata.connectors.http.TestPostRequestCallbackFactory
+com.getindata.connectors.http.internal.table.lookup.querycreators.CustomFormatFactory