fix(expressions): fix equals SCeL expression should support null argument (#187)

Resolved: #187

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/TypedValue.java

@@ -33,6 +33,10 @@
  */
 public class TypedValue implements GettableByType {
 
+    public static TypedValue none() {
+        return new TypedValue(Schema.none(), null);
+    }
+
     private final Object value;
     private final SchemaSupplier schema;
 

connect-file-pulse-expression/src/main/java/io/streamthoughts/kafka/connect/filepulse/expression/function/ExecutionContext.java

@@ -44,23 +44,44 @@ void addArgument(final String name, final int index, final TypedValue value) {
         this.argumentByIndex.put(index, value);
     }
 
+    /**
+     * Retrieves the argument to which the specified index is mapped.
+     *
+     * @param index the argument index.
+     * @return      the {@link TypedValue}.
+     * @throws      IndexOutOfBoundsException if the given index is out of range.
+     */
     public TypedValue get(final int index) {
-        return Optional.ofNullable(argumentByIndex.get(index)).orElseThrow(IndexOutOfBoundsException::new);
+        if (!argumentByIndex.containsKey(index)) {
+            throw new IndexOutOfBoundsException(index);
+        }
+        return Optional.ofNullable(argumentByIndex.get(index)).orElse(TypedValue.none());
     }
 
+    /**
+     * Retrieves the argument to which the specified name is mapped.
+     *
+     * @param name  the argument name.
+     * @return      the {@link TypedValue}.
+     */
     public TypedValue get(final String name) {
-        return argumentByName.get(name);
+        return Optional.ofNullable(argumentByName.get(name)).orElse(TypedValue.none());
     }
 
     public List<TypedValue> get(final int index, final int to) {
         return values().subList(index, to);
     }
 
-
+    /**
+     * @return the number of arguments.
+     */
     public int size() {
         return argumentByName.size();
     }
 
+    /**
+     * @return values for all arguments.
+     */
     public List<TypedValue> values() {
         return new ArrayList<>(argumentByIndex.values());
     }

connect-file-pulse-expression/src/main/java/io/streamthoughts/kafka/connect/filepulse/expression/function/conditions/Equals.java

@@ -61,6 +61,12 @@ public TypedValue invoke(final ExecutionContext context) throws ExpressionExcept
                 final TypedValue left = context.get(OBJECT_L_ARG);
                 final TypedValue right = context.get(OBJECT_R_ARG);
 
+                if (left.isNull() && right.isNull())
+                    return TypedValue.bool(true);
+
+                if (left.isNull() || right.isNull())
+                    return TypedValue.bool(false);
+
                 // attempt to convert argument value to the field value before applying equality.
                 final Object leftValue = left.value();
                 final Object rightValue = right.as(left.type()).value();

connect-file-pulse-expression/src/test/java/io/streamthoughts/kafka/connect/filepulse/expression/function/FunctionsTest.java

@@ -134,12 +134,6 @@ public void should_execute_extract_array() {
         Assert.assertEquals("two", expression.readValue(context, TypedValue.class).getString());
     }
 
-    @Test
-    public void should_execute_equals_function() {
-        Expression expression = parseExpression("{{ equals('foo', 'foo') }}");
-        Assert.assertTrue(expression.readValue(EMPTY_CONTEXT, TypedValue.class).value());
-    }
-
     @Test
     public void should_execute_end_with_function() {
         Expression expressionTrue = parseExpression("{{ ends_with('FOO-suffix', 'suffix') }}");

connect-file-pulse-expression/src/test/java/io/streamthoughts/kafka/connect/filepulse/expression/function/conditions/EqualsTest.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2019-2020 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamthoughts.kafka.connect.filepulse.expression.function.conditions;
+
+import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
+import io.streamthoughts.kafka.connect.filepulse.expression.Expression;
+import io.streamthoughts.kafka.connect.filepulse.expression.StandardEvaluationContext;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static io.streamthoughts.kafka.connect.filepulse.expression.parser.ExpressionParsers.parseExpression;
+
+public class EqualsTest {
+
+    private static final StandardEvaluationContext EMPTY_CONTEXT  = new StandardEvaluationContext(new Object());
+
+    @Test
+    public void test_equals_given_two_equals_strings() {
+        // Given
+        Expression expression = parseExpression("{{ equals('foo', 'foo') }}");
+
+        // When
+        final TypedValue result = expression.readValue(EMPTY_CONTEXT, TypedValue.class);
+
+        // Then
+        Assert.assertTrue(result.value());
+    }
+
+    @Test
+    public void test_equals_return_false_given_a_null_value() {
+        // Given
+        Expression expression = parseExpression("{{ equals($value, 'foo') }}");
+
+        final StandardEvaluationContext context = new StandardEvaluationContext(TypedValue.string(null));
+
+        // When
+        final TypedValue result = expression.readValue(context, TypedValue.class);
+
+        // Then
+        Assert.assertFalse(result.value());
+    }
+
+    @Test
+    public void test_equals_return_true_given_null_values() {
+        // Given
+        Expression expression = parseExpression("{{ equals($value, null) }}");
+
+        final StandardEvaluationContext context = new StandardEvaluationContext(TypedValue.string(null));
+
+        // When
+        final TypedValue result = expression.readValue(context, TypedValue.class);
+
+        // Then
+        Assert.assertTrue(result.value());
+    }
+}