Repository: apache/streampipes
Pull request #2840 (2024-05-08 07:01:10)

tests(#2822): remove pipeline export test
<!--
  ~ 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.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->
Removes test for feature deleted in #2822 

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no


Patch:
diff --git a/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml b/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml
index 884581a1cd..5e7493d707 100644
--- a/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml
+++ b/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml
@@ -52,6 +52,11 @@
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-test-utils</artifactId>
diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java
index 4838f16a56..9b70a74a49 100644
--- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java
+++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java
@@ -78,7 +78,7 @@ public class SiddhiAppGenerator {
         .getQueries()
         .forEach(query -> this.siddhiAppString.append(query).append("\n"));
 
-    LOG.info("Registering statement: \n" + this.siddhiAppString);
+    LOG.info("Registering statement: \n{}", this.siddhiAppString);
 
   }
 }
diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java
index 175f53e2c1..4d59c1fe8e 100644
--- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java
+++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java
@@ -19,10 +19,16 @@ package org.apache.streampipes.wrapper.siddhi.model;
 
 public class EventPropertyDef {
 
+  public static final String WHITESPACE_REPLACEMENT = "__w__";
+
   private String selectorPrefix;
   private final String fieldName;
   private final String fieldType;
 
+  public static String toOriginalFieldName(String sanitizedFieldName) {
+    return sanitizedFieldName.replaceAll(WHITESPACE_REPLACEMENT, " ");
+  }
+
   public EventPropertyDef(String fieldName,
                           String fieldType) {
     this.fieldName = fieldName;
@@ -40,7 +46,7 @@ public class EventPropertyDef {
   }
 
   public String getFieldName() {
-    return fieldName;
+    return fieldName.replaceAll(" ", WHITESPACE_REPLACEMENT);
   }
 
   public String getFieldType() {
diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java
index 041bbce894..e2e589dc41 100644
--- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java
+++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java
@@ -19,10 +19,12 @@ package org.apache.streampipes.wrapper.siddhi.utils;
 
 
 import org.apache.streampipes.extensions.api.pe.param.IDataProcessorParameters;
+import org.apache.streampipes.model.constants.PropertySelectorConstants;
 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.wrapper.siddhi.constants.SiddhiConstants;
+import org.apache.streampipes.wrapper.siddhi.model.EventPropertyDef;
 
 import io.siddhi.core.event.Event;
 import io.siddhi.query.api.definition.Attribute;
@@ -68,7 +70,7 @@ public class SiddhiUtils {
         outputKey = outputKey.substring(2);
       }
       Object data = event.getData(i);
-      outMap.put(outputKey, data);
+      outMap.put(EventPropertyDef.toOriginalFieldName(outputKey), data);
     }
 
     return outMap;
@@ -101,11 +103,13 @@ public class SiddhiUtils {
     return eventName
         .replaceAll("\\.", "")
         .replaceAll("-", "")
-        .replaceAll("::", "");
+        .replaceAll(PropertySelectorConstants.PROPERTY_DELIMITER, "");
   }
 
   public static String prepareProperty(String propertyName) {
-    return propertyName.replaceAll("::", "");
+    return propertyName
+        .replaceAll(PropertySelectorConstants.PROPERTY_DELIMITER, "")
+        .replaceAll(" ", EventPropertyDef.WHITESPACE_REPLACEMENT);
   }
 
 }


Test patch:
diff --git a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
index a77552bfbb..9b96316a35 100644
--- a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
+++ b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
@@ -17,141 +17,161 @@
  */
 package org.apache.streampipes.processors.siddhi.trend;
 
-//@RunWith(Parameterized.class)
+
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+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.helpers.Tuple2;
+import org.apache.streampipes.test.generator.EventStreamGenerator;
+import org.apache.streampipes.test.generator.InvocationGraphGenerator;
+import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.apache.streampipes.wrapper.params.generator.DataProcessorParameterGenerator;
+import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
+
+import org.junit.jupiter.api.Assertions;
+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.Collectors;
+import java.util.stream.Stream;
+
+
 public class TestTrendProcessor {
-//
-//  private static final Logger LOG = LoggerFactory.getLogger(TrendProcessor.class);
-//
-//  @org.junit.runners.Parameterized.Parameters
-//  public static Iterable<Object[]> data() {
-//    return Arrays.asList(new Object[][]{
-//        {1, 100, TrendOperator.INCREASE, Arrays.asList(new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2)), 1},
-//        {1, 100, TrendOperator.INCREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2))
-//            , 1},
-//        {1, 100, TrendOperator.DECREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 0))
-//            , 1},
-//        {1, 100, TrendOperator.INCREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2))
-//            , 2},
-//        {1, 200, TrendOperator.INCREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2))
-//            , 0},
-//
-//    });
-//  }
-//
-//  @org.junit.runners.Parameterized.Parameter
-//  public Integer timeWindow;
-//
-//  @org.junit.runners.Parameterized.Parameter(1)
-//  public Integer increase;
-//
-//  @org.junit.runners.Parameterized.Parameter(2)
-//  public TrendOperator trendOperator;
-//
-//  @org.junit.runners.Parameterized.Parameter(3)
-//  public List<Tuple2<Integer, Integer>> eventSettings;
-//
-//  @org.junit.runners.Parameterized.Parameter(4)
-//  public Integer expectedMatchCount;
-//
-//  @Test
-//  public void testTrend() {
-//    final Integer[] actualMatchCount = {0};
-//    DataProcessorDescription originalGraph = new TrendProcessor().declareModel();
-//    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-//    DataProcessorInvocation graph =
-//        InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-//
-//    graph.setInputStreams(Collections
-//        .singletonList(EventStreamGenerator
-//            .makeStreamWithProperties(Collections.singletonList("randomValue"))));
-//
-//    graph.setOutputStrategies(
-//        graph.getOutputStrategies()
-//            .stream()
-//            .filter(o -> o instanceof CustomOutputStrategy)
-//            .peek(o -> ((CustomOutputStrategy) o).setSelectedPropertyKeys(Arrays.asList("s0::randomValue")))
-//            .collect(Collectors.toList())
-//    );
-//
-//    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("randomValue")));
-//
-//    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
-//        .setActualTopicName("output-topic");
-//
-//    var visitor = new TrendConfigurationVisitor("s0::randomValue",
-//        trendOperator,
-//        increase,
-//        timeWindow);
-//    graph.getStaticProperties().forEach(sp -> sp.accept(visitor));
-//
-//    var processorParams = new DataProcessorParameterGenerator().makeParameters(graph);
-//
-//    SiddhiDebugCallback callback = new SiddhiDebugCallback() {
-//      @Override
-//      public void onEvent(io.siddhi.core.event.Event event) {
-//        actualMatchCount[0]++;
-//      }
-//
-//      @Override
-//      public void onEvent(List<io.siddhi.core.event.Event> events) {
-//
-//      }
-//    };
-//
-//    TrendProcessor trend = new TrendProcessor(callback);
-//    trend.onPipelineStarted(processorParams, null, null);
-//
-//    sendEvents(trend);
-//    LOG.info("Expected match count is {}", expectedMatchCount);
-//    LOG.info("Actual match count is {}", actualMatchCount[0]);
-//    assertEquals(expectedMatchCount, actualMatchCount[0]);
-//  }
-//
-//  private void sendEvents(TrendProcessor trend) {
-//    List<Tuple2<Integer, Event>> events = makeEvents();
-//    for (Tuple2<Integer, Event> event : events) {
-//      LOG.info("Sending event with value " + event.v.getFieldBySelector("s0::randomValue"));
-//      trend.onEvent(event.v, null);
-//      try {
-//        Thread.sleep(event.k);
-//      } catch (InterruptedException e) {
-//        e.printStackTrace();
-//      }
-//    }
-//  }
-//
-//  private List<Tuple2<Integer, Event>> makeEvents() {
-//    List<Tuple2<Integer, Event>> events = new ArrayList<>();
-//    for (Tuple2<Integer, Integer> eventSetting : eventSettings) {
-//      events.add(makeEvent(eventSetting.k, eventSetting.v));
-//    }
-//    return events;
-//  }
-//
-//  private Tuple2<Integer, Event> makeEvent(Integer timeout, Integer value) {
-//    Map<String, Object> map = new HashMap<>();
-//    map.put("randomValue", value);
-//    return new Tuple2<>(timeout, EventFactory.fromMap(map, new SourceInfo("test"
-//            + "-topic", "s0"),
-//        new SchemaInfo(null,
-//            new ArrayList<>())));
-//  }
+
+  private static final Logger LOG = LoggerFactory.getLogger(TrendProcessor.class);
+
+  static Stream<Arguments> data() {
+    return Stream.of(
+        Arguments.of(1, 100, TrendOperator.INCREASE, Arrays.asList(new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2)), 1),
+        Arguments.of(1, 100, TrendOperator.INCREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2))
+            , 1),
+        Arguments.of(1, 100, TrendOperator.DECREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 0))
+            , 1),
+        Arguments.of(1, 100, TrendOperator.INCREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2))
+            , 2),
+        Arguments.of(1, 200, TrendOperator.INCREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2))
+            , 0)
+    );
+  }
+
+  @ParameterizedTest
+  @MethodSource("data")
+  public void testTrend(int timeWindow,
+                        int increase,
+                        TrendOperator trendOperator,
+                        List<Tuple2<Integer, Integer>> eventSettings,
+                        int expectedMatchCount) {
+    final Integer[] actualMatchCount = {0};
+    DataProcessorDescription originalGraph = new TrendProcessor().declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph =
+        InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+
+    graph.setInputStreams(Collections
+        .singletonList(EventStreamGenerator
+            .makeStreamWithProperties(Collections.singletonList("randomValue"))));
+
+    graph.setOutputStrategies(
+        graph.getOutputStrategies()
+            .stream()
+            .filter(o -> o instanceof CustomOutputStrategy)
+            .peek(o -> ((CustomOutputStrategy) o).setSelectedPropertyKeys(List.of("s0::randomValue")))
+            .collect(Collectors.toList())
+    );
+
+    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("randomValue")));
+
+    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
+        .setActualTopicName("output-topic");
+
+    var visitor = new TrendConfigurationVisitor("s0::randomValue",
+        trendOperator,
+        increase,
+        timeWindow);
+    graph.getStaticProperties().forEach(sp -> sp.accept(visitor));
+
+    var processorParams = new DataProcessorParameterGenerator().makeParameters(graph);
+
+    SiddhiDebugCallback callback = new SiddhiDebugCallback() {
+      @Override
+      public void onEvent(io.siddhi.core.event.Event event) {
+        actualMatchCount[0]++;
+      }
+
+      @Override
+      public void onEvent(List<io.siddhi.core.event.Event> events) {
+
+      }
+    };
+
+    TrendProcessor trend = new TrendProcessor(callback);
+    trend.onPipelineStarted(processorParams, null, null);
+
+    sendEvents(trend, eventSettings);
+    LOG.info("Expected match count is {}", expectedMatchCount);
+    LOG.info("Actual match count is {}", actualMatchCount[0]);
+    Assertions.assertEquals(expectedMatchCount, actualMatchCount[0]);
+  }
+
+  private void sendEvents(TrendProcessor trend,
+                          List<Tuple2<Integer, Integer>> eventSettings) {
+    List<Tuple2<Integer, Event>> events = makeEvents(eventSettings);
+    for (Tuple2<Integer, Event> event : events) {
+      LOG.info("Sending event with value " + event.v.getFieldBySelector("s0::randomValue"));
+      trend.onEvent(event.v, null);
+      try {
+        Thread.sleep(event.k);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  private List<Tuple2<Integer, Event>> makeEvents(List<Tuple2<Integer, Integer>> eventSettings) {
+    List<Tuple2<Integer, Event>> events = new ArrayList<>();
+    for (Tuple2<Integer, Integer> eventSetting : eventSettings) {
+      events.add(makeEvent(eventSetting.k, eventSetting.v));
+    }
+    return events;
+  }
+
+  private Tuple2<Integer, Event> makeEvent(Integer timeout, Integer value) {
+    Map<String, Object> map = new HashMap<>();
+    map.put("randomValue", value);
+    return new Tuple2<>(timeout, EventFactory.fromMap(map, new SourceInfo("test"
+            + "-topic", "s0"),
+        new SchemaInfo(null,
+            new ArrayList<>())));
+  }
 }
diff --git a/ui/cypress/tests/pipeline/pipelineExport.spec.ts b/ui/cypress/tests/pipeline/pipelineExport.spec.ts
deleted file mode 100644
index d125d56307..0000000000
--- a/ui/cypress/tests/pipeline/pipelineExport.spec.ts
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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.
- *
- */
-
-import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
-import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
-import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
-
-const adapterName = 'simulator';
-
-describe('Test Export and Import Functionality of Pipelines', () => {
-    beforeEach('Setup Test', () => {
-        cy.initStreamPipesTest();
-        ConnectUtils.addMachineDataSimulator(adapterName);
-        const pipelineInput = PipelineBuilder.create('Pipeline Test')
-            .addSource(adapterName)
-            .addProcessingElement(
-                PipelineElementBuilder.create('field_renamer')
-                    .addInput('drop-down', 'convert-property', 'timestamp')
-                    .addInput('input', 'field-name', 't')
-                    .build(),
-            )
-            .addSink(
-                PipelineElementBuilder.create('data_lake')
-                    .addInput('input', 'db_measurement', 'demo')
-                    .build(),
-            )
-            .build();
-
-        PipelineUtils.addPipeline(pipelineInput);
-    });
-
-    it('Perform Test', () => {
-        PipelineUtils.goToPipelines();
-
-        // Check if pipeline exists
-        cy.dataCy('all-pipelines-table').should('have.length', 1);
-
-        // Export Pipeline
-        cy.dataCy('export-pipelines-btn').click();
-
-        // Delete created pipeline
-        PipelineUtils.deletePipeline();
-        cy.dataCy('all-pipelines-table').should('have.length', 0);
-
-        // Import pipeline
-        cy.dataCy('import-pipelines-btn').click();
-        cy.get('input[type="file"]').selectFile(
-            'cypress/downloads/pipelines.json',
-            { force: true },
-        );
-        // Select pipeline checkbox
-        cy.get('[type="checkbox"]').check();
-        cy.dataCy('import-selected-pipelines-btn').click();
-
-        // Check if pipeline import was succesful
-        cy.dataCy('all-pipelines-table').should('have.length', 1);
-
-        // Start pipeline and check if pipeline has started correctly
-        cy.dataCy('start-pipeline-button').click();
-        cy.get('[class="success-message"]').should('be.visible');
-        cy.dataCy('sp-pipeline-dialog-close').click();
-    });
-});