This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new d69504c915 feat(#2374): Add suggestion to test processing elements
(#2375)
d69504c915 is described below
commit d69504c91540b4f93be267910d41d1dae44abfd1
Author: Philipp Zehnder <[email protected]>
AuthorDate: Thu May 2 08:38:14 2024 +0200
feat(#2374): Add suggestion to test processing elements (#2375)
* feat(#2374): Add suggestion to test processing elements
* #2374 refactored unit tests to use ProcessingElementTestExecutor
* #2374 cleanup
* #2374 cleanup
* #2374 updated test executor to non-static methods
* #2374 updated test executor to non-static methods
* test(#2374): provide a framework to easily unit test processing elements
(#2730)
* #2374 refactored unit tests to use ProcessingElementTestExecutor
* #2374 cleanup
* #2374 cleanup
* #2374 updated test executor to non-static methods
* #2374 updated test executor to non-static methods
* #2374 Added test configuration builder and resolved selector prefixes
* #2374 Moved test executor to a new package
* feat(#2374): Fix cyclic dependencies and checkstyle
* feat(#2374): Fix checkstyle
* #2374 extracted TestConfigurationBuilder into a separate file
---------
Co-authored-by: Isaak <[email protected]>
---
pom.xml | 1 +
.../streampipes-processors-filters-jvm/pom.xml | 2 +-
.../numericalfilter/NumericalFilterProcessor.java | 6 +-
.../booleanfilter/TestBooleanFilterProcessor.java | 118 +++----
.../processor/compose/TestComposeProcessor.java | 208 ++++++-------
.../processor/merge/TestMergeByTimeProcessor.java | 276 +++++++++--------
.../NumericalFilterProcessorTest.java | 83 +++++
.../TestSwingingDoorTrendingFilterProcessor.java | 339 +++++++++++++--------
.../textfilter/TestTextFilterProcessor.java | 188 ++++++------
.../template/PipelineElementTemplateVisitor.java | 5 +-
.../pom.xml | 51 ++--
.../streampipes/test/executors/PrefixStrategy.java | 23 ++
.../executors/ProcessingElementTestExecutor.java | 190 ++++++++++++
.../test/executors/TestConfiguration.java | 47 +++
.../test/executors/TestConfigurationBuilder.java | 62 ++++
15 files changed, 1012 insertions(+), 587 deletions(-)
diff --git a/pom.xml b/pom.xml
index f1ceafa3e6..967ff16e7b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -868,6 +868,7 @@
<module>streampipes-storage-couchdb</module>
<module>streampipes-storage-management</module>
<module>streampipes-test-utils</module>
+ <module>streampipes-test-utils-executors</module>
<module>streampipes-user-management</module>
<module>streampipes-vocabulary</module>
<module>streampipes-wrapper</module>
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
b/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
index ed05e4cbc5..536e4c9929 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
@@ -62,7 +62,7 @@
</dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
- <artifactId>streampipes-test-utils</artifactId>
+ <artifactId>streampipes-test-utils-executors</artifactId>
<version>0.95.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
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 14062127cb..5f9a030d5a 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/booleanfilter/TestBooleanFilterProcessor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/booleanfilter/TestBooleanFilterProcessor.java
index 3f473f6f61..a6d1a73727 100644
---
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/booleanfilter/TestBooleanFilterProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/booleanfilter/TestBooleanFilterProcessor.java
@@ -18,106 +18,80 @@
package org.apache.streampipes.processors.filters.jvm.processor.booleanfilter;
-import
org.apache.streampipes.extensions.api.pe.context.EventProcessorRuntimeContext;
-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.sdk.extractor.ProcessingElementParameterExtractor;
-import org.apache.streampipes.test.extensions.api.StoreEventCollector;
-import org.apache.streampipes.wrapper.params.compat.ProcessorParams;
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
public class TestBooleanFilterProcessor {
- private static final String STREAM_PREFIX = "s0::";
- private static final Logger LOG =
LoggerFactory.getLogger(TestBooleanFilterProcessor.class);
+ BooleanFilterProcessor processor;
+ private static final String FIELD_NAME = "Test";
- static Stream<Arguments> data() {
- return Stream.of(
- Arguments.of("True", Arrays.asList(true, true, false, true, false,
true, false, true), 5),
- Arguments.of("True", Arrays.asList(true, true, true), 3),
- Arguments.of("True", Arrays.asList(false, false, false), 0),
- Arguments.of("True", Collections.emptyList(), 0),
- Arguments.of("False", Arrays.asList(true, false, true, false, true,
false, true, false, true), 4),
- Arguments.of("False", Arrays.asList(true, true, true), 0),
- Arguments.of("False", Arrays.asList(false, false, false), 3),
- Arguments.of("False", Collections.emptyList(), 0)
- );
+ @BeforeEach
+ public void setup(){
+ processor = new BooleanFilterProcessor();
}
-
@ParameterizedTest
@MethodSource("data")
- public void testBoolenFilter(
+ public void test(
String boolToKeep,
List<Boolean> eventBooleans,
- int expectedFilteredBooleanCount
+ List<Boolean> outputEventBooleans
) {
- var fieldName = "Test";
- var processorParams = mock(ProcessorParams.class);
- var eventProcessorRuntimeContext =
mock(EventProcessorRuntimeContext.class);
+ TestConfiguration configuration = TestConfiguration.builder()
+ .configWithDefaultPrefix(BooleanFilterProcessor.BOOLEAN_MAPPING,
FIELD_NAME)
+ .config(BooleanFilterProcessor.VALUE, boolToKeep)
+ .build();
- var processor = new BooleanFilterProcessor();
- var extractor = mock(ProcessingElementParameterExtractor.class);
- when(processorParams.extractor()).thenReturn(extractor);
-
when(extractor.mappingPropertyValue(BooleanFilterProcessor.BOOLEAN_MAPPING)).thenReturn(STREAM_PREFIX
+ fieldName);
- when(extractor.selectedSingleValue(BooleanFilterProcessor.VALUE,
String.class)).thenReturn(boolToKeep);
+ List<Map<String, Object>> events = new ArrayList<>();
+ eventBooleans.forEach(bool->events.add(Map.of(FIELD_NAME, bool)));
- var collector = new StoreEventCollector();
- processor.onInvocation(processorParams, collector,
eventProcessorRuntimeContext);
+ List<Map<String, Object>> outputEvents = new ArrayList<>();
+ outputEventBooleans.forEach(bool->outputEvents.add(Map.of(FIELD_NAME,
bool)));
- sendEvents(processor, collector, eventBooleans, fieldName);
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
- assertEquals(expectedFilteredBooleanCount, collector.getEvents().size());
+ testExecutor.run(events, outputEvents);
}
- private void sendEvents(
- BooleanFilterProcessor processor,
- StoreEventCollector collector,
- List<Boolean> eventBooleans,
- String fieldName) {
- List<Event> events = makeEvents(eventBooleans, fieldName);
- for (Event event : events) {
- LOG.info("Sending event with value "
- + event.getFieldBySelector(STREAM_PREFIX + fieldName)
- .getAsPrimitive()
- .getAsBoolean());
- processor.onEvent(event, collector);
-
- }
- }
-
- private List<Event> makeEvents(List<Boolean> eventBooleans, String
fieldName) {
- List<Event> events = new ArrayList<>();
- for (Boolean eventSetting : eventBooleans) {
- events.add(makeEvent(eventSetting, fieldName));
- }
- return events;
- }
-
- private Event makeEvent(Boolean value, String fieldName) {
- Map<String, Object> map = new HashMap<>();
- map.put(fieldName, value);
- return EventFactory.fromMap(map, new SourceInfo("test" + "-topic", "s0"),
- new SchemaInfo(null, new ArrayList<>())
+ static Stream<Arguments> data() {
+ return Stream.of(
+ Arguments.of("True",
+ Arrays.asList(true, true, false, true, false, true, false, true),
+ Arrays.asList(true, true, true, true, true)),
+ Arguments.of("True",
+ Arrays.asList(true, true, true),
+ Arrays.asList(true, true, true)),
+ Arguments.of("True",
+ Arrays.asList(false, false, false),
+ Collections.emptyList()),
+ Arguments.of("True",
+ Collections.emptyList(),
+ Collections.emptyList()),
+ Arguments.of("False",
+ Arrays.asList(true, false, true, false, true, false, true, false,
true),
+ Arrays.asList(false, false, false, false)),
+ Arguments.of("False",
+ Arrays.asList(true, true, true),
+ Collections.emptyList()),
+ Arguments.of("False",
+ Arrays.asList(false, false, false),
+ Arrays.asList(false, false, false)),
+ Arguments.of("False",
+ Collections.emptyList(),
+ Collections.emptyList())
);
}
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/compose/TestComposeProcessor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/compose/TestComposeProcessor.java
index fd56bb23ee..05bf18cccd 100644
---
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/compose/TestComposeProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/compose/TestComposeProcessor.java
@@ -18,124 +18,98 @@
package org.apache.streampipes.processors.filters.jvm.processor.compose;
-//@RunWith(Parameterized.class)
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.model.output.OutputStrategy;
+import org.apache.streampipes.test.executors.PrefixStrategy;
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
public class TestComposeProcessor {
-//
-// private static final Logger LOG =
LoggerFactory.getLogger(TestComposeProcessor.class);
-//
-// @org.junit.runners.Parameterized.Parameter
-// public String testName;
-//
-// @org.junit.runners.Parameterized.Parameter(1)
-// public List<Map<String, Object>> eventMaps;
-//
-// @org.junit.runners.Parameterized.Parameter(2)
-// public List<String> selectorPrefixes;
-//
-// @org.junit.runners.Parameterized.Parameter(3)
-// public int expectedNumOfEvents;
-//
-// @org.junit.runners.Parameterized.Parameter(4)
-// public int expectedEventSize;
-//
-// private static final String outputKeySelector1 = "key-selector1";
-// private static final String outputKeySelector2 = "key-selector2";
-//
-// @org.junit.runners.Parameterized.Parameters
-// public static Iterable<Object[]> data() {
-// Map<String, Object> mapWithFirstOutputSelector = new HashMap<>();
-// mapWithFirstOutputSelector.put(outputKeySelector1, new Object());
-//
-// Map<String, Object> mapWithSecondOutputSelector = new HashMap<>();
-// mapWithSecondOutputSelector.put(outputKeySelector2, new Object());
-//
-// Map<String, Object> mapWithInvalidOutputSelector = new HashMap<>();
-// mapWithInvalidOutputSelector.put("invalid-selector", new Object());
-//
-// List<Map<String, Object>> singleMap = new ArrayList<>();
-// singleMap.add(mapWithFirstOutputSelector);
-//
-// List<Map<String, Object>> twoMapsMatching = new ArrayList<>();
-// twoMapsMatching.add(mapWithFirstOutputSelector);
-// twoMapsMatching.add(mapWithSecondOutputSelector);
-//
-// List<Map<String, Object>> twoMapsOneMatching = new ArrayList<>();
-// twoMapsOneMatching.add(mapWithFirstOutputSelector);
-// twoMapsOneMatching.add(mapWithInvalidOutputSelector);
-//
-// List<Map<String, Object>> twoMapsNoneMatching = new ArrayList<>();
-// twoMapsNoneMatching.add(mapWithInvalidOutputSelector);
-// twoMapsNoneMatching.add(new HashMap<>(mapWithInvalidOutputSelector));
-//
-// return Arrays.asList(new Object[][]{
-// {"testWithOneEvent", singleMap, List.of("s0"), 0, 0},
-// {"testWithTwoEventsSamePrefix", twoMapsMatching, List.of("s0",
"s0"), 0, 0},
-// {"testWithTwoEvents", twoMapsMatching, List.of("s0", "s1"), 1,
2},
-// {"testWithTwoEventsAnd1InvalidSelector", twoMapsOneMatching,
List.of("s0", "s1"), 1, 1},
-// {"testWithTwoEventsWithInvalidSelectors", twoMapsNoneMatching,
List.of("s0", "s1"), 1, 0}
-// });
-// }
-//
-//
-//
-// @Test
-// public void testComposeProcessor() {
-// LOG.info("Executing test: {}", testName);
-// var processor = new ComposeProcessor();
-// var originalGraph = processor.declareModel();
-//
originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-// var graph = InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-// List<OutputStrategy> outputStrategies = new ArrayList<>();
-// outputStrategies.add(new CustomOutputStrategy(List.of("s0::" +
outputKeySelector1, "s1::" + outputKeySelector2)));
-// graph.setOutputStrategies(outputStrategies);
-// var params = new ProcessorParams(graph);
-//
-// var eventCollector = new StoreEventCollector();
-// processor.onInvocation(params, eventCollector, null);
-//
-// List<Event> collectedEvents = sendEvents(processor, eventCollector);
-//
-// LOG.info("Expected collected event count is: {}", expectedNumOfEvents);
-// LOG.info("Actual collected event count is: {}", collectedEvents.size());
-// assertEquals(expectedNumOfEvents, collectedEvents.size());
-//
-// if (!collectedEvents.isEmpty()){
-// int eventSize = collectedEvents.get(0).getFields().size();
-//
-// LOG.info("Expected event size is: {}", expectedEventSize);
-// LOG.info("Actual event size is: {}", eventSize);
-// assertEquals(expectedEventSize, eventSize);
-// }
-// }
-//
-// private List<Event> sendEvents(ComposeProcessor processor,
StoreEventCollector collector) {
-// List<Event> events = makeEvents();
-// for (Event event : events) {
-// LOG.info("Sending event with map: " + event.getFields()
-// + ", and prefix selector: " +
event.getSourceInfo().getSelectorPrefix());
-// processor.onEvent(event, collector);
-// try {
-// Thread.sleep(100);
-// } catch (InterruptedException e) {
-// e.printStackTrace();
-// }
-// }
-// return collector.getEvents();
-// }
-// private List<Event> makeEvents() {
-// List<Event> events = new ArrayList<>();
-// for (int i = 0; i < eventMaps.size(); i++) {
-// events.add(makeEvent(eventMaps.get(i), selectorPrefixes.get(i)));
-// }
-// return events;
-// }
-//
-// private Event makeEvent(Map<String, Object> eventMap, String
selectorPrefix) {
-// return EventFactory.fromMap(eventMap, new SourceInfo("test" + "-topic",
selectorPrefix),
-// new SchemaInfo(null, new ArrayList<>()));
-// }
-}
+ ComposeProcessor processor;
+ private static final String SELECTOR_1 = "key-selector1";
+ private static final String SELECTOR_2 = "key-selector2";
+ private static final String INVALID_SELECTOR = "invalid-selector";
+ private static final String S0_PREFIX = "s0::";
+ private static final String S1_PREFIX = "s1::";
+
+ @BeforeEach
+ public void setup(){
+ processor = new ComposeProcessor();
+ }
+
+ @ParameterizedTest
+ @MethodSource("data")
+ public void test(List<Map<String, Object>> events,
+ List<Map<String, Object>> outputEvents,
+ PrefixStrategy prefixStrategy) {
+ Consumer<DataProcessorInvocation> invocationConfig = (invocation->{
+ List<OutputStrategy> outputStrategies = new ArrayList<>();
+ outputStrategies.add(new CustomOutputStrategy(List.of(S0_PREFIX +
SELECTOR_1, S1_PREFIX + SELECTOR_2)));
+ invocation.setOutputStrategies(outputStrategies);
+ });
+ TestConfiguration configuration =
TestConfiguration.builder().prefixStrategy(prefixStrategy).build();
+
+ ProcessingElementTestExecutor testExecutor =
+ new ProcessingElementTestExecutor(processor, configuration,
invocationConfig);
+
+ testExecutor.run(events, outputEvents);
+ }
+ static Stream<Arguments> data() {
+
+ var object1 = new Object();
+ var object2 = new Object();
+
+ return Stream.of(
+ Arguments.of(
+ List.of(
+ Map.of(SELECTOR_1, object1)),
+ List.of(),
+ PrefixStrategy.SAME_PREFIX),
+ Arguments.of(
+ List.of(
+ Map.of(SELECTOR_1, object1),
+ Map.of(SELECTOR_2, object2)),
+ List.of(),
+ PrefixStrategy.SAME_PREFIX),
+ Arguments.of(
+ List.of(
+ Map.of(SELECTOR_1, object1),
+ Map.of(SELECTOR_2, object2)),
+ List.of(
+ Map.of(SELECTOR_1, object1, SELECTOR_2, object2)
+ ),
+ PrefixStrategy.ALTERNATE),
+ Arguments.of(
+ List.of(
+ Map.of(SELECTOR_1, object1),
+ Map.of(INVALID_SELECTOR, object2)),
+ List.of(
+ Map.of(SELECTOR_1, object1)
+ ),
+ PrefixStrategy.ALTERNATE),
+ Arguments.of(
+ List.of(
+ Map.of(INVALID_SELECTOR, object1),
+ Map.of(INVALID_SELECTOR, object2)),
+ List.of(
+ Map.of()
+ ),
+ PrefixStrategy.ALTERNATE)
+ );
+ }
+}
diff --git
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
index 2c4334fb00..953080fb68 100644
---
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
@@ -17,145 +17,141 @@
*/
package org.apache.streampipes.processors.filters.jvm.processor.merge;
-//@RunWith(Parameterized.class)
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.test.executors.PrefixStrategy;
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+import org.apache.streampipes.test.executors.TestConfigurationBuilder;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+
public class TestMergeByTimeProcessor {
-//
-// private static final Logger LOG =
LoggerFactory.getLogger(TestMergeByTimeProcessor.class);
-//
-// private static final Integer timeInterval = 100;
-// @org.junit.runners.Parameterized.Parameter
-// public String testName;
-// @org.junit.runners.Parameterized.Parameter(1)
-// public List<String> eventStrings;
-// @org.junit.runners.Parameterized.Parameter(2)
-// public List<String> expectedValue;
-//
-// @org.junit.runners.Parameterized.Parameters
-// public static Iterable<Object[]> data() {
-// return Arrays.asList(new Object[][]{
-// {"testWithInInterval", Arrays.asList("s0:0", "s1:90"),
List.of("(90,0)")},
-// {"testNotWithInInterval", Arrays.asList("s0:0", "s1:110"),
List.of()},
-// {"testWithInAndNotWithInInterval", Arrays.asList("s0:0", "s1:80",
"s0:110", "s1:500"),
-// List.of("(80,0)")},
-// {"testFigGvnInDocs",
-// Arrays.asList("s1:0", "s0:10", "s0:110", "s1:115", "s0:120",
"s1:230", "s0:340", "s0:500",
-// "s1:510"),
-// Arrays.asList("(0,10)", "(115,110)", "(510,500)")}
-// });
-// }
-//
-// @Test
-// public void testMergeByTimeProcessor() {
-// MergeByTimeProcessor mergeByTimeProcessor = new MergeByTimeProcessor();
-// DataProcessorDescription originalGraph =
mergeByTimeProcessor.declareModel();
-//
originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-// DataProcessorInvocation graph =
InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-// graph.setInputStreams(Arrays.asList(
-//
EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("in-stram0")),
-//
EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("in-stream1"))
-// ));
-//
-// graph.setOutputStream(
-//
EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("out-stream"))
-// );
-//
graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
-// .setActualTopicName("output-topic");
-//
-// List<String> outputKeySelectors = graph.getOutputStrategies()
-// .stream()
-// .filter(CustomOutputStrategy.class::isInstance)
-// .map(o -> (CustomOutputStrategy) o)
-// .findFirst()
-// .map(CustomOutputStrategy::getSelectedPropertyKeys)
-// .orElse(new ArrayList<>());
-// outputKeySelectors.add("s0::timestamp_mapping_stream_1");
-// outputKeySelectors.add("s1::timestamp_mapping_stream_2");
-//
-// List<MappingPropertyUnary> mappingPropertyUnaries =
graph.getStaticProperties()
-// .stream()
-// .filter(p -> p instanceof MappingPropertyUnary)
-// .map((p -> (MappingPropertyUnary) p))
-// .filter(p -> Arrays.asList(
-// MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
-// MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY)
-// .contains(p.getInternalName()))
-// .collect(Collectors.toList());
-//
-// assert mappingPropertyUnaries.size() == 2;
-// mappingPropertyUnaries.get(0)
-// .setSelectedProperty("s0::" +
MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY);
-// mappingPropertyUnaries.get(1)
-// .setSelectedProperty("s1::" +
MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY);
-//
-// FreeTextStaticProperty fsp = graph.getStaticProperties().stream()
-// .filter(p -> p instanceof FreeTextStaticProperty)
-// .map((p -> (FreeTextStaticProperty) p))
-// .filter(p ->
p.getInternalName().equals(MergeByTimeProcessor.TIME_INTERVAL))
-// .findFirst().orElse(null);
-// assert fsp != null;
-// fsp.setValue(String.valueOf(timeInterval));
-//
-// ProcessorParams params = new ProcessorParams(graph);
-//
-// StoreEventCollector collector = new StoreEventCollector();
-//
-// mergeByTimeProcessor.onInvocation(params, collector, null);
-// sendEvents(mergeByTimeProcessor, collector);
-//
-// List<String> actualCollectedEvents = collector.getEvents().stream()
-// .map(e -> formatMergedEvent(e))
-// .collect(Collectors.toList());
-//
-// LOG.info("Expected merged event is {}", expectedValue);
-// LOG.info("Actual merged event is {}", actualCollectedEvents);
-// assertTrue(eventsEquals(expectedValue, actualCollectedEvents));
-// }
-//
-// private boolean eventsEquals(List<String> expectedValue, List<String>
actualCollectedEvents) {
-// if (expectedValue.size() != actualCollectedEvents.size()) {
-// return false;
-// }
-// for (int i = 0; i < expectedValue.size(); i++) {
-// if
(!expectedValue.get(i).equalsIgnoreCase(actualCollectedEvents.get(i))) {
-// return false;
-// }
-// }
-// return true;
-// }
-//
-// private String formatMergedEvent(Event mergedEvent) {
-// return String.format("(%s)", mergedEvent.getFields().values().stream()
-// .map(m ->
m.getAsPrimitive().getAsString()).collect(Collectors.joining(",")));
-// }
-//
-// private void sendEvents(MergeByTimeProcessor mergeByTimeProcessor,
StoreEventCollector spOut) {
-// List<Event> events = makeEvents();
-// for (Event event : events) {
-// mergeByTimeProcessor.onEvent(event, spOut);
-// }
-//
-// }
-//
-// private List<Event> makeEvents() {
-// List<Event> events = Lists.newArrayList();
-// for (String eventString : eventStrings) {
-// events.add(makeEvent(eventString));
-// }
-// return events;
-// }
-//
-// private Event makeEvent(String eventString) {
-// Map<String, Object> map = Maps.newHashMap();
-// String streamId = eventString.split(":")[0];
-// String timestamp = eventString.split(":")[1];
-// if (streamId.equals("s0")) {
-// map.put(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
timestamp);
-// } else {
-// map.put(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
timestamp);
-// }
-// return EventFactory.fromMap(map,
-// new SourceInfo("test", streamId),
-// new SchemaInfo(null, Lists.newArrayList()));
-// }
+
+ private static final String S0_PREFIX = "s0";
+ private static final String S1_PREFIX = "s1";
+ private static final Integer timeInterval = 100;
+
+ MergeByTimeProcessor processor;
+
+ @BeforeEach
+ public void setup(){
+ processor = new MergeByTimeProcessor();
+ }
+ @ParameterizedTest
+ @MethodSource("data")
+ public void test(List<Map<String, Object>> events,
+ List<Map<String, Object>> outputEvents,
+ PrefixStrategy prefixStrategy,
+ List<String> customPrefixes){
+
+
+ TestConfigurationBuilder configurationBuilder = TestConfiguration.builder()
+ .config(MergeByTimeProcessor.TIME_INTERVAL, timeInterval)
+ .configWithPrefix(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, S0_PREFIX)
+ .configWithPrefix(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, S1_PREFIX);
+
+ if (customPrefixes != null){
+ configurationBuilder.customPrefixStrategy(customPrefixes);
+ } else {
+ configurationBuilder.prefixStrategy(prefixStrategy);
+ }
+
+
+ Consumer<DataProcessorInvocation> invocationConfig = (invocation->{
+ List<String> outputKeySelectors = invocation.getOutputStrategies()
+ .stream()
+ .filter(CustomOutputStrategy.class::isInstance)
+ .map(o -> (CustomOutputStrategy) o)
+ .findFirst()
+ .map(CustomOutputStrategy::getSelectedPropertyKeys)
+ .orElse(new ArrayList<>());
+ outputKeySelectors.add(S0_PREFIX + "::" +
MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY);
+ outputKeySelectors.add(S1_PREFIX + "::" +
MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY);
+ });
+
+ ProcessingElementTestExecutor testExecutor =
+ new ProcessingElementTestExecutor(processor,
configurationBuilder.build(), invocationConfig);
+
+ testExecutor.run(events, outputEvents);
+ }
+
+ static Stream<Arguments> data() {
+ return Stream.of(
+ Arguments.of(
+ List.of(
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"0"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"90")
+ ),
+ List.of(
+ Map.of(
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, "90",
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, "0"
+ )),
+ PrefixStrategy.ALTERNATE,
+ null),
+ Arguments.of(
+ List.of(
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, "0"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, "110")
+ ),
+ List.of(),
+ PrefixStrategy.ALTERNATE,
+ null),
+ Arguments.of(
+ List.of(
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"0"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"80"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"110"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"500")
+ ),
+ List.of(
+ Map.of(
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, "80",
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, "0"
+ )),
+ PrefixStrategy.ALTERNATE,
+ null),
+ Arguments.of(
+ List.of(
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"0"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"10"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"110"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"115"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"120"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"230"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"340"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
"500"),
+ Map.of(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY,
"510")
+ ),
+ List.of(
+ Map.of(
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, "0",
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, "10"
+ ),
+ Map.of(
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, "115",
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, "110"
+ ),
+ Map.of(
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, "510",
+ MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, "500"
+ )
+ ),
+ null,
+ List.of(S1_PREFIX, S0_PREFIX, S0_PREFIX, S1_PREFIX, S0_PREFIX,
S1_PREFIX, S0_PREFIX, S0_PREFIX, S1_PREFIX))
+ );
+ }
}
\ 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/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 0000000000..ed08eecd9f
--- /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,83 @@
+/*
+ * 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.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+public class NumericalFilterProcessorTest {
+
+ private static final String PROPERTY_NAME = "propertyName";
+ private NumericalFilterProcessor processor;
+
+
+ @BeforeEach
+ public void setup() {
+ processor = new NumericalFilterProcessor();
+ }
+
+ @Test
+ public void testLowerThenOperatorFilterNotApplied() {
+
+ TestConfiguration configuration = TestConfiguration.builder()
+ .configWithDefaultPrefix(NumericalFilterProcessor.NUMBER_MAPPING,
PROPERTY_NAME)
+ .config(NumericalFilterProcessor.VALUE, 10.0)
+ .config(NumericalFilterProcessor.OPERATION, "<")
+ .build();
+
+ 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 testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ testExecutor.run(inputEvents, outputEvents);
+ }
+
+ @Test
+ public void testLowerThenOperatorFilterApplied() {
+
+ TestConfiguration configuration = TestConfiguration.builder()
+ .configWithDefaultPrefix(NumericalFilterProcessor.NUMBER_MAPPING,
PROPERTY_NAME)
+ .config(NumericalFilterProcessor.VALUE, 10.0)
+ .config(NumericalFilterProcessor.OPERATION, "<")
+ .build();
+
+ List<Map<String, Object>> inputEvents = List.of(
+ Map.of(PROPERTY_NAME, 11.0f)
+ );
+
+ List<Map<String, Object>> outputEvents = List.of();
+
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ testExecutor.run(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/sdt/TestSwingingDoorTrendingFilterProcessor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/TestSwingingDoorTrendingFilterProcessor.java
index 3a5dbf2288..07f14e2ad5 100644
---
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/TestSwingingDoorTrendingFilterProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/TestSwingingDoorTrendingFilterProcessor.java
@@ -18,133 +18,216 @@
package org.apache.streampipes.processors.filters.jvm.processor.sdt;
-//@RunWith(Parameterized.class)
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
public class TestSwingingDoorTrendingFilterProcessor {
-//
-// private static final Logger LOG =
LoggerFactory.getLogger(TestSwingingDoorTrendingFilterProcessor.class);
-//
-// @org.junit.runners.Parameterized.Parameter
-// public String testName;
-// @org.junit.runners.Parameterized.Parameter(1)
-// public String sdtCompressionDeviation;
-// @org.junit.runners.Parameterized.Parameter(2)
-// public String sdtCompressionMinTimeInterval;
-// @org.junit.runners.Parameterized.Parameter(3)
-// public String sdtCompressionMaxTimeInterval;
-// @org.junit.runners.Parameterized.Parameter(4)
-// public int expectedFilteredCount;
-// @org.junit.runners.Parameterized.Parameter(5)
-// public List<Pair<Long, Double>> eventSettings;
-// @org.junit.runners.Parameterized.Parameter(6)
-// public boolean expectException;
-// @org.junit.runners.Parameterized.Parameter(7)
-// public String expectedErrorMessage;
-//
-// private final String sdtTimestampField = "sdtTimestampField";
-// private final String sdtValueField = "sdtValueField";
-//
-// @org.junit.runners.Parameterized.Parameters
-// public static Iterable<Object[]> data() {
-//
-// return Arrays.asList(new Object[][]{
-// {"testWithOneEvent", "10.0", "100", "500", 1, List.of(Pair.of(9,
50.0)), false, ""},
-// {"testFullFilter", "10.0", "100", "500", 4, List.of(
-// Pair.of(0, 50.0), //true
-// Pair.of(50, 50.0), //false
-// Pair.of(200, 100.0), //false
-// Pair.of(270, 140.0), //false
-// Pair.of(300, 250.0), //true
-// Pair.of(900, 500.0), //true
-// Pair.of(1100, 800.0), //false
-// Pair.of(1250, 1600.0) //true
-// )
-// , false, ""},
-// {"testWithNegativeCompressionDeviation", "-10.0", "100", "500",
1, new ArrayList<>(), true
-// , "Compression Deviation should be positive!"},
-// {"testWithNegativeMinInterval", "10.0", "-100", "500", 1, new
ArrayList<>(), true
-// , "Compression Minimum Time Interval should be >= 0!"},
-// {"testWithMinInterval>MaxInterval", "10.0", "1000", "500", 1,
new ArrayList<>(), true
-// , "Compression Minimum Time Interval should be <
Compression Maximum Time Interval!"}
-// });
-// }
-//
-// @Rule
-// public ExpectedException exceptionRule = ExpectedException.none();
-//
-// @Test
-// public void testSdtFilter() {
-// LOG.info("Executing test: {}", testName);
-// SwingingDoorTrendingFilterProcessor processor = new
SwingingDoorTrendingFilterProcessor();
-// DataProcessorDescription originalGraph = processor.declareModel();
-//
originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-// DataProcessorInvocation graph =
-// InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-//
-// ProcessorParams params = new ProcessorParams(graph);
-//
params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_TIMESTAMP_FIELD_KEY
-// , MappingPropertyUnary.class)
-// .setSelectedProperty("test::" + sdtTimestampField);
-//
params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_VALUE_FIELD_KEY
-// , MappingPropertyUnary.class)
-// .setSelectedProperty("test::" + sdtValueField);
-//
params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_DEVIATION_KEY
-// , FreeTextStaticProperty.class)
-// .setValue(sdtCompressionDeviation);
-//
params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MIN_INTERVAL_KEY
-// , FreeTextStaticProperty.class)
-// .setValue(sdtCompressionMinTimeInterval);
-//
params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MAX_INTERVAL_KEY
-// , FreeTextStaticProperty.class)
-// .setValue(sdtCompressionMaxTimeInterval);
-//
-// if (expectException){
-// LOG.info("Expecting Error Message: {}", expectedErrorMessage);
-// exceptionRule.expect(SpRuntimeException.class);
-// exceptionRule.expectMessage(expectedErrorMessage);
-// }
-//
-// StoreEventCollector eventCollector = new StoreEventCollector();
-// processor.onInvocation(params, eventCollector, null);
-//
-// int result = sendEvents(processor, eventCollector);
-//
-// LOG.info("Expected SDT filtered count is: {}", expectedFilteredCount);
-// LOG.info("Actual SDT filtered count is: {}", result);
-// assertEquals(expectedFilteredCount, result);
-// }
-//
-//
-// private int sendEvents(SwingingDoorTrendingFilterProcessor processor,
StoreEventCollector collector) {
-// List<Event> events = makeEvents();
-// for (Event event : events) {
-// LOG.info("Sending event with timestamp: "
-// + event.getFieldBySelector("test::" +
sdtTimestampField).getAsPrimitive().getAsLong()
-// + ", and value: "
-// + event.getFieldBySelector("test::" +
sdtValueField).getAsPrimitive().getAsFloat());
-// processor.onEvent(event, collector);
-// try {
-// Thread.sleep(100);
-// } catch (InterruptedException e) {
-// e.printStackTrace();
-// }
-// }
-// return collector.getEvents().size();
-// }
-//
-// private List<Event> makeEvents() {
-// List<Event> events = new ArrayList<>();
-// for (Pair<Long, Double> eventSetting: eventSettings) {
-// events.add(makeEvent(eventSetting));
-// }
-// return events;
-// }
-//
-// private Event makeEvent(Pair<Long, Double> eventSetting) {
-// Map<String, Object> map = new HashMap<>();
-// map.put(sdtTimestampField, eventSetting.getKey());
-// map.put(sdtValueField, eventSetting.getValue());
-// return EventFactory.fromMap(map, new SourceInfo("test" + "-topic",
"test"),
-// new SchemaInfo(null, new ArrayList<>()));
-// }
+
+ private final String sdtTimestampField = "sdtTimestampField";
+ private final String sdtValueField = "sdtValueField";
+
+ private SwingingDoorTrendingFilterProcessor processor;
+
+ @BeforeEach
+ public void setup() {
+ processor = new SwingingDoorTrendingFilterProcessor();
+ }
+
+ @Test
+ public void test() {
+ TestConfiguration configuration = TestConfiguration
+ .builder()
+
.configWithDefaultPrefix(SwingingDoorTrendingFilterProcessor.SDT_TIMESTAMP_FIELD_KEY,
sdtTimestampField)
+
.configWithDefaultPrefix(SwingingDoorTrendingFilterProcessor.SDT_VALUE_FIELD_KEY,
sdtValueField)
+
.config(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_DEVIATION_KEY,
"10.0")
+
.config(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MIN_INTERVAL_KEY,
"100")
+
.config(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MAX_INTERVAL_KEY,
"500")
+ .build();
+
+ List<Map<String, Object>> inputEvents = List.of(
+ Map.of(sdtTimestampField, 0,
+ sdtValueField, 50.0
+ ),
+ Map.of(sdtTimestampField, 50,
+ sdtValueField, 50.0
+ ),
+ Map.of(sdtTimestampField, 200,
+ sdtValueField, 100.0
+ ),
+ Map.of(sdtTimestampField, 270,
+ sdtValueField, 140.0
+ ),
+ Map.of(sdtTimestampField, 300,
+ sdtValueField, 250.0
+ ),
+ Map.of(sdtTimestampField, 900,
+ sdtValueField, 500.0
+ ),
+ Map.of(sdtTimestampField, 1100,
+ sdtValueField, 800.0
+ ),
+ Map.of(sdtTimestampField, 1250,
+ sdtValueField, 1600.0
+ )
+ );
+
+ List<Map<String, Object>> outputEvents = List.of(
+ Map.of(sdtTimestampField, 0,
+ sdtValueField, 50.0
+ ),
+ Map.of(sdtTimestampField, 270,
+ sdtValueField, 140.0
+ ),
+ Map.of(sdtTimestampField, 900,
+ sdtValueField, 500.0
+ ),
+ Map.of(sdtTimestampField, 1100,
+ sdtValueField, 800.0
+ )
+ );
+
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ testExecutor.run(inputEvents, outputEvents);
+ }
+
+ @Test
+ public void testInvalidDeviationKey() {
+ TestConfiguration configuration = TestConfiguration
+ .builder()
+ .configWithDefaultPrefix(
+ SwingingDoorTrendingFilterProcessor.SDT_TIMESTAMP_FIELD_KEY,
+ sdtTimestampField
+ )
+ .configWithDefaultPrefix(
+ SwingingDoorTrendingFilterProcessor.SDT_VALUE_FIELD_KEY,
+ sdtValueField
+ )
+ .config(
+ SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_DEVIATION_KEY,
+ "-10.0"
+ )
+ .config(
+
SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MIN_INTERVAL_KEY,
+ "100"
+ )
+ .config(
+
SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MAX_INTERVAL_KEY,
+ "500"
+ )
+ .build();
+
+ List<Map<String, Object>> inputEvents = new ArrayList<>();
+
+ List<Map<String, Object>> outputEvents = new ArrayList<>();
+
+ Exception expectedException = new SpRuntimeException("Compression
Deviation should be positive!");
+
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ Exception exception = assertThrows(expectedException.getClass(), () -> {
+ testExecutor.run(inputEvents, outputEvents);
+ });
+ String expectedMessage = expectedException.getMessage();
+ String actualMessage = exception.getMessage();
+ assertTrue(actualMessage.contains(expectedMessage));
+
+ }
+
+ @Test
+ public void testNegativeMinimumTime() {
+ TestConfiguration configuration = TestConfiguration
+ .builder()
+ .configWithDefaultPrefix(
+ SwingingDoorTrendingFilterProcessor.SDT_TIMESTAMP_FIELD_KEY,
+ sdtTimestampField
+ )
+ .configWithDefaultPrefix(
+ SwingingDoorTrendingFilterProcessor.SDT_VALUE_FIELD_KEY,
+ sdtValueField
+ )
+ .config(
+ SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_DEVIATION_KEY,
+ "10.0"
+ )
+ .config(
+
SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MIN_INTERVAL_KEY,
+ "-100"
+ )
+ .config(
+
SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MAX_INTERVAL_KEY,
+ "500"
+ )
+ .build();
+
+ List<Map<String, Object>> inputEvents = new ArrayList<>();
+
+ List<Map<String, Object>> outputEvents = new ArrayList<>();
+
+ Exception expectedException = new SpRuntimeException("Compression Minimum
Time Interval should be >= 0!");
+
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ Exception exception = assertThrows(expectedException.getClass(), () -> {
+ testExecutor.run(inputEvents, outputEvents);
+ });
+ String expectedMessage = expectedException.getMessage();
+ String actualMessage = exception.getMessage();
+ assertTrue(actualMessage.contains(expectedMessage));
+ }
+
+ @Test
+ public void testInvalidTimeInterval() {
+ TestConfiguration configuration = TestConfiguration
+ .builder()
+ .configWithDefaultPrefix(
+ SwingingDoorTrendingFilterProcessor.SDT_TIMESTAMP_FIELD_KEY,
+ sdtTimestampField
+ )
+ .configWithDefaultPrefix(
+ SwingingDoorTrendingFilterProcessor.SDT_VALUE_FIELD_KEY,
+ sdtValueField
+ )
+ .config(
+ SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_DEVIATION_KEY,
+ "10.0"
+ )
+ .config(
+
SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MIN_INTERVAL_KEY,
+ "1000"
+ )
+ .config(
+
SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MAX_INTERVAL_KEY,
+ "500"
+ )
+ .build();
+
+ List<Map<String, Object>> inputEvents = new ArrayList<>();
+
+ List<Map<String, Object>> outputEvents = new ArrayList<>();
+
+ Exception expectedException = new
+ SpRuntimeException("Compression Minimum Time Interval should be <
Compression Maximum Time Interval!");
+
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ Exception exception = assertThrows(expectedException.getClass(), () -> {
+ testExecutor.run(inputEvents, outputEvents);
+ });
+ String expectedMessage = expectedException.getMessage();
+ String actualMessage = exception.getMessage();
+ assertTrue(actualMessage.contains(expectedMessage));
+ }
}
\ 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/textfilter/TestTextFilterProcessor.java
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TestTextFilterProcessor.java
index 54dc91129a..d283eb04ce 100644
---
a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TestTextFilterProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TestTextFilterProcessor.java
@@ -18,101 +18,99 @@
package org.apache.streampipes.processors.filters.jvm.processor.textfilter;
-//@RunWith(Parameterized.class)
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
public class TestTextFilterProcessor {
-// private static final Logger LOG =
LoggerFactory.getLogger(TestTextFilterProcessor.class);
-//
-// @org.junit.runners.Parameterized.Parameters
-// public static Iterable<Object[]> data() {
-// return Arrays.asList(new Object[][] {
-// {"TestLowerCaseMatch", "keyword", StringOperator.MATCHES,
Arrays.asList("keyword", "KeyWord", "KEYWORD"), 1},
-// {"TestUpperCaseMatch", "KEYWORD", StringOperator.MATCHES,
Arrays.asList("keyword", "KeyWord", "KEYWORD"), 1},
-// {"TestMixMatch", "KeyWord", StringOperator.MATCHES,
Arrays.asList("keyword", "KeyWord", "KEYWORD"), 1},
-// {"TestEmptyMatch", "keYWord", StringOperator.MATCHES,
Arrays.asList("keyword", "KeyWord", "KEYWORD"), 0},
-// {"TestMultipleMatch", "KeyWord", StringOperator.MATCHES,
-// Arrays.asList("keyword", "KeyWord", "KEYWORD", "KeyWord"), 2},
-//
-// {"TestContainsWord", "keyword", StringOperator.CONTAINS,
-// Arrays.asList("text contains keyword", "text doesn't have
word"), 1},
-// {"TestNotContainsWord", "keyword", StringOperator.CONTAINS,
-// Arrays.asList("text is empty", "text doesn't have word"), 0},
-// });
-// }
-//
-// @org.junit.runners.Parameterized.Parameter
-// public String selectedFieldName;
-//
-// @org.junit.runners.Parameterized.Parameter(1)
-// public String keyword;
-//
-// @org.junit.runners.Parameterized.Parameter(2)
-// public StringOperator stringOperator;
-//
-// @org.junit.runners.Parameterized.Parameter(3)
-// public List<String> eventStrings;
-//
-// @org.junit.runners.Parameterized.Parameter(4)
-// public int expectedFilteredTextCount;
-//
-// @Test
-// public void testTextFilter() {
-// TextFilterProcessor textFilterProcessor = new TextFilterProcessor();
-// DataProcessorDescription originalGraph =
textFilterProcessor.declareModel();
-//
originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-// DataProcessorInvocation graph =
InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-// graph.setInputStreams(Collections
-// .singletonList(EventStreamGenerator
-//
.makeStreamWithProperties(Collections.singletonList(selectedFieldName))));
-// graph.setOutputStream(EventStreamGenerator
-// .makeStreamWithProperties(Collections.singletonList(selectedFieldName)));
-//
graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
-// .setActualTopicName("output-topic");
-//
-// graph.getStaticProperties().stream()
-// .filter(p -> p instanceof MappingPropertyUnary)
-// .map(p -> (MappingPropertyUnary) p)
-// .filter(p ->
p.getInternalName().equals(TextFilterProcessor.MAPPING_PROPERTY_ID))
-// .findFirst().get().setSelectedProperty("s0::" + selectedFieldName);
-// ProcessorParams params = new ProcessorParams(graph);
-//
params.extractor().getStaticPropertyByName(TextFilterProcessor.OPERATION_ID,
OneOfStaticProperty.class)
-// .getOptions()
-// .stream().filter(o ->
o.getName().equals(stringOperator.name())).findFirst().get().setSelected(true);
-//
params.extractor().getStaticPropertyByName(TextFilterProcessor.KEYWORD_ID,
FreeTextStaticProperty.class)
-// .setValue(keyword);
-// StoreEventCollector collector = new StoreEventCollector();
-//
-// textFilterProcessor.onInvocation(params, collector, null);
-// int result = sendEvents(textFilterProcessor, collector);
-//
-// LOG.info("Expected filtered text count is {}",
expectedFilteredTextCount);
-// LOG.info("Actual filtered text count is {}", result);
-// assertEquals(expectedFilteredTextCount, result);
-//
-// }
-//
-// private int sendEvents(TextFilterProcessor textFilterProcessor,
StoreEventCollector collector) {
-// List<Event> events = makeEvents();
-// for (Event e : events) {
-// textFilterProcessor.onEvent(e, collector);
-// }
-// return collector.getEvents().size();
-// }
-//
-// private List<Event> makeEvents() {
-// List<Event> events = new ArrayList<>();
-// for (String eventString : eventStrings) {
-// events.add(makeEvent(eventString));
-// }
-// return events;
-// }
-//
-// private Event makeEvent(String eventString) {
-// Map<String, Object> map = Maps.newHashMap();
-// map.put(selectedFieldName, eventString);
-// return EventFactory.fromMap(map,
-// new SourceInfo("test", "s0"),
-// new SchemaInfo(null, Lists.newArrayList()));
-// }
+ TextFilterProcessor processor;
+
+ public static final String FIELD_NAME = "selectedField";
+
+ @BeforeEach
+ public void setup(){
+ processor = new TextFilterProcessor();
+ }
+
+ @ParameterizedTest
+ @MethodSource("data")
+ public void test(String keyword,
+ StringOperator stringOperator,
+ List<String> eventValues,
+ List<String> outputEventValues){
+
+ TestConfiguration configuration = TestConfiguration.builder()
+ .configWithDefaultPrefix(TextFilterProcessor.MAPPING_PROPERTY_ID,
FIELD_NAME)
+ .config(TextFilterProcessor.OPERATION_ID, stringOperator)
+ .config(TextFilterProcessor.KEYWORD_ID, keyword)
+ .build();
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ eventValues.forEach(value->events.add(Map.of(FIELD_NAME, value)));
+
+ List<Map<String, Object>> outputEvents = new ArrayList<>();
+ outputEventValues.forEach(value->outputEvents.add(Map.of(FIELD_NAME,
value)));
+
+ ProcessingElementTestExecutor testExecutor = new
ProcessingElementTestExecutor(processor, configuration);
+
+ testExecutor.run(events, outputEvents);
+ }
+
+
+ static Stream<Arguments> data() {
+ return Stream.of(
+ Arguments.of(
+ "keyword",
+ StringOperator.MATCHES,
+ List.of("keyword", "KeyWord", "KEYWORD"),
+ List.of("keyword")
+ ),
+ Arguments.of(
+ "KEYWORD",
+ StringOperator.MATCHES,
+ List.of("keyword", "KeyWord", "KEYWORD"),
+ List.of("KEYWORD")
+ ),
+ Arguments.of(
+ "KeyWord",
+ StringOperator.MATCHES,
+ List.of("keyword", "KeyWord", "KEYWORD"),
+ List.of("KeyWord")
+ ),
+ Arguments.of(
+ "keYWord",
+ StringOperator.MATCHES,
+ List.of("keyword", "KeyWord", "KEYWORD"),
+ List.of()
+ ),
+ Arguments.of(
+ "KeyWord",
+ StringOperator.MATCHES,
+ List.of("keyword", "KeyWord", "KEYWORD", "KeyWord"),
+ List.of("KeyWord", "KeyWord")
+ ),
+ Arguments.of(
+ "keyword",
+ StringOperator.CONTAINS,
+ List.of("text contains keyword", "text doesn't have word"),
+ List.of("text contains keyword")
+ ),
+ Arguments.of(
+ "keyword",
+ StringOperator.CONTAINS,
+ List.of("text is empty", "text doesn't have word"),
+ List.of()
+ )
+ );
+ }
}
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 627eedb250..742417ddd8 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
@@ -116,7 +116,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
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
b/streampipes-test-utils-executors/pom.xml
similarity index 71%
copy from streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
copy to streampipes-test-utils-executors/pom.xml
index ed05e4cbc5..8004169299 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
+++ b/streampipes-test-utils-executors/pom.xml
@@ -19,65 +19,56 @@
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<parent>
- <artifactId>streampipes-extensions</artifactId>
+ <artifactId>streampipes-parent</artifactId>
<groupId>org.apache.streampipes</groupId>
<version>0.95.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>streampipes-processors-filters-jvm</artifactId>
+ <artifactId>streampipes-test-utils-executors</artifactId>
<dependencies>
- <!-- StreamPipes dependencies -->
- <dependency>
- <groupId>org.apache.streampipes</groupId>
- <artifactId>streampipes-extensions-management</artifactId>
- <version>0.95.0-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>org.apache.streampipes</groupId>
- <artifactId>streampipes-wrapper-standalone</artifactId>
- <version>0.95.0-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>org.quartz-scheduler</groupId>
- <artifactId>quartz</artifactId>
- </dependency>
-
<!-- Test dependencies -->
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
- <scope>test</scope>
+ <scope>compile</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
- <scope>test</scope>
+ <scope>compile</scope>
</dependency>
<dependency>
- <groupId>org.junit.jupiter</groupId>
- <artifactId>junit-jupiter-params</artifactId>
- <scope>test</scope>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-pipeline-management</artifactId>
+ <version>0.95.0-SNAPSHOT</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.objenesis</groupId>
+ <artifactId>objenesis</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>net.bytebuddy</groupId>
+ <artifactId>byte-buddy</artifactId>
+ </exclusion>
+ </exclusions>
</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.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
</dependency>
</dependencies>
-
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
- <configuration>
- <propertyExpansion>
-
checkstyle.config.base.path=${project.parent.parent.basedir}/tools/maven
- </propertyExpansion>
- </configuration>
</plugin>
</plugins>
</build>
diff --git
a/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/PrefixStrategy.java
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/PrefixStrategy.java
new file mode 100644
index 0000000000..3cc2a1ee60
--- /dev/null
+++
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/PrefixStrategy.java
@@ -0,0 +1,23 @@
+/*
+ * 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.test.executors;
+
+public enum PrefixStrategy {
+ SAME_PREFIX, ALTERNATE
+}
diff --git
a/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/ProcessingElementTestExecutor.java
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/ProcessingElementTestExecutor.java
new file mode 100644
index 0000000000..f2d87197ad
--- /dev/null
+++
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/ProcessingElementTestExecutor.java
@@ -0,0 +1,190 @@
+/*
+ * 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.test.executors;
+
+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.apache.streampipes.test.generator.EventStreamGenerator;
+
+import org.junit.jupiter.api.Assertions;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+
+
+public class ProcessingElementTestExecutor {
+
+ private final IStreamPipesDataProcessor processor;
+ private TestConfiguration testConfiguration;
+ private Iterator<String> selectorPrefixes;
+ private Consumer<DataProcessorInvocation> invocationConfig;
+
+ public ProcessingElementTestExecutor(IStreamPipesDataProcessor processor,
TestConfiguration testConfiguration,
+ Consumer<DataProcessorInvocation>
invocationConfig) {
+ this.processor = processor;
+ this.testConfiguration = testConfiguration;
+ this.selectorPrefixes = testConfiguration.getPrefixes().iterator();
+ this.invocationConfig = invocationConfig;
+ }
+
+ public ProcessingElementTestExecutor(IStreamPipesDataProcessor processor,
TestConfiguration testConfiguration) {
+ this.processor = processor;
+ this.testConfiguration = testConfiguration;
+ this.selectorPrefixes = testConfiguration.getPrefixes().iterator();
+ }
+
+ public ProcessingElementTestExecutor(IStreamPipesDataProcessor processor,
+ Consumer<DataProcessorInvocation>
invocationConfig) {
+ this.processor = processor;
+ this.invocationConfig = invocationConfig;
+ this.selectorPrefixes = List.of("").iterator();
+ }
+
+
+ /**
+ * 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 inputEvents The list of input events to be processed.
+ * @param expectedOutputEvents The list of expected output events.
+ */
+ public void run(
+ List<Map<String, Object>> inputEvents,
+ List<Map<String, Object>> expectedOutputEvents
+ ) {
+
+
+ // initialize the extractor with the provided configuration of the user
input
+ var dataProcessorInvocation = getProcessorInvocation();
+ if (invocationConfig != null){
+ invocationConfig.accept(dataProcessorInvocation);
+ }
+
+ var e = getProcessingElementParameterExtractor(dataProcessorInvocation);
+ var mockParams = Mockito.mock(IDataProcessorParameters.class);
+
+ Mockito.when(mockParams.getModel()).thenReturn(dataProcessorInvocation);
+ Mockito.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 = Mockito.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
+ Mockito.verify(mockCollector,
+
Mockito.times(expectedOutputEvents.size())).collect(spOutputCollectorCaptor.capture());
+ var resultingEvents = spOutputCollectorCaptor.getAllValues();
+ IntStream.range(0, expectedOutputEvents.size())
+ .forEach(i -> Assertions.assertEquals(
+ expectedOutputEvents.get(i),
+ resultingEvents.get(i)
+ .getRaw()
+ ));
+
+ // validate that the processor is stopped correctly
+ processor.onPipelineStopped();
+ }
+
+ private static ProcessingElementParameterExtractor
getProcessingElementParameterExtractor(
+ DataProcessorInvocation dataProcessorInvocation
+ ) {
+ return ProcessingElementParameterExtractor.from(dataProcessorInvocation);
+ }
+
+ private DataProcessorInvocation getProcessorInvocation() {
+ var pipelineElementTemplate = getPipelineElementTemplate();
+
+ var invocation = new DataProcessorInvocation(
+ processor
+ .declareConfig()
+ .getDescription()
+ );
+
+ invocation.setOutputStream(EventStreamGenerator.makeEmptyStream());
+
+
+ return new DataProcessorTemplateHandler(
+ pipelineElementTemplate,
+ invocation,
+ true
+ )
+ .applyTemplateOnPipelineElement();
+ }
+
+ private PipelineElementTemplate getPipelineElementTemplate() {
+ var staticProperties = processor
+ .declareConfig()
+ .getDescription()
+ .getStaticProperties();
+
+
+ var configs = new HashMap<String, PipelineElementTemplateConfig>();
+
+ staticProperties.forEach(staticProperty -> {
+ var value =
testConfiguration.getFieldConfiguration().get(staticProperty.getInternalName());
+ configs.put(
+ staticProperty.getInternalName(),
+ new PipelineElementTemplateConfig(true, true, value)
+ );
+ });
+
+ return new PipelineElementTemplate("name", "description", configs);
+ }
+
+ private Event getEvent(Map<String, Object> rawEvent) {
+
+ if (!selectorPrefixes.hasNext()){
+ selectorPrefixes = testConfiguration.getPrefixes().iterator();
+ }
+
+ String selectorPrefix = selectorPrefixes.next();
+
+
+ var sourceInfo = new SourceInfo("", selectorPrefix);
+ var schemaInfo = new SchemaInfo(null, new ArrayList<>());
+
+ return EventFactory.fromMap(rawEvent, sourceInfo, schemaInfo);
+ }
+}
diff --git
a/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/TestConfiguration.java
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/TestConfiguration.java
new file mode 100644
index 0000000000..fb2270f3a4
--- /dev/null
+++
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/TestConfiguration.java
@@ -0,0 +1,47 @@
+/*
+ * 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.test.executors;
+
+import java.util.List;
+import java.util.Map;
+
+public class TestConfiguration {
+
+ private final Map<String, Object> fieldConfiguration;
+
+ private final List<String> prefixes;
+
+ public TestConfiguration(Map<String, Object> fieldConfiguration,
+ List<String> eventPrefixes) {
+ this.fieldConfiguration = fieldConfiguration;
+ this.prefixes = eventPrefixes;
+ }
+
+ public Map<String, Object> getFieldConfiguration() {
+ return fieldConfiguration;
+ }
+
+ public List<String> getPrefixes() {
+ return prefixes;
+ }
+
+ public static TestConfigurationBuilder builder() {
+ return new TestConfigurationBuilder();
+ }
+}
diff --git
a/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/TestConfigurationBuilder.java
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/TestConfigurationBuilder.java
new file mode 100644
index 0000000000..e2260c5366
--- /dev/null
+++
b/streampipes-test-utils-executors/src/main/java/org/apache/streampipes/test/executors/TestConfigurationBuilder.java
@@ -0,0 +1,62 @@
+/*
+ * 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.test.executors;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TestConfigurationBuilder{
+ private Map<String, Object> fieldConfiguration = new HashMap<>();
+ private List<String> eventPrefixes = List.of("");
+
+ public TestConfigurationBuilder config(String key, Object value){
+ this.fieldConfiguration.put(key, value);
+ return this;
+ }
+ public TestConfigurationBuilder configWithPrefix(String key, Object value,
String prefix){
+ this.fieldConfiguration.put(key, prefix + "::" + value);
+ return this;
+ }
+
+ public TestConfigurationBuilder configWithDefaultPrefix(String key, Object
value){
+ return this.configWithPrefix(key, value, "");
+ }
+
+ public TestConfigurationBuilder config(Map<String, Object> config){
+ this.fieldConfiguration = config;
+ return this;
+ }
+
+ public TestConfigurationBuilder prefixStrategy(PrefixStrategy strategy){
+ this.eventPrefixes = switch (strategy){
+ case SAME_PREFIX -> List.of("s0");
+ case ALTERNATE -> List.of("s0", "s1");
+ };
+ return this;
+ }
+
+ public TestConfigurationBuilder customPrefixStrategy(List<String>
eventPrefixes) {
+ this.eventPrefixes = eventPrefixes;
+ return this;
+ }
+
+ public TestConfiguration build(){
+ return new TestConfiguration(this.fieldConfiguration, this.eventPrefixes);
+ }
+}
\ No newline at end of file
