Repository: apache/streampipes (java)

Issue:
Python client error when using Port 443
### Body

When I used the python client with the following configuration:
 ```python
config = StreamPipesClientConfig(
    ...
    https_disabled=True,
    port=443
)
```
I got this error message:
```
HTTPError: 
Oops, there seems to be an issue with the Python Client calling the API inappropriately.
This should not happen, but unfortunately did.
If you don't mind, it would be awesome to let us know by creating an issue at https://github.com/apache/streampipes.
Please paste the following information to the issue description:
```

when I changed the value of `https_disabled` to `False` everything worked as expected.
Maybe we can change the returned error message. It should state that the developer should check the configuration settings.

### StreamPipes Committer

I acknowledge that I am a maintainer/committer of the Apache StreamPipes project.

Pull request:
fix(#1670): validate client config and create proper error message
<!--
  ~ 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.
-->
fix #1670 

### 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-client-python/poetry.lock b/streampipes-client-python/poetry.lock
index 1d2d712c5e..f1136c22de 100644
--- a/streampipes-client-python/poetry.lock
+++ b/streampipes-client-python/poetry.lock
@@ -1983,13 +1983,13 @@ testing = ["pytest", "pytest-benchmark"]
 
 [[package]]
 name = "pre-commit"
-version = "3.4.0"
+version = "3.5.0"
 description = "A framework for managing and maintaining multi-language pre-commit hooks."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "pre_commit-3.4.0-py2.py3-none-any.whl", hash = "sha256:96d529a951f8b677f730a7212442027e8ba53f9b04d217c4c67dc56c393ad945"},
-    {file = "pre_commit-3.4.0.tar.gz", hash = "sha256:6bbd5129a64cad4c0dfaeeb12cd8f7ea7e15b77028d985341478c8af3c759522"},
+    {file = "pre_commit-3.5.0-py2.py3-none-any.whl", hash = "sha256:841dc9aef25daba9a0238cd27984041fa0467b4199fc4852e27950664919f660"},
+    {file = "pre_commit-3.5.0.tar.gz", hash = "sha256:5804465c675b659b0862f07907f96295d490822a450c4c40e747d0b1c6ebcb32"},
 ]
 
 [package.dependencies]
@@ -3290,4 +3290,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = ">3.8.1,<3.12"
-content-hash = "f523de24f08c24e6ea7d26e792c7092d630b0451597f656a7ed19292d01201bb"
+content-hash = "41a26070e9f75bd6c9cabed820e02dc87fdae3e8053e4f20997be7d4796878ef"
diff --git a/streampipes-client-python/pyproject.toml b/streampipes-client-python/pyproject.toml
index 8b9cae9c83..3d6d73aabf 100644
--- a/streampipes-client-python/pyproject.toml
+++ b/streampipes-client-python/pyproject.toml
@@ -73,7 +73,7 @@ interrogate = { version = "1.5.0", extras = ["png"] }
 isort = "5.12.0"
 mypy = "1.6.0"
 ruff = "0.0.280"
-pre-commit = "3.4.0"
+pre-commit = "3.5.0"
 pytest = "7.4.0"
 pytest-cov = "4.1.0"
 pyupgrade = "3.15.0"
diff --git a/streampipes-client-python/streampipes/client/client.py b/streampipes-client-python/streampipes/client/client.py
index 7d613d8704..047e52cc89 100644
--- a/streampipes-client-python/streampipes/client/client.py
+++ b/streampipes-client-python/streampipes/client/client.py
@@ -65,6 +65,11 @@ class StreamPipesClient:
     dataStreamApi: DataStreamEndpoint
         Instance of the data stream endpoint
 
+    Raises
+    ------
+    AttributeError:
+        In case an invalid configuration of the `StreamPipesClientConfig` is passed
+
     Examples
     --------
 
@@ -117,6 +122,15 @@ class StreamPipesClient:
         client_config: StreamPipesClientConfig,
         logging_level: Optional[int] = logging.INFO,
     ):
+        # validate client config
+        # `https_disabled` and `port` 443 is an invalid configuration
+        if client_config.https_disabled and client_config.port == 443:
+            raise AttributeError(
+                "Invalid configuration passed! The given client configuration has "
+                "`https_disabled` set to `True` and `port` set to `443`.\n "
+                "If you want to connect to port 443, use `https_disabled=False` or "
+                "alternatively connect to port `80`."
+            )
         self.client_config = client_config
 
         # set up a requests session
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
index fcc5c7e229..0edb33b964 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
@@ -40,10 +40,10 @@ import java.util.List;
 
 public class MergeByTimeProcessor extends StreamPipesDataProcessor {
 
-  private static final String TIMESTAMP_MAPPING_STREAM_1_KEY = "timestamp_mapping_stream_1";
-  private static final String TIMESTAMP_MAPPING_STREAM_2_KEY = "timestamp_mapping_stream_2";
-  private static final String NUMBER_MAPPING = "number_mapping";
-  private static final String TIME_INTERVAL = "time-interval";
+  protected static final String TIMESTAMP_MAPPING_STREAM_1_KEY = "timestamp_mapping_stream_1";
+  protected static final String TIMESTAMP_MAPPING_STREAM_2_KEY = "timestamp_mapping_stream_2";
+  protected static final String NUMBER_MAPPING = "number_mapping";
+  protected static final String TIME_INTERVAL = "time-interval";
 
   private List<String> outputKeySelectors;
   private String timestampFieldStream0;
diff --git a/ui/package-lock.json b/ui/package-lock.json
index 3be0585a6d..bbd1db46e6 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -82,7 +82,7 @@
         "@cypress/webpack-preprocessor": "5.17.0",
         "@ngtools/webpack": "^15.2.9",
         "@types/angular": "^1.8.4",
-        "@types/jasmine": "~4.3.1",
+        "@types/jasmine": "~5.1.0",
         "@types/jqueryui": "^1.12.16",
         "@types/jshint": "^2.12.1",
         "@types/leaflet": "^1.9.0",
@@ -7088,9 +7088,9 @@
       }
     },
     "node_modules/@types/jasmine": {
-      "version": "4.3.1",
-      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-4.3.1.tgz",
-      "integrity": "sha512-Vu8l+UGcshYmV1VWwULgnV/2RDbBaO6i2Ptx7nd//oJPIZGhoI1YLST4VKagD2Pq/Bc2/7zvtvhM7F3p4SN7kQ==",
+      "version": "5.1.0",
+      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-5.1.0.tgz",
+      "integrity": "sha512-XOV0KsqXNX2gUSqk05RWeolIMgaAQ7+l/ozOBoQ8NGwLg+E7J9vgagODtNgfim4jCzEUP0oJ3gnXeC+Zv+Xi1A==",
       "dev": true
     },
     "node_modules/@types/jquery": {
@@ -33617,9 +33617,9 @@
       }
     },
     "@types/jasmine": {
-      "version": "4.3.1",
-      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-4.3.1.tgz",
-      "integrity": "sha512-Vu8l+UGcshYmV1VWwULgnV/2RDbBaO6i2Ptx7nd//oJPIZGhoI1YLST4VKagD2Pq/Bc2/7zvtvhM7F3p4SN7kQ==",
+      "version": "5.1.0",
+      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-5.1.0.tgz",
+      "integrity": "sha512-XOV0KsqXNX2gUSqk05RWeolIMgaAQ7+l/ozOBoQ8NGwLg+E7J9vgagODtNgfim4jCzEUP0oJ3gnXeC+Zv+Xi1A==",
       "dev": true
     },
     "@types/jquery": {
diff --git a/ui/package.json b/ui/package.json
index 002515edb1..5197195e07 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -104,7 +104,7 @@
     "@cypress/webpack-preprocessor": "5.17.0",
     "@ngtools/webpack": "^15.2.9",
     "@types/angular": "^1.8.4",
-    "@types/jasmine": "~4.3.1",
+    "@types/jasmine": "~5.1.0",
     "@types/jqueryui": "^1.12.16",
     "@types/jshint": "^2.12.1",
     "@types/leaflet": "^1.9.0",


Test patch:
diff --git a/streampipes-client-python/tests/client/test_client.py b/streampipes-client-python/tests/client/test_client.py
index a9c1ffc265..260ba35ee4 100644
--- a/streampipes-client-python/tests/client/test_client.py
+++ b/streampipes-client-python/tests/client/test_client.py
@@ -79,6 +79,18 @@ class TestStreamPipesClient(TestCase):
         self.assertTrue(isinstance(result.dataLakeMeasureApi, DataLakeMeasureEndpoint))
         self.assertEqual(result.base_api_path, "https://localhost:443/streampipes-backend/")
 
+    def test_client_create_invalid_config(self):
+
+        with self.assertRaises(AttributeError):
+            StreamPipesClient.create(
+                client_config=StreamPipesClientConfig(
+                    credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"),
+                    host_address="localhost",
+                    https_disabled=True,
+                    port=443,
+                )
+            )
+
     @patch("streampipes.client.client.logger", autospec=True)
     @patch("streampipes.endpoint.endpoint.APIEndpoint._make_request", autospec=True)
     def test_client_describe(self, make_request: MagicMock, mocked_logger: MagicMock):
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
new file mode 100644
index 0000000000..e2efbd6957
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
@@ -0,0 +1,193 @@
+/*
+ * 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.merge;
+
+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.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.test.extensions.api.StoreEventCollector;
+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.compat.ProcessorParams;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+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()));
+  }
+}
\ No newline at end of file