Hardened the tests for the new data mining components

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestAbstractParallelProcessorFinishing.java

@@ -6,6 +6,7 @@ import static org.junit.Assert.fail;
 
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.concurrent.Callable;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -59,10 +60,10 @@ public class TestAbstractParallelProcessorFinishing {
     private AbstractSimpleParallelProcessor<Integer, Integer> createProcessor(Collection<Processor<Integer>> receivers) {
         return new AbstractSimpleParallelProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers) {
             @Override
-            protected Runnable createInstruction(Integer partialElement) {
-                return new Runnable() {
+            protected Callable<Integer> createInstruction(Integer partialElement) {
+                return new Callable<Integer>() {
                     @Override
-                    public void run() {
+                    public Integer call() {
                         while (instructionIsWorking) {
                             try {
                                 Thread.sleep(100);
@@ -70,6 +71,7 @@ public class TestAbstractParallelProcessorFinishing {
                                 fail("Instruction was interrupted.");
                             }
                         }
+                        return 0;
                     }
                 };
             }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestAbstractParallelProcessorWithManySimpleInstructions.java

@@ -5,12 +5,12 @@ import static org.junit.Assert.assertThat;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.concurrent.Callable;
 
 import org.junit.Before;
 import org.junit.Test;
 
 import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
-import com.sap.sse.datamining.test.components.util.InputForwardingInstruction;
 import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
 
 public class TestAbstractParallelProcessorWithManySimpleInstructions {
@@ -37,8 +37,13 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
         receivers.add(receiver);
         processor = new AbstractSimpleParallelProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers) {
             @Override
-            protected Runnable createInstruction(Integer partialElement) {
-                return new InputForwardingInstruction<Integer>(partialElement, getResultReceivers());
+            protected Callable<Integer> createInstruction(final Integer element) {
+                return new Callable<Integer>() {
+                    @Override
+                    public Integer call() throws Exception {
+                        return element;
+                    }
+                };
             }
         };
     }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestAbstractProcessor.java

@@ -0,0 +1,47 @@
+package com.sap.sse.datamining.components;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.*;
+
+import java.util.Collection;
+import java.util.HashSet;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import com.sap.sse.datamining.impl.components.AbstractProcessor;
+
+public class TestAbstractProcessor {
+    
+    private Collection<Processor<Integer>> receivers;
+    private boolean receiverWasToldToFinish;
+    
+    @Before
+    public void initializeReceivers() {
+        Processor<Integer> receiver = new Processor<Integer>() {
+            @Override
+            public void onElement(Integer element) {
+            }
+            @Override
+            public void finish() throws InterruptedException {
+                receiverWasToldToFinish = true;
+            }
+        };
+        
+        receivers = new HashSet<>();
+        receivers.add(receiver);
+    }
+
+    @Test
+    public void testFinishing() throws InterruptedException {
+        Processor<Integer> processor = new AbstractProcessor<Integer, Integer>(receivers) {
+            @Override
+            protected Integer processElement(Integer element) {
+                return null;
+            }
+        };
+        processor.finish();
+        assertThat(receiverWasToldToFinish, is(true));
+    }
+
+}

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestFilteringProcessors.java

@@ -12,6 +12,7 @@ import java.util.Map;
 import org.junit.Before;
 import org.junit.Test;
 
+import com.sap.sse.datamining.impl.components.ParallelFilteringProcessor;
 import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
 
 public class TestFilteringProcessors {

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestParallelExtractionProcessor.java

@@ -30,6 +30,11 @@ public class TestParallelExtractionProcessor {
     
     private Function<Integer> getCrossSumFunction;
     
+    /**
+     * A function that is invalid with the type {@link Number} and will return <code>null</code> with every invocation.
+     */
+    private Function<Integer> invalidFunction;
+    
     @Before
     public void initializeReceivers() {
         receivedValues = new HashMap<>();
@@ -49,9 +54,12 @@ public class TestParallelExtractionProcessor {
     }
     
     @Before
-    public void initializeExtractionFunction() {
-        Method extractionMethod = FunctionTestsUtil.getMethodFromClass(Number.class, "getCrossSum");
-        getCrossSumFunction = FunctionFactory.createMethodWrappingFunction(extractionMethod);
+    public void initializeExtractionFunctions() {
+        Method getCrossSumMethod = FunctionTestsUtil.getMethodFromClass(Number.class, "getCrossSum");
+        getCrossSumFunction = FunctionFactory.createMethodWrappingFunction(getCrossSumMethod);
+        
+        Method invalidMethod = FunctionTestsUtil.getMethodFromSimpleClassWithMarkedMethod("sideEffectFreeValue");
+        invalidFunction = FunctionFactory.createMethodWrappingFunction(invalidMethod);
     }
 
     @Test
@@ -59,10 +67,27 @@ public class TestParallelExtractionProcessor {
         Processor<GroupedDataEntry<Number>> processor = new ParallelGroupedElementsValueExtractionProcessor<Number, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, getCrossSumFunction);
         Collection<GroupedDataEntry<Number>> elements = createElements();
         processElements(processor, elements);
+        ConcurrencyTestsUtil.sleepFor(100); //Giving the processor time to finish the instructions
+        
         Collection<GroupedDataEntry<Integer>> expectedReceivedValues = buildExpectedReceivedValues(elements);
         verifyThatExpectedElementsHaveBeenReceived(expectedReceivedValues);
     }
 
+    private void verifyThatExpectedElementsHaveBeenReceived(Collection<GroupedDataEntry<Integer>> expectedReceivedValues) {
+        for (GroupedDataEntry<Integer> expectedReceivedValue : expectedReceivedValues) {
+            assertThat("The expected value '" + expectedReceivedValue + "' wasn't received.", receivedValues.containsKey(expectedReceivedValue.getKey()), is(true));
+            assertThat(receivedValues.get(expectedReceivedValue.getKey()), is(expectedReceivedValue.getDataEntry()));
+        }
+    }
+    
+    @Test
+    public void testValueExtractionWithInvalidFunction() {
+        Processor<GroupedDataEntry<Number>> processor = new ParallelGroupedElementsValueExtractionProcessor<Number, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, invalidFunction);
+        processElements(processor, createElements());
+        ConcurrencyTestsUtil.sleepFor(100); //Giving the processor time to finish the instructions
+        assertThat("Values have been received, but the processor function is invalid.", receivedValues.isEmpty(), is(true));
+    }
+
     private Collection<GroupedDataEntry<Number>> createElements() {
         Collection<GroupedDataEntry<Number>> elements = new ArrayList<>();
         elements.add(new GroupedDataEntry<Number>(new GenericGroupKey<Integer>(1), new Number(1)));
@@ -88,11 +113,4 @@ public class TestParallelExtractionProcessor {
         return expectedReceivedValues;
     }
 
-    private void verifyThatExpectedElementsHaveBeenReceived(Collection<GroupedDataEntry<Integer>> expectedReceivedValues) {
-        for (GroupedDataEntry<Integer> expectedReceivedValue : expectedReceivedValues) {
-            assertThat("The expected value '" + expectedReceivedValue + "' wasn't received.", receivedValues.containsKey(expectedReceivedValue.getKey()), is(true));
-            assertThat(receivedValues.get(expectedReceivedValue.getKey()), is(expectedReceivedValue.getDataEntry()));
-        }
-    }
-
 }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestParallelMultiDimensionalGroupingProcessor.java

@@ -3,6 +3,7 @@ package com.sap.sse.datamining.components;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -10,6 +11,7 @@ import java.util.Collection;
 import org.junit.Before;
 import org.junit.Test;
 
+import com.sap.sse.datamining.factories.FunctionFactory;
 import com.sap.sse.datamining.functions.Function;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.ParallelMultiDimensionalGroupingProcessor;
@@ -19,10 +21,12 @@ import com.sap.sse.datamining.shared.impl.CompoundGroupKey;
 import com.sap.sse.datamining.shared.impl.GenericGroupKey;
 import com.sap.sse.datamining.test.components.util.Number;
 import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
+import com.sap.sse.datamining.test.util.FunctionTestsUtil;
 
 public class TestParallelMultiDimensionalGroupingProcessor {
     
     private Processor<Iterable<Number>> processor;
+    private Collection<Processor<GroupedDataEntry<Number>>> receivers;
 
     private GroupedDataEntry<Number> groupedElement;
 
@@ -37,13 +41,34 @@ public class TestParallelMultiDimensionalGroupingProcessor {
             public void finish() throws InterruptedException { }
         };
         
-        Collection<Processor<GroupedDataEntry<Number>>> receivers = new ArrayList<>();
+        receivers = new ArrayList<>();
         receivers.add(receiver);
+        
         Collection<Function<?>> dimensions = new ArrayList<Function<?>>();
         dimensions.add(new MethodWrappingFunction<>(Number.class.getMethod("getLength", new Class<?>[0]), int.class));
         dimensions.add(new MethodWrappingFunction<>(Number.class.getMethod("getCrossSum", new Class<?>[0]), int.class));
+        
         processor = new ParallelMultiDimensionalGroupingProcessor<Number>(ConcurrencyTestsUtil.getExecutor(), receivers, dimensions);
     }
+    
+    @Test(expected=IllegalArgumentException.class)
+    public void testConstructionWithNullDimensions() {
+        new ParallelMultiDimensionalGroupingProcessor<>(ConcurrencyTestsUtil.getExecutor(), receivers, null);
+    }
+    
+    @Test(expected=IllegalArgumentException.class)
+    public void testConstructionWithEmptyDimensions() {
+        Iterable<Function<?>> dimensions = new ArrayList<>();
+        new ParallelMultiDimensionalGroupingProcessor<>(ConcurrencyTestsUtil.getExecutor(), receivers, dimensions);
+    }
+    
+    @Test(expected=IllegalArgumentException.class)
+    public void testConstructionWithFunctionsInsteadOfDimensions() {
+        Collection<Function<?>> functions = new ArrayList<>();
+        Method method = FunctionTestsUtil.getMethodFromSimpleClassWithMarkedMethod("sideEffectFreeValue");
+        functions.add(FunctionFactory.createMethodWrappingFunction(method));
+        new ParallelMultiDimensionalGroupingProcessor<>(ConcurrencyTestsUtil.getExecutor(), receivers, functions);
+    }
 
     @Test
     public void testGroupKeyGeneration() {

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestPartitioningOfAbstractSimpleParallelProcessor.java

@@ -5,6 +5,7 @@ import static org.junit.Assert.assertThat;
 
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.concurrent.Callable;
 
 import org.junit.Test;
 
@@ -26,10 +27,11 @@ public class TestPartitioningOfAbstractSimpleParallelProcessor {
             }
 
             @Override
-            protected Runnable createInstruction(Integer partialElement) {
-                return new Runnable() {
+            protected Callable<Integer> createInstruction(Integer partialElement) {
+                return new Callable<Integer>() {
                     @Override
-                    public void run() {
+                    public Integer call() throws Exception {
+                        return 0;
                     }
                 };
             }