This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch
2374-provide-a-framework-to-easily-unit-test-processing-elements
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/2374-provide-a-framework-to-easily-unit-test-processing-elements by
this push:
new eea898c09 feat(#2374): Add suggestion to test processing elements
eea898c09 is described below
commit eea898c0905c1606dc10d4b216c2d498a68f0430
Author: Philipp Zehnder <[email protected]>
AuthorDate: Wed Dec 27 11:07:00 2023 +0100
feat(#2374): Add suggestion to test processing elements
---
.../streampipes-processors-filters-jvm/pom.xml | 18 +++
.../numericalfilter/NumericalFilterProcessor.java | 6 +-
.../NumericalFilterProcessorTest.java | 81 +++++++++++
.../ProcessingElementTestExecutor.java | 159 +++++++++++++++++++++
.../template/PipelineElementTemplateVisitor.java | 5 +-
5 files changed, 265 insertions(+), 4 deletions(-)
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
b/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
index 28f71ab1a..be6657ad8 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
@@ -43,12 +43,30 @@
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
</dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-test-utils</artifactId>
<version>0.95.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-pipeline-management</artifactId>
+ <version>0.95.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessor.java
index 14062127c..5f9a030d5 100644
---
a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessor.java
@@ -39,9 +39,9 @@ import org.apache.streampipes.sdk.utils.Assets;
public class NumericalFilterProcessor implements IStreamPipesDataProcessor {
- private static final String NUMBER_MAPPING = "number-mapping";
- private static final String VALUE = "value";
- private static final String OPERATION = "operation";
+ protected static final String NUMBER_MAPPING = "number-mapping";
+ protected static final String VALUE = "value";
+ protected static final String OPERATION = "operation";
private double threshold;
private NumericalOperator numericalOperator;
diff --git
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessorTest.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessorTest.java
new file mode 100644
index 000000000..6846f0074
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilterProcessorTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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
org.apache.streampipes.processors.filters.jvm.processor.numericalfilter;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.Map;
+
+public class NumericalFilterProcessorTest {
+
+ private static final String PROPERTY_NAME = "proptertyName";
+ private NumericalFilterProcessor processor;
+
+
+ @Before
+ public void setup() {
+ processor = new NumericalFilterProcessor();
+ }
+
+ @Test
+ public void testLowerThenOperatorFilterNotApplied() {
+
+ Map<String, Object> userConfiguration = Map.of(
+ // TODO Here we need a better solution for the mapping to deal with
the selector prefix
+ NumericalFilterProcessor.NUMBER_MAPPING, "::" + PROPERTY_NAME,
+ NumericalFilterProcessor.VALUE, 10.0,
+ NumericalFilterProcessor.OPERATION, "<"
+ );
+
+ List<Map<String, Object>> inputEvents = List.of(
+ Map.of(PROPERTY_NAME, 1.0f)
+ );
+
+ List<Map<String, Object>> outputEvents = List.of(
+ Map.of(PROPERTY_NAME, 1.0f)
+ );
+
+
+ ProcessingElementTestExecutor.run(processor, userConfiguration,
inputEvents, outputEvents);
+ }
+
+ @Test
+ public void testLowerThenOperatorFilterApplied() {
+
+ Map<String, Object> userConfiguration = Map.of(
+ // TODO Here we need a better solution for the mapping to deal with
the selector prefix
+ NumericalFilterProcessor.NUMBER_MAPPING, "::" + PROPERTY_NAME,
+ NumericalFilterProcessor.VALUE, 10.0,
+ NumericalFilterProcessor.OPERATION, "<"
+ );
+
+ List<Map<String, Object>> inputEvents = List.of(
+ Map.of(PROPERTY_NAME, 11.0f)
+ );
+
+ List<Map<String, Object>> outputEvents = List.of();
+
+
+ ProcessingElementTestExecutor.run(processor, userConfiguration,
inputEvents, outputEvents);
+ }
+
+
+}
\ No newline at end of file
diff --git
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/ProcessingElementTestExecutor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/ProcessingElementTestExecutor.java
new file mode 100644
index 000000000..3426b674d
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/ProcessingElementTestExecutor.java
@@ -0,0 +1,159 @@
+/*
+ * 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
org.apache.streampipes.processors.filters.jvm.processor.numericalfilter;
+
+import org.apache.streampipes.extensions.api.pe.IStreamPipesDataProcessor;
+import org.apache.streampipes.extensions.api.pe.param.IDataProcessorParameters;
+import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
+import org.apache.streampipes.manager.template.DataProcessorTemplateHandler;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.template.PipelineElementTemplate;
+import org.apache.streampipes.model.template.PipelineElementTemplateConfig;
+import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
+
+import org.mockito.ArgumentCaptor;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class ProcessingElementTestExecutor {
+
+ /**
+ * This method is used to run a data processor with a given configuration
and a list of input events.
+ * It then verifies the output events against the expected output events.
+ *
+ * @param processor The data processor under test.
+ * @param userConfiguration The user input configuration for the
processor.
+ * @param inputEvents The list of input events to be processed.
+ * @param expectedOutputEvents The list of expected output events.
+ */
+ public static void run(
+ IStreamPipesDataProcessor processor,
+ Map<String, Object> userConfiguration,
+ List<Map<String, Object>> inputEvents,
+ List<Map<String, Object>> expectedOutputEvents
+ ) {
+
+ // initialize the extractor with the provided configuration of the user
input
+ var e = getProcessingElementParameterExtractor(processor,
userConfiguration);
+ var mockParams = mock(IDataProcessorParameters.class);
+ when(mockParams.extractor()).thenReturn(e);
+
+
+ // calls the onPipelineStarted method of the processor to initialize it
+ processor.onPipelineStarted(mockParams, null, null);
+
+ // mock the output collector to capture the output events and validate the
results later
+ var mockCollector = mock(SpOutputCollector.class);
+ var spOutputCollectorCaptor = ArgumentCaptor.forClass(Event.class);
+
+
+ // Iterate over all input events and call the onEvent method of the
processor
+ for (Map<String, Object> inputRawEvent : inputEvents) {
+ processor.onEvent(getEvent(inputRawEvent), mockCollector);
+ }
+
+ // Validate the output of the processor
+ verify(mockCollector,
times(expectedOutputEvents.size())).collect(spOutputCollectorCaptor.capture());
+ var resultingEvents = spOutputCollectorCaptor.getAllValues();
+ IntStream.range(0, expectedOutputEvents.size())
+ .forEach(i -> assertEquals(
+ expectedOutputEvents.get(i),
+ resultingEvents.get(i)
+ .getRaw()
+ ));
+
+ // vlaidate that the processor is stopped correctly
+ processor.onPipelineStopped();
+ }
+
+ private static ProcessingElementParameterExtractor
getProcessingElementParameterExtractor(
+ IStreamPipesDataProcessor processor,
+ Map<String, Object> userConfiguration
+ ) {
+
+ var dataProcessorInvocation = getProcessorInvocation(processor,
userConfiguration);
+
+ return ProcessingElementParameterExtractor.from(dataProcessorInvocation);
+ }
+
+ private static DataProcessorInvocation getProcessorInvocation(
+ IStreamPipesDataProcessor processor,
+ Map<String, Object> userConfiguration
+ ) {
+ var pipelineElementTemplate = getPipelineElementTemplate(processor,
userConfiguration);
+
+ var invocation = new DataProcessorInvocation(
+ processor
+ .declareConfig()
+ .getDescription()
+ );
+
+ return new DataProcessorTemplateHandler(
+ pipelineElementTemplate,
+ invocation,
+ true
+ )
+ .applyTemplateOnPipelineElement();
+ }
+
+ private static PipelineElementTemplate getPipelineElementTemplate(
+ IStreamPipesDataProcessor processor,
+ Map<String, Object> userConfiguration
+ ) {
+ var staticProperties = processor
+ .declareConfig()
+ .getDescription()
+ .getStaticProperties();
+
+
+ var configs = new HashMap<String, PipelineElementTemplateConfig>();
+
+ staticProperties.forEach(staticProperty -> {
+ var value = userConfiguration.get(staticProperty.getInternalName());
+ configs.put(
+ staticProperty.getInternalName(),
+ new PipelineElementTemplateConfig(true, true, value)
+ );
+ });
+
+ return new PipelineElementTemplate("name", "description", configs);
+ }
+
+ private static Event getEvent(Map<String, Object> rawEvent) {
+ // TODO here we need a better solution to deal with the event schema and
the selector prefix
+ var sourceInfo = new SourceInfo("", "");
+ var schemaInfo = new SchemaInfo(null, new ArrayList<>());
+
+ return EventFactory.fromMap(rawEvent, sourceInfo, schemaInfo);
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
index f30e07330..0d2ea491a 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
@@ -115,7 +115,10 @@ public class PipelineElementTemplateVisitor implements
StaticPropertyVisitor {
@Override
public void visit(MappingPropertyUnary mappingPropertyUnary) {
- // Do nothing, not supported by pipeline element templates
+ if (hasKey(mappingPropertyUnary)) {
+ var selectedProperty = getAsString(mappingPropertyUnary);
+ mappingPropertyUnary.setSelectedProperty(selectedProperty);
+ }
}
@Override
