Andrew Shvayka
2 years ago
committed by
GitHub
GitHub
35 changed files with 737 additions and 79 deletions
@ -0,0 +1,204 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.server.msa.rule.node; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import lombok.Data; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.awaitility.Awaitility; |
||||
|
import org.eclipse.paho.client.mqttv3.IMqttMessageListener; |
||||
|
import org.eclipse.paho.client.mqttv3.MqttClient; |
||||
|
import org.eclipse.paho.client.mqttv3.MqttConnectOptions; |
||||
|
import org.eclipse.paho.client.mqttv3.MqttMessage; |
||||
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; |
||||
|
import org.testng.annotations.AfterMethod; |
||||
|
import org.testng.annotations.BeforeMethod; |
||||
|
import org.testng.annotations.Test; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.EventInfo; |
||||
|
import org.thingsboard.server.common.data.StringUtils; |
||||
|
import org.thingsboard.server.common.data.event.EventType; |
||||
|
import org.thingsboard.server.common.data.id.RuleChainId; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.page.PageLink; |
||||
|
import org.thingsboard.server.common.data.page.TimePageLink; |
||||
|
import org.thingsboard.server.common.data.rule.NodeConnectionInfo; |
||||
|
import org.thingsboard.server.common.data.rule.RuleChain; |
||||
|
import org.thingsboard.server.common.data.rule.RuleChainMetaData; |
||||
|
import org.thingsboard.server.common.data.rule.RuleNode; |
||||
|
import org.thingsboard.server.common.data.security.DeviceCredentials; |
||||
|
import org.thingsboard.server.msa.AbstractContainerTest; |
||||
|
import org.thingsboard.server.msa.DisableUIListeners; |
||||
|
import org.thingsboard.server.msa.TestProperties; |
||||
|
import org.thingsboard.server.msa.WsClient; |
||||
|
import org.thingsboard.server.msa.mapper.WsTelemetryResponse; |
||||
|
|
||||
|
import java.util.Arrays; |
||||
|
import java.util.List; |
||||
|
import java.util.Objects; |
||||
|
import java.util.Optional; |
||||
|
import java.util.concurrent.ArrayBlockingQueue; |
||||
|
import java.util.concurrent.BlockingQueue; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.testng.Assert.fail; |
||||
|
import static org.thingsboard.server.msa.prototypes.DevicePrototypes.defaultDevicePrototype; |
||||
|
|
||||
|
@DisableUIListeners |
||||
|
@Slf4j |
||||
|
public class MqttNodeTest extends AbstractContainerTest { |
||||
|
|
||||
|
private static final String TOPIC = "tb/mqtt/device"; |
||||
|
|
||||
|
private Device device; |
||||
|
|
||||
|
@BeforeMethod |
||||
|
public void setUp() { |
||||
|
testRestClient.login("tenant@thingsboard.org", "tenant"); |
||||
|
device = testRestClient.postDevice("", defaultDevicePrototype("mqtt_")); |
||||
|
} |
||||
|
|
||||
|
@AfterMethod |
||||
|
public void tearDown() { |
||||
|
testRestClient.deleteDeviceIfExists(device.getId()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void telemetryUpload() throws Exception { |
||||
|
RuleChainId defaultRuleChainId = getDefaultRuleChainId(); |
||||
|
|
||||
|
createRootRuleChainWithTestNode("MqttRuleNodeTestMetadata.json", "org.thingsboard.rule.engine.mqtt.TbMqttNode", 2); |
||||
|
|
||||
|
DeviceCredentials deviceCredentials = testRestClient.getDeviceCredentialsByDeviceId(device.getId()); |
||||
|
|
||||
|
WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); |
||||
|
|
||||
|
MqttMessageListener messageListener = new MqttMessageListener(); |
||||
|
MqttClient responseClient = new MqttClient(TestProperties.getMqttBrokerUrl(), StringUtils.randomAlphanumeric(10), new MemoryPersistence()); |
||||
|
responseClient.connect(); |
||||
|
responseClient.subscribe(TOPIC, messageListener); |
||||
|
|
||||
|
MqttClient mqttClient = new MqttClient("tcp://localhost:1883", StringUtils.randomAlphanumeric(10), new MemoryPersistence()); |
||||
|
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions(); |
||||
|
mqttConnectOptions.setUserName(deviceCredentials.getCredentialsId()); |
||||
|
mqttClient.connect(mqttConnectOptions); |
||||
|
mqttClient.publish("v1/devices/me/telemetry", new MqttMessage(createPayload().toString().getBytes())); |
||||
|
|
||||
|
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); |
||||
|
log.info("Received telemetry: {}", actualLatestTelemetry); |
||||
|
wsClient.closeBlocking(); |
||||
|
|
||||
|
assertThat(actualLatestTelemetry.getData()).hasSize(4); |
||||
|
assertThat(actualLatestTelemetry.getLatestValues().keySet()).containsOnlyOnceElementsOf(Arrays.asList("booleanKey", "stringKey", "doubleKey", "longKey")); |
||||
|
|
||||
|
assertThat(actualLatestTelemetry.getDataValuesByKey("booleanKey").get(1)).isEqualTo(Boolean.TRUE.toString()); |
||||
|
assertThat(actualLatestTelemetry.getDataValuesByKey("stringKey").get(1)).isEqualTo("value1"); |
||||
|
assertThat(actualLatestTelemetry.getDataValuesByKey("doubleKey").get(1)).isEqualTo(Double.toString(42.0)); |
||||
|
assertThat(actualLatestTelemetry.getDataValuesByKey("longKey").get(1)).isEqualTo(Long.toString(73)); |
||||
|
|
||||
|
Awaitility |
||||
|
.await() |
||||
|
.alias("Get integration events") |
||||
|
.atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> messageListener.getEvents().size() > 0); |
||||
|
|
||||
|
BlockingQueue<MqttEvent> events = messageListener.getEvents(); |
||||
|
JsonNode actual = JacksonUtil.toJsonNode(Objects.requireNonNull(events.poll()).message); |
||||
|
|
||||
|
assertThat(actual.get("stringKey").asText()).isEqualTo("value1"); |
||||
|
assertThat(actual.get("booleanKey").asBoolean()).isEqualTo(Boolean.TRUE); |
||||
|
assertThat(actual.get("doubleKey").asDouble()).isEqualTo(42.0); |
||||
|
assertThat(actual.get("longKey").asLong()).isEqualTo(73); |
||||
|
|
||||
|
testRestClient.setRootRuleChain(defaultRuleChainId); |
||||
|
} |
||||
|
|
||||
|
@Data |
||||
|
private class MqttMessageListener implements IMqttMessageListener { |
||||
|
private final BlockingQueue<MqttEvent> events; |
||||
|
|
||||
|
private MqttMessageListener() { |
||||
|
events = new ArrayBlockingQueue<>(100); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void messageArrived(String s, MqttMessage mqttMessage) { |
||||
|
log.info("MQTT message [{}], topic [{}]", mqttMessage.toString(), s); |
||||
|
events.add(new MqttEvent(s, mqttMessage.toString())); |
||||
|
} |
||||
|
|
||||
|
public BlockingQueue<MqttEvent> getEvents() { |
||||
|
return events; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Data |
||||
|
private class MqttEvent { |
||||
|
private final String topic; |
||||
|
private final String message; |
||||
|
} |
||||
|
|
||||
|
private RuleChainId getDefaultRuleChainId() { |
||||
|
PageData<RuleChain> ruleChains = testRestClient.getRuleChains(new PageLink(40, 0)); |
||||
|
|
||||
|
Optional<RuleChain> defaultRuleChain = ruleChains.getData() |
||||
|
.stream() |
||||
|
.filter(RuleChain::isRoot) |
||||
|
.findFirst(); |
||||
|
if (!defaultRuleChain.isPresent()) { |
||||
|
fail("Root rule chain wasn't found"); |
||||
|
} |
||||
|
return defaultRuleChain.get().getId(); |
||||
|
} |
||||
|
|
||||
|
protected RuleChainId createRootRuleChainWithTestNode(String ruleChainMetadataFile, String ruleNodeType, int eventsCount) throws Exception { |
||||
|
RuleChain newRuleChain = new RuleChain(); |
||||
|
newRuleChain.setName("testRuleChain"); |
||||
|
RuleChain ruleChain = testRestClient.postRuleChain(newRuleChain); |
||||
|
|
||||
|
JsonNode configuration = JacksonUtil.OBJECT_MAPPER.readTree(this.getClass().getClassLoader().getResourceAsStream(ruleChainMetadataFile)); |
||||
|
RuleChainMetaData ruleChainMetaData = new RuleChainMetaData(); |
||||
|
ruleChainMetaData.setRuleChainId(ruleChain.getId()); |
||||
|
ruleChainMetaData.setFirstNodeIndex(configuration.get("firstNodeIndex").asInt()); |
||||
|
ruleChainMetaData.setNodes(Arrays.asList(JacksonUtil.OBJECT_MAPPER.treeToValue(configuration.get("nodes"), RuleNode[].class))); |
||||
|
ruleChainMetaData.setConnections(Arrays.asList(JacksonUtil.OBJECT_MAPPER.treeToValue(configuration.get("connections"), NodeConnectionInfo[].class))); |
||||
|
|
||||
|
ruleChainMetaData = testRestClient.postRuleChainMetadata(ruleChainMetaData); |
||||
|
|
||||
|
testRestClient.setRootRuleChain(ruleChain.getId()); |
||||
|
|
||||
|
RuleNode node = ruleChainMetaData.getNodes().stream().filter(ruleNode -> ruleNode.getType().equals(ruleNodeType)).findFirst().get(); |
||||
|
|
||||
|
Awaitility |
||||
|
.await() |
||||
|
.alias("Get events from rule chain") |
||||
|
.atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
PageData<EventInfo> events = testRestClient.getEvents(node.getId(), EventType.LC_EVENT, ruleChain.getTenantId(), new TimePageLink(1024)); |
||||
|
List<EventInfo> eventInfos = events.getData().stream().filter(eventInfo -> |
||||
|
"STARTED".equals(eventInfo.getBody().get("event").asText()) && |
||||
|
"true".equals(eventInfo.getBody().get("success").asText())) |
||||
|
.collect(Collectors.toList()); |
||||
|
|
||||
|
return eventInfos.size() == eventsCount; |
||||
|
}); |
||||
|
|
||||
|
return ruleChain.getId(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,79 @@ |
|||||
|
{ |
||||
|
"firstNodeIndex": 2, |
||||
|
"nodes": [ |
||||
|
{ |
||||
|
"additionalInfo": { |
||||
|
"description": "", |
||||
|
"layoutX": 626, |
||||
|
"layoutY": 152 |
||||
|
}, |
||||
|
"type": "org.thingsboard.rule.engine.mqtt.TbMqttNode", |
||||
|
"name": "test mqtt", |
||||
|
"debugMode": true, |
||||
|
"singletonMode": true, |
||||
|
"queueName": "HighPriority", |
||||
|
"configurationVersion": 0, |
||||
|
"configuration": { |
||||
|
"topicPattern": "tb/mqtt/device", |
||||
|
"host": "broker", |
||||
|
"port": 1883, |
||||
|
"connectTimeoutSec": 10, |
||||
|
"clientId": null, |
||||
|
"cleanSession": true, |
||||
|
"retainedMessage": false, |
||||
|
"ssl": false, |
||||
|
"credentials": { |
||||
|
"type": "anonymous" |
||||
|
} |
||||
|
}, |
||||
|
"externalId": null |
||||
|
}, |
||||
|
{ |
||||
|
"additionalInfo": { |
||||
|
"description": "", |
||||
|
"layoutX": 949, |
||||
|
"layoutY": 153 |
||||
|
}, |
||||
|
"type": "org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode", |
||||
|
"name": "save timeseries", |
||||
|
"debugMode": true, |
||||
|
"singletonMode": false, |
||||
|
"configurationVersion": 0, |
||||
|
"configuration": { |
||||
|
"defaultTTL": 0, |
||||
|
"skipLatestPersistence": false, |
||||
|
"useServerTs": false |
||||
|
}, |
||||
|
"externalId": null |
||||
|
}, |
||||
|
{ |
||||
|
"additionalInfo": { |
||||
|
"description": "", |
||||
|
"layoutX": 305, |
||||
|
"layoutY": 151 |
||||
|
}, |
||||
|
"type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode", |
||||
|
"name": "switch", |
||||
|
"debugMode": false, |
||||
|
"singletonMode": false, |
||||
|
"configurationVersion": 0, |
||||
|
"configuration": { |
||||
|
"version": 0 |
||||
|
}, |
||||
|
"externalId": null |
||||
|
} |
||||
|
], |
||||
|
"connections": [ |
||||
|
{ |
||||
|
"fromIndex": 0, |
||||
|
"toIndex": 1, |
||||
|
"type": "Success" |
||||
|
}, |
||||
|
{ |
||||
|
"fromIndex": 2, |
||||
|
"toIndex": 0, |
||||
|
"type": "Post telemetry" |
||||
|
} |
||||
|
], |
||||
|
"ruleChainConnections": null |
||||
|
} |
||||
@ -0,0 +1,25 @@ |
|||||
|
# |
||||
|
# Copyright © 2016-2024 The Thingsboard Authors |
||||
|
# |
||||
|
# Licensed 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. |
||||
|
# |
||||
|
|
||||
|
version: '3.0' |
||||
|
services: |
||||
|
broker: |
||||
|
image: eclipse-mosquitto |
||||
|
volumes: |
||||
|
- ./mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf |
||||
|
ports: |
||||
|
- "1883" |
||||
|
restart: always |
||||
@ -0,0 +1,18 @@ |
|||||
|
# |
||||
|
# Copyright © 2016-2024 The Thingsboard Authors |
||||
|
# |
||||
|
# Licensed 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. |
||||
|
# |
||||
|
|
||||
|
listener 1883 |
||||
|
allow_anonymous true |
||||
@ -1,30 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2024 The Thingsboard Authors |
|
||||
* |
|
||||
* Licensed 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.thingsboard.rule.engine.flow; |
|
||||
|
|
||||
import lombok.Data; |
|
||||
import org.thingsboard.rule.engine.api.NodeConfiguration; |
|
||||
|
|
||||
@Data |
|
||||
public class TbCheckpointNodeConfiguration implements NodeConfiguration<TbCheckpointNodeConfiguration> { |
|
||||
|
|
||||
private String queueName; |
|
||||
|
|
||||
@Override |
|
||||
public TbCheckpointNodeConfiguration defaultConfiguration() { |
|
||||
return new TbCheckpointNodeConfiguration(); |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,52 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.rule.engine; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.fasterxml.jackson.databind.node.ObjectNode; |
||||
|
import org.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.MethodSource; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.rule.engine.api.TbNode; |
||||
|
import org.thingsboard.rule.engine.api.TbNodeException; |
||||
|
import org.thingsboard.server.common.data.util.TbPair; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.anyInt; |
||||
|
import static org.mockito.BDDMockito.willCallRealMethod; |
||||
|
|
||||
|
public abstract class AbstractRuleNodeUpgradeTest { |
||||
|
|
||||
|
protected abstract TbNode getTestNode(); |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@MethodSource |
||||
|
public void givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig(int givenVersion, String givenConfigStr, boolean hasChanges, String expectedConfigStr) throws TbNodeException { |
||||
|
// GIVEN
|
||||
|
willCallRealMethod().given(getTestNode()).upgrade(anyInt(), any()); |
||||
|
JsonNode givenConfig = JacksonUtil.toJsonNode(givenConfigStr); |
||||
|
JsonNode expectedConfig = JacksonUtil.toJsonNode(expectedConfigStr); |
||||
|
|
||||
|
// WHEN
|
||||
|
TbPair<Boolean, JsonNode> upgradeResult = getTestNode().upgrade(givenVersion, givenConfig); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(upgradeResult.getFirst()).isEqualTo(hasChanges); |
||||
|
ObjectNode upgradedConfig = (ObjectNode) upgradeResult.getSecond(); |
||||
|
assertThat(upgradedConfig).isEqualTo(expectedConfig); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,53 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.rule.engine.debug; |
||||
|
|
||||
|
import org.junit.jupiter.params.provider.Arguments; |
||||
|
import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest; |
||||
|
import org.thingsboard.rule.engine.api.TbNode; |
||||
|
|
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.mockito.Mockito.spy; |
||||
|
|
||||
|
public class TbMsgGeneratorNodeTest extends AbstractRuleNodeUpgradeTest { |
||||
|
|
||||
|
// Rule nodes upgrade
|
||||
|
private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { |
||||
|
return Stream.of( |
||||
|
// default config for version 0
|
||||
|
Arguments.of(0, |
||||
|
"{\"msgCount\":0,\"periodInSeconds\":1,\"originatorId\":null,\"originatorType\":null, \"queueName\":null, \"scriptLang\":\"TBEL\",\"jsScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\",\"tbelScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\"}", |
||||
|
true, |
||||
|
"{\"msgCount\":0,\"periodInSeconds\":1,\"originatorId\":null,\"originatorType\":null, \"scriptLang\":\"TBEL\",\"jsScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\",\"tbelScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\"}"), |
||||
|
// default config for version 0 with queueName
|
||||
|
Arguments.of(0, |
||||
|
"{\"msgCount\":0,\"periodInSeconds\":1,\"originatorId\":null,\"originatorType\":null, \"queueName\":\"Main\", \"scriptLang\":\"TBEL\",\"jsScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\",\"tbelScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\"}", |
||||
|
true, |
||||
|
"{\"msgCount\":0,\"periodInSeconds\":1,\"originatorId\":null,\"originatorType\":null, \"scriptLang\":\"TBEL\",\"jsScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\",\"tbelScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\"}"), |
||||
|
// default config for version 1 with upgrade from version 0
|
||||
|
Arguments.of(0, |
||||
|
"{\"msgCount\":0,\"periodInSeconds\":1,\"originatorId\":null,\"originatorType\":null, \"scriptLang\":\"TBEL\",\"jsScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\",\"tbelScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\"}", |
||||
|
false, |
||||
|
"{\"msgCount\":0,\"periodInSeconds\":1,\"originatorId\":null,\"originatorType\":null, \"scriptLang\":\"TBEL\",\"jsScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\",\"tbelScript\":\"var msg = { temp: 42, humidity: 77 };\\nvar metadata = { data: 40 };\\nvar msgType = \\\"POST_TELEMETRY_REQUEST\\\";\\n\\nreturn { msg: msg, metadata: metadata, msgType: msgType };\"}") |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected TbNode getTestNode() { |
||||
|
return spy(TbMsgGeneratorNode.class); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,55 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.rule.engine.flow; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.jupiter.params.provider.Arguments; |
||||
|
import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest; |
||||
|
import org.thingsboard.rule.engine.api.TbNode; |
||||
|
|
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.mockito.Mockito.spy; |
||||
|
|
||||
|
@Slf4j |
||||
|
public class TbCheckpointNodeTest extends AbstractRuleNodeUpgradeTest { |
||||
|
|
||||
|
// Rule nodes upgrade
|
||||
|
private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { |
||||
|
return Stream.of( |
||||
|
// default config for version 0
|
||||
|
Arguments.of(0, |
||||
|
"{\"queueName\":null}", |
||||
|
true, |
||||
|
"{}"), |
||||
|
// default config for version 0 with queueName
|
||||
|
Arguments.of(0, |
||||
|
"{\"queueName\":\"Main\"}", |
||||
|
true, |
||||
|
"{}"), |
||||
|
// default config for version 1 with upgrade from version 0
|
||||
|
Arguments.of(0, |
||||
|
"{}", |
||||
|
false, |
||||
|
"{}") |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected TbNode getTestNode() { |
||||
|
return spy(TbCheckpointNode.class); |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue