Repository: apache/streampipes
Pull request #2030 (2023-10-16 03:04:06)

chore(deps-dev): bump pre-commit from 3.4.0 to 3.5.0 in /streampipes-client-python
Bumps [pre-commit](https://github.com/pre-commit/pre-commit) from 3.4.0 to 3.5.0.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/pre-commit/pre-commit/releases">pre-commit's releases</a>.</em></p>
<blockquote>
<h2>pre-commit v3.5.0</h2>
<h3>Features</h3>
<ul>
<li>Improve performance of <code>check-hooks-apply</code> and <code>check-useless-excludes</code>.
<ul>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/2998">#2998</a> PR by <a href="https://github.com/mxr"><code>@​mxr</code></a>.</li>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/2935">#2935</a> issue by <a href="https://github.com/mxr"><code>@​mxr</code></a>.</li>
</ul>
</li>
</ul>
<h3>Fixes</h3>
<ul>
<li>Use <code>time.monotonic()</code> for more accurate hook timing.
<ul>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/3024">#3024</a> PR by <a href="https://github.com/adamchainz"><code>@​adamchainz</code></a>.</li>
</ul>
</li>
</ul>
<h3>Migrating</h3>
<ul>
<li>Require npm 6.x+ for <code>language: node</code> hooks.
<ul>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/2996">#2996</a> PR by <a href="https://github.com/RoelAdriaans"><code>@​RoelAdriaans</code></a>.</li>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/1983">#1983</a> issue by <a href="https://github.com/henryiii"><code>@​henryiii</code></a>.</li>
</ul>
</li>
</ul>
</blockquote>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/pre-commit/pre-commit/blob/main/CHANGELOG.md">pre-commit's changelog</a>.</em></p>
<blockquote>
<h1>3.5.0 - 2023-10-13</h1>
<h3>Features</h3>
<ul>
<li>Improve performance of <code>check-hooks-apply</code> and <code>check-useless-excludes</code>.
<ul>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/2998">#2998</a> PR by <a href="https://github.com/mxr"><code>@​mxr</code></a>.</li>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/2935">#2935</a> issue by <a href="https://github.com/mxr"><code>@​mxr</code></a>.</li>
</ul>
</li>
</ul>
<h3>Fixes</h3>
<ul>
<li>Use <code>time.monotonic()</code> for more accurate hook timing.
<ul>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/3024">#3024</a> PR by <a href="https://github.com/adamchainz"><code>@​adamchainz</code></a>.</li>
</ul>
</li>
</ul>
<h3>Migrating</h3>
<ul>
<li>Require npm 6.x+ for <code>language: node</code> hooks.
<ul>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/2996">#2996</a> PR by <a href="https://github.com/RoelAdriaans"><code>@​RoelAdriaans</code></a>.</li>
<li><a href="https://redirect.github.com/pre-commit/pre-commit/issues/1983">#1983</a> issue by <a href="https://github.com/henryiii"><code>@​henryiii</code></a>.</li>
</ul>
</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/pre-commit/pre-commit/commit/61cc55a59cc63c7405dd3cd7c96b169fdb750333"><code>61cc55a</code></a> v3.5.0</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/c9945b9aa3d90113ad2bce4ab6648ef637edbf7c"><code>c9945b9</code></a> Merge pull request <a href="https://redirect.github.com/pre-commit/pre-commit/issues/3029">#3029</a> from adamchainz/improve_duration_timing</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/d988767b414495bdab9ea24532ad337e8ee3fd1f"><code>d988767</code></a> Improve hook duration timing</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/0d8b2451ca61d93c3d26746da83b63c7821c73f4"><code>0d8b245</code></a> Merge pull request <a href="https://redirect.github.com/pre-commit/pre-commit/issues/3023">#3023</a> from pre-commit/pre-commit-ci-update-config</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/155c52134848b05b0092a446cdd2c336a03a85c0"><code>155c521</code></a> [pre-commit.ci] pre-commit autoupdate</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/676e51aa5edd91fd8749ea4c7ebe65ac2609c5e9"><code>676e51a</code></a> Merge pull request <a href="https://redirect.github.com/pre-commit/pre-commit/issues/3024">#3024</a> from pre-commit/pick-shebang-path-without-spaces</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/997ea0ad52074c3e6474f3d99f76f7965e2d05f0"><code>997ea0a</code></a> use sys.executable instead of echo.exe in parse_shebang</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/19aa121db0e8e7f15dd88765735657e73f94e7a6"><code>19aa121</code></a> Merge pull request <a href="https://redirect.github.com/pre-commit/pre-commit/issues/3016">#3016</a> from pre-commit/pre-commit-ci-update-config</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/a4ab977cc36e06fff8a8c69cca652162407b55cf"><code>a4ab977</code></a> [pre-commit.ci] pre-commit autoupdate</li>
<li><a href="https://github.com/pre-commit/pre-commit/commit/3f3760b86c4e08de82d00b7ba55505f7d0656271"><code>3f3760b</code></a> Merge pull request <a href="https://redirect.github.com/pre-commit/pre-commit/issues/3011">#3011</a> from hack3ric/bump-node-and-go-version</li>
<li>Additional commits viewable in <a href="https://github.com/pre-commit/pre-commit/compare/v3.4.0...v3.5.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=pre-commit&package-manager=pip&previous-version=3.4.0&new-version=3.5.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


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-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;


Test patch:
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