unit tests

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

flink-kubernetes-operator-api/pom.xml

@@ -145,6 +145,13 @@ under the License.
             <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <profiles>

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/SpecUtils.java

@@ -32,9 +32,8 @@
 
 import javax.annotation.Nullable;
 
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
-import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
 
@@ -132,7 +131,6 @@ public static <T> T clone(T object) {
         }
     }
 
-    // todo unit test
     public static JsonNode toJsonNode(Map<String, String> properties) {
         ObjectNode jsonNode = yamlObjectMapper.createObjectNode();
         for (Map.Entry<String, String> entry : properties.entrySet()) {
@@ -143,12 +141,12 @@ public static JsonNode toJsonNode(Map<String, String> properties) {
 
     public static Map<String, String> toStringMap(JsonNode node) {
         if (node == null) {
-            return Collections.emptyMap();
+            return new HashMap<>();
         }
         if (node instanceof NullNode) {
-            return Collections.emptyMap();
+            return new HashMap<>();
         }
-        Map<String, String> flatMap = new LinkedHashMap<>();
+        Map<String, String> flatMap = new HashMap<>();
         flattenHelper(node, "", flatMap);
         return flatMap;
     }

flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/utils/SpecUtilsTest.java

@@ -24,16 +24,22 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import org.junit.jupiter.api.Test;
 
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
 /** Test for {@link SpecUtils}. */
-public class SpecUtilsTest {
+class SpecUtilsTest {
+
+    private static final ObjectMapper yamlObjectMapper = new ObjectMapper(new YAMLFactory());
 
     @Test
-    public void testSpecSerializationWithVersion() throws JsonProcessingException {
+    void testSpecSerializationWithVersion() throws JsonProcessingException {
         FlinkDeployment app = BaseTestUtils.buildApplicationCluster();
         String serialized = SpecUtils.writeSpecWithMeta(app.getSpec(), app);
         ObjectNode node = (ObjectNode) new ObjectMapper().readTree(serialized);
@@ -56,7 +62,7 @@ public void testSpecSerializationWithVersion() throws JsonProcessingException {
     }
 
     @Test
-    public void testSpecSerializationWithoutGeneration() throws JsonProcessingException {
+    void testSpecSerializationWithoutGeneration() throws JsonProcessingException {
         // with regards to ReconcialiationMetadata & SpecWithMeta
         FlinkDeployment app = BaseTestUtils.buildApplicationCluster();
         app.getMetadata().setGeneration(12L);
@@ -76,4 +82,57 @@ public void testSpecSerializationWithoutGeneration() throws JsonProcessingExcept
         var migrated = SpecUtils.deserializeSpecWithMeta(oldSerialized, FlinkDeploymentSpec.class);
         assertNull(migrated.getMeta());
     }
+
+    @Test
+    void convertsStringMapToJsonNode() {
+        var map = Map.of("k1", "v1", "k2", "v2", "k3.nested", "v3");
+        var node = SpecUtils.mapToJsonNode(map);
+
+        assertThat(node).hasSize(3);
+        assertThat(node.get("k1").asText()).isEqualTo("v1");
+        assertThat(node.get("k2").asText()).isEqualTo("v2");
+        assertThat(node.get("k3.nested").asText()).isEqualTo("v3");
+    }
+
+    @Test
+    void convertsJsonNodeToMap() throws JsonProcessingException {
+        var node =
+                yamlObjectMapper.readTree("k1: v1 \n" + "k2: v2 \n" + "k3:\n" + "  nested: v3\n");
+
+        var map = SpecUtils.toStringMap(node);
+        assertThat(map).hasSize(3);
+        assertThat(map.get("k1")).isEqualTo("v1");
+        assertThat(map.get("k2")).isEqualTo("v2");
+        assertThat(map.get("k3.nested")).isEqualTo("v3");
+    }
+
+    @Test
+    void addConfigPropertyToSpec() {
+        var spec = new FlinkDeploymentSpec();
+
+        SpecUtils.addConfigProperty(spec, "k1", "v1");
+
+        assertThat(spec.getFlinkConfiguration().get("k1").asText()).isEqualTo("v1");
+    }
+
+    @Test
+    void addConfigPropertiesToSpec() {
+        var spec = new FlinkDeploymentSpec();
+
+        SpecUtils.addConfigProperties(spec, Map.of("k1", "v1", "k2", "v2"));
+
+        assertThat(spec.getFlinkConfiguration().get("k1").asText()).isEqualTo("v1");
+        assertThat(spec.getFlinkConfiguration().get("k2").asText()).isEqualTo("v2");
+    }
+
+    @Test
+    void removeConfigPropertiesFromSpec() {
+        var spec = new FlinkDeploymentSpec();
+        SpecUtils.addConfigProperties(spec, Map.of("k1", "v1", "k2", "v2"));
+
+        SpecUtils.removeConfigProperties(spec, "k1");
+
+        assertThat(spec.getFlinkConfiguration().get("k1")).isNull();
+        assertThat(spec.getFlinkConfiguration().get("k2").asText()).isEqualTo("v2");
+    }
 }