tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #11121 from irynamatveieva/improvements/change-originator-node 

Change originator node: added tests
pull/11458/head
Viacheslav Klimov 2 years ago
committed by GitHub
parent
20894ee0eb 7da0855180
commit
d5219bdd0a
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
5 changed files with 296 additions and 104 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 24
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/OriginatorSource.java
  2. 49
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java
  3. 9
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeConfiguration.java
  4. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesByNameAndTypeLoader.java
  5. 316
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java

24
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/OriginatorSource.java
View File

@ -0,0 +1,24 @@
/**
* 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.transform;
public enum OriginatorSource {
CUSTOMER,
TENANT,
RELATED,
ALARM_ORIGINATOR,
ENTITY
}

49
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java
View File

@ -15,7 +15,6 @@
*/ */
package org.thingsboard.rule.engine.transform; package org.thingsboard.rule.engine.transform;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -34,10 +33,12 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsg;
import java.util.HashSet;
import java.util.List; import java.util.List;
import java.util.NoSuchElementException; import java.util.NoSuchElementException;
import static org.thingsboard.rule.engine.transform.OriginatorSource.ENTITY;
import static org.thingsboard.rule.engine.transform.OriginatorSource.RELATED;
@Slf4j @Slf4j
@RuleNode( @RuleNode(
type = ComponentType.TRANSFORMATION, type = ComponentType.TRANSFORMATION,
@ -59,12 +60,6 @@ import java.util.NoSuchElementException;
) )
public class TbChangeOriginatorNode extends TbAbstractTransformNode<TbChangeOriginatorNodeConfiguration> { public class TbChangeOriginatorNode extends TbAbstractTransformNode<TbChangeOriginatorNodeConfiguration> {
private static final String CUSTOMER_SOURCE = "CUSTOMER";
private static final String TENANT_SOURCE = "TENANT";
private static final String RELATED_SOURCE = "RELATED";
private static final String ALARM_ORIGINATOR_SOURCE = "ALARM_ORIGINATOR";
private static final String ENTITY_SOURCE = "ENTITY";
@Override @Override
protected TbChangeOriginatorNodeConfiguration loadNodeConfiguration(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { protected TbChangeOriginatorNodeConfiguration loadNodeConfiguration(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
var config = TbNodeUtils.convert(configuration, TbChangeOriginatorNodeConfiguration.class); var config = TbNodeUtils.convert(configuration, TbChangeOriginatorNodeConfiguration.class);
@ -85,15 +80,15 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode<TbChangeOrig
private ListenableFuture<? extends EntityId> getNewOriginator(TbContext ctx, TbMsg msg) { private ListenableFuture<? extends EntityId> getNewOriginator(TbContext ctx, TbMsg msg) {
switch (config.getOriginatorSource()) { switch (config.getOriginatorSource()) {
case CUSTOMER_SOURCE: case CUSTOMER:
return EntitiesCustomerIdAsyncLoader.findEntityIdAsync(ctx, msg.getOriginator()); return EntitiesCustomerIdAsyncLoader.findEntityIdAsync(ctx, msg.getOriginator());
case TENANT_SOURCE: case TENANT:
return Futures.immediateFuture(ctx.getTenantId()); return Futures.immediateFuture(ctx.getTenantId());
case RELATED_SOURCE: case RELATED:
return EntitiesRelatedEntityIdAsyncLoader.findEntityAsync(ctx, msg.getOriginator(), config.getRelationsQuery()); return EntitiesRelatedEntityIdAsyncLoader.findEntityAsync(ctx, msg.getOriginator(), config.getRelationsQuery());
case ALARM_ORIGINATOR_SOURCE: case ALARM_ORIGINATOR:
return EntitiesAlarmOriginatorIdAsyncLoader.findEntityIdAsync(ctx, msg.getOriginator()); return EntitiesAlarmOriginatorIdAsyncLoader.findEntityIdAsync(ctx, msg.getOriginator());
case ENTITY_SOURCE: case ENTITY:
EntityType entityType = EntityType.valueOf(config.getEntityType()); EntityType entityType = EntityType.valueOf(config.getEntityType());
String entityName = TbNodeUtils.processPattern(config.getEntityNamePattern(), msg); String entityName = TbNodeUtils.processPattern(config.getEntityNamePattern(), msg);
try { try {
@ -108,28 +103,22 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode<TbChangeOrig
} }
private void validateConfig(TbChangeOriginatorNodeConfiguration conf) { private void validateConfig(TbChangeOriginatorNodeConfiguration conf) {
HashSet<String> knownSources = Sets.newHashSet(CUSTOMER_SOURCE, TENANT_SOURCE, RELATED_SOURCE, ALARM_ORIGINATOR_SOURCE, ENTITY_SOURCE); if (conf.getOriginatorSource() == null) {
if (!knownSources.contains(conf.getOriginatorSource())) { log.debug("Originator source should be specified.");
log.error("Unsupported source [{}] for TbChangeOriginatorNode", conf.getOriginatorSource()); throw new IllegalArgumentException("Originator source should be specified.");
throw new IllegalArgumentException("Unsupported source TbChangeOriginatorNode" + conf.getOriginatorSource());
} }
if (conf.getOriginatorSource().equals(RELATED) && conf.getRelationsQuery() == null) {
if (conf.getOriginatorSource().equals(RELATED_SOURCE)) { log.debug("Relations query should be specified if 'Related entity' source is selected.");
if (conf.getRelationsQuery() == null) { throw new IllegalArgumentException("Relations query should be specified if 'Related entity' source is selected.");
log.error("Related source for TbChangeOriginatorNode should have relations query. Actual [{}]",
conf.getRelationsQuery());
throw new IllegalArgumentException("Wrong config for RElated Source in TbChangeOriginatorNode" + conf.getOriginatorSource());
}
} }
if (conf.getOriginatorSource().equals(ENTITY)) {
if (conf.getOriginatorSource().equals(ENTITY_SOURCE)) {
if (conf.getEntityType() == null) { if (conf.getEntityType() == null) {
log.error("Entity type not specified for [{}]", ENTITY_SOURCE); log.debug("Entity type should be specified if '{}' source is selected.", ENTITY);
throw new IllegalArgumentException("Wrong config for [{}] in TbChangeOriginatorNode!" + ENTITY_SOURCE); throw new IllegalArgumentException("Entity type should be specified if 'Entity by name pattern' source is selected.");
} }
if (StringUtils.isEmpty(conf.getEntityNamePattern())) { if (StringUtils.isEmpty(conf.getEntityNamePattern())) {
log.error("EntityNamePattern not specified for type [{}]", conf.getEntityType()); log.debug("Name pattern should be specified if '{}' source is selected.", ENTITY);
throw new IllegalArgumentException("Wrong config for [{}] in TbChangeOriginatorNode!" + ENTITY_SOURCE); throw new IllegalArgumentException("Name pattern should be specified if 'Entity by name pattern' source is selected.");
} }
EntitiesByNameAndTypeLoader.checkEntityType(EntityType.valueOf(conf.getEntityType())); EntitiesByNameAndTypeLoader.checkEntityType(EntityType.valueOf(conf.getEntityType()));
} }

9
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeConfiguration.java
View File

@ -24,13 +24,12 @@ import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter;
import java.util.Collections; import java.util.Collections;
import static org.thingsboard.rule.engine.transform.OriginatorSource.CUSTOMER;
@Data @Data
public class TbChangeOriginatorNodeConfiguration implements NodeConfiguration<TbChangeOriginatorNodeConfiguration> { public class TbChangeOriginatorNodeConfiguration implements NodeConfiguration<TbChangeOriginatorNodeConfiguration> {
private static final String CUSTOMER_SOURCE = "CUSTOMER"; private OriginatorSource originatorSource;
private String originatorSource;
private RelationsQuery relationsQuery; private RelationsQuery relationsQuery;
private String entityType; private String entityType;
private String entityNamePattern; private String entityNamePattern;
@ -38,7 +37,7 @@ public class TbChangeOriginatorNodeConfiguration implements NodeConfiguration<Tb
@Override @Override
public TbChangeOriginatorNodeConfiguration defaultConfiguration() { public TbChangeOriginatorNodeConfiguration defaultConfiguration() {
TbChangeOriginatorNodeConfiguration configuration = new TbChangeOriginatorNodeConfiguration(); TbChangeOriginatorNodeConfiguration configuration = new TbChangeOriginatorNodeConfiguration();
configuration.setOriginatorSource(CUSTOMER_SOURCE); configuration.setOriginatorSource(CUSTOMER);
RelationsQuery relationsQuery = new RelationsQuery(); RelationsQuery relationsQuery = new RelationsQuery();
relationsQuery.setDirection(EntitySearchDirection.FROM); relationsQuery.setDirection(EntitySearchDirection.FROM);

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesByNameAndTypeLoader.java
View File

@ -53,7 +53,7 @@ public class EntitiesByNameAndTypeLoader {
throw new IllegalStateException("Unexpected entity type " + entityType.name()); throw new IllegalStateException("Unexpected entity type " + entityType.name());
} }
if (targetEntity == null) { if (targetEntity == null) {
throw new IllegalStateException("Failed to found " + entityType.name() + " entity by name: '" + entityName + "'!"); throw new IllegalStateException("Failed to find " + entityType.getNormalName().toLowerCase() + " with name '" + entityName + "'!");
} }
return targetEntity.getId(); return targetEntity.getId();
} }

316
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java
View File

@ -15,139 +15,319 @@
*/ */
package org.thingsboard.rule.engine.transform; package org.thingsboard.rule.engine.transform;
import com.datastax.oss.driver.api.core.uuid.Uuids;
import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.Futures;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.mockito.ArgumentCaptor; import org.mockito.ArgumentCaptor;
import org.mockito.Mock; import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoExtension;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.common.util.ListeningExecutor;
import org.thingsboard.rule.engine.TestDbCallbackExecutor; import org.thingsboard.rule.engine.TestDbCallbackExecutor;
import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.rule.engine.data.RelationsQuery;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.id.AlarmId;
import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntityRelationsQuery;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter;
import org.thingsboard.server.common.data.relation.RelationsSearchParameters;
import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgDataType;
import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.relation.RelationService;
import java.util.Collections;
import java.util.Map;
import java.util.NoSuchElementException; import java.util.NoSuchElementException;
import java.util.UUID;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same; import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.verify; import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.when; import static org.thingsboard.rule.engine.transform.OriginatorSource.ALARM_ORIGINATOR;
import static org.thingsboard.rule.engine.transform.OriginatorSource.CUSTOMER;
import static org.thingsboard.rule.engine.transform.OriginatorSource.ENTITY;
import static org.thingsboard.rule.engine.transform.OriginatorSource.RELATED;
import static org.thingsboard.rule.engine.transform.OriginatorSource.TENANT;
@ExtendWith(MockitoExtension.class) @ExtendWith(MockitoExtension.class)
public class TbChangeOriginatorNodeTest { public class TbChangeOriginatorNodeTest {
private static final String CUSTOMER_SOURCE = "CUSTOMER"; private final TenantId TENANT_ID = TenantId.fromUUID(UUID.fromString("79830b6d-4f93-49bd-9b5b-d31ce51da77b"));
private final CustomerId CUSTOMER_ID = new CustomerId(UUID.fromString("c6b2c94b-5517-4f20-bf8e-ae9407eb8a7a"));
private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("990605a4-db46-4ed4-942f-e18200453571"));
private final AssetId ASSET_ID = new AssetId(UUID.fromString("55de3f10-1b55-4950-b711-ed132896b260"));
private final ListeningExecutor dbExecutor = new TestDbCallbackExecutor();
private TbChangeOriginatorNode node; private TbChangeOriginatorNode node;
private TbChangeOriginatorNodeConfiguration config;
@Mock @Mock
private TbContext ctx; private TbContext ctxMock;
@Mock @Mock
private AssetService assetService; private AssetService assetServiceMock;
@Mock
private ListeningExecutor dbExecutor; private DeviceService deviceServiceMock;
@Mock
private RelationService relationServiceMock;
@Mock
private RuleEngineAlarmService alarmServiceMock;
@BeforeEach @BeforeEach
public void before() throws TbNodeException { public void before() throws TbNodeException {
dbExecutor = new TestDbCallbackExecutor(); node = new TbChangeOriginatorNode();
init(); config = new TbChangeOriginatorNodeConfiguration().defaultConfiguration();
} }
@Test @Test
public void originatorCanBeChangedToCustomerId() { public void verifyDefaultConfig() {
AssetId assetId = new AssetId(Uuids.timeBased()); var config = new TbChangeOriginatorNodeConfiguration().defaultConfiguration();
CustomerId customerId = new CustomerId(Uuids.timeBased()); assertThat(config.getOriginatorSource()).isEqualTo(CUSTOMER);
Asset asset = new Asset(); RelationsQuery relationsQuery = new RelationsQuery();
asset.setCustomerId(customerId); relationsQuery.setDirection(EntitySearchDirection.FROM);
relationsQuery.setMaxLevel(1);
RuleChainId ruleChainId = new RuleChainId(Uuids.timeBased()); RelationEntityTypeFilter relationEntityTypeFilter = new RelationEntityTypeFilter(EntityRelation.CONTAINS_TYPE, Collections.emptyList());
RuleNodeId ruleNodeId = new RuleNodeId(Uuids.timeBased()); relationsQuery.setFilters(Collections.singletonList(relationEntityTypeFilter));
assertThat(config.getRelationsQuery()).isEqualTo(relationsQuery);
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, assetId, TbMsgMetaData.EMPTY, TbMsgDataType.JSON, TbMsg.EMPTY_JSON_OBJECT, ruleChainId, ruleNodeId); assertThat(config.getEntityType()).isNull();
assertThat(config.getEntityNamePattern()).isNull();
}
when(ctx.getAssetService()).thenReturn(assetService); @Test
when(assetService.findAssetByIdAsync(any(),eq( assetId))).thenReturn(Futures.immediateFuture(asset)); public void givenRelatedSourceIsNull_whenInit_thenThrowsException() {
config.setOriginatorSource(null);
node.onMsg(ctx, msg); assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Originator source should be specified.");
}
ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class); @Test
ArgumentCaptor<EntityId> originatorCaptor = ArgumentCaptor.forClass(EntityId.class); public void givenRelatedSourceAndRelatedQueryIsNull_whenInit_thenThrowsException() {
verify(ctx).transformMsgOriginator(msgCaptor.capture(), originatorCaptor.capture()); config.setOriginatorSource(RELATED);
config.setRelationsQuery(null);
assertEquals(customerId, originatorCaptor.getValue()); assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Relations query should be specified if 'Related entity' source is selected.");
} }
@Test @Test
public void newChainCanBeStarted() { public void givenEntitySourceAndEntityTypeIsNull_whenInit_thenThrowsException() {
AssetId assetId = new AssetId(Uuids.timeBased()); config.setOriginatorSource(ENTITY);
CustomerId customerId = new CustomerId(Uuids.timeBased()); config.setEntityType(null);
Asset asset = new Asset();
asset.setCustomerId(customerId);
RuleChainId ruleChainId = new RuleChainId(Uuids.timeBased()); assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))))
RuleNodeId ruleNodeId = new RuleNodeId(Uuids.timeBased()); .isInstanceOf(IllegalArgumentException.class)
.hasMessage("Entity type should be specified if 'Entity by name pattern' source is selected.");
}
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, assetId, TbMsgMetaData.EMPTY, TbMsgDataType.JSON,TbMsg.EMPTY_JSON_OBJECT, ruleChainId, ruleNodeId); @ParameterizedTest
@NullAndEmptySource
public void givenEntitySourceAndEntityNamePatternIsEmpty_whenInit_thenThrowsException(String entityName) {
config.setOriginatorSource(ENTITY);
config.setEntityType(EntityType.DEVICE.name());
config.setEntityNamePattern(entityName);
when(ctx.getAssetService()).thenReturn(assetService); assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))))
when(assetService.findAssetByIdAsync(any(), eq(assetId))).thenReturn(Futures.immediateFuture(asset)); .isInstanceOf(IllegalArgumentException.class)
.hasMessage("Name pattern should be specified if 'Entity by name pattern' source is selected.");
}
node.onMsg(ctx, msg); @Test
ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class); public void givenEntitySourceAndUnexpectedEntityType_whenInit_thenThrowsException() {
ArgumentCaptor<EntityId> originatorCaptor = ArgumentCaptor.forClass(EntityId.class); config.setOriginatorSource(ENTITY);
verify(ctx).transformMsgOriginator(msgCaptor.capture(), originatorCaptor.capture()); config.setEntityType(EntityType.TENANT.name());
config.setEntityNamePattern("tenant-A");
assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))))
.isInstanceOf(IllegalStateException.class)
.hasMessage("Unexpected entity type TENANT");
}
assertEquals(customerId, originatorCaptor.getValue()); @Test
public void givenOriginatorSourceIsCustomer_whenOnMsg_thenTellSuccess() throws TbNodeException {
Device device = new Device(DEVICE_ID);
device.setCustomerId(CUSTOMER_ID);
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
TbMsg expectedMsg = TbMsg.transformMsgOriginator(msg, CUSTOMER_ID);
given(ctxMock.getDbCallbackExecutor()).willReturn(dbExecutor);
given(ctxMock.getDeviceService()).willReturn(deviceServiceMock);
given(ctxMock.getTenantId()).willReturn(TENANT_ID);
given(deviceServiceMock.findDeviceById(any(TenantId.class), any(DeviceId.class))).willReturn(device);
given(ctxMock.transformMsgOriginator(any(TbMsg.class), any(EntityId.class))).willReturn(expectedMsg);
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
node.onMsg(ctxMock, msg);
then(deviceServiceMock).should().findDeviceById(TENANT_ID, DEVICE_ID);
then(ctxMock).should().transformMsgOriginator(msg, CUSTOMER_ID);
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
then(ctxMock).should().tellSuccess(actualMsg.capture());
assertThat(actualMsg.getValue()).usingRecursiveComparison().ignoringFields("ctx").isEqualTo(expectedMsg);
} }
@Test @Test
public void exceptionThrownIfCannotFindNewOriginator() { public void givenOriginatorSourceIsTenant_whenOnMsg_thenTellSuccess() throws TbNodeException {
AssetId assetId = new AssetId(Uuids.timeBased()); config.setOriginatorSource(TENANT);
CustomerId customerId = new CustomerId(Uuids.timeBased());
Asset asset = new Asset();
asset.setCustomerId(customerId);
RuleChainId ruleChainId = new RuleChainId(Uuids.timeBased()); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, ASSET_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
RuleNodeId ruleNodeId = new RuleNodeId(Uuids.timeBased()); TbMsg expectedMsg = TbMsg.transformMsgOriginator(msg, TENANT_ID);
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, assetId, TbMsgMetaData.EMPTY, TbMsgDataType.JSON,TbMsg.EMPTY_JSON_OBJECT, ruleChainId, ruleNodeId); given(ctxMock.getDbCallbackExecutor()).willReturn(dbExecutor);
given(ctxMock.getTenantId()).willReturn(TENANT_ID);
given(ctxMock.transformMsgOriginator(any(TbMsg.class), any(EntityId.class))).willReturn(expectedMsg);
when(ctx.getAssetService()).thenReturn(assetService); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
when(assetService.findAssetByIdAsync(any(), eq(assetId))).thenReturn(Futures.immediateFuture(null)); node.onMsg(ctxMock, msg);
ArgumentCaptor<NoSuchElementException> exceptionCaptor = ArgumentCaptor.forClass(NoSuchElementException.class); then(ctxMock).should().transformMsgOriginator(msg, TENANT_ID);
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
then(ctxMock).should().tellSuccess(actualMsg.capture());
assertThat(actualMsg.getValue()).usingRecursiveComparison().ignoringFields("ctx").isEqualTo(expectedMsg);
}
node.onMsg(ctx, msg); @Test
verify(ctx).tellFailure(same(msg), exceptionCaptor.capture()); public void givenOriginatorSourceIsRelatedAndNewOriginatorIsNull_whenOnMsg_thenTellFailure() throws TbNodeException {
config.setOriginatorSource(RELATED);
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, ASSET_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
given(ctxMock.getDbCallbackExecutor()).willReturn(dbExecutor);
given(ctxMock.getRelationService()).willReturn(relationServiceMock);
given(ctxMock.getTenantId()).willReturn(TENANT_ID);
given(relationServiceMock.findByQuery(any(TenantId.class), any(EntityRelationsQuery.class))).willReturn(Futures.immediateFuture(Collections.emptyList()));
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
node.onMsg(ctxMock, msg);
var query = new EntityRelationsQuery();
var relationsQuery = config.getRelationsQuery();
var parameters = new RelationsSearchParameters(
ASSET_ID,
relationsQuery.getDirection(),
relationsQuery.getMaxLevel(),
relationsQuery.isFetchLastLevelOnly()
);
query.setParameters(parameters);
query.setFilters(relationsQuery.getFilters());
then(relationServiceMock).should().findByQuery(TENANT_ID, query);
ArgumentCaptor<Throwable> throwable = ArgumentCaptor.forClass(Throwable.class);
then(ctxMock).should().tellFailure(eq(msg), throwable.capture());
assertThat(throwable.getValue()).isInstanceOf(NoSuchElementException.class).hasMessage("Failed to find new originator!");
}
assertEquals("Failed to find new originator!", exceptionCaptor.getValue().getMessage()); @Test
public void givenOriginatorSourceIsAlarmOriginator_whenOnMsg_thenTellSuccess() throws TbNodeException {
config.setOriginatorSource(ALARM_ORIGINATOR);
AlarmId alarmId = new AlarmId(UUID.fromString("6b43f694-cb5f-4199-9023-e9e40eeb82dd"));
Alarm alarm = new Alarm(alarmId);
alarm.setOriginator(DEVICE_ID);
TbMsg msg = TbMsg.newMsg(TbMsgType.ALARM, alarmId, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
TbMsg expectedMsg = TbMsg.transformMsgOriginator(msg, DEVICE_ID);
given(ctxMock.getDbCallbackExecutor()).willReturn(dbExecutor);
given(ctxMock.getAlarmService()).willReturn(alarmServiceMock);
given(ctxMock.getTenantId()).willReturn(TENANT_ID);
given(alarmServiceMock.findAlarmByIdAsync(any(TenantId.class), any(AlarmId.class))).willReturn(Futures.immediateFuture(alarm));
given(ctxMock.transformMsgOriginator(any(TbMsg.class), any(EntityId.class))).willReturn(expectedMsg);
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
node.onMsg(ctxMock, msg);
then(alarmServiceMock).should().findAlarmByIdAsync(TENANT_ID, alarmId);
then(ctxMock).should().transformMsgOriginator(msg, DEVICE_ID);
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
then(ctxMock).should().tellSuccess(actualMsg.capture());
assertThat(actualMsg.getValue()).usingRecursiveComparison().ignoringFields("ctx").isEqualTo(expectedMsg);
} }
public void init() throws TbNodeException { @ParameterizedTest
TbChangeOriginatorNodeConfiguration config = new TbChangeOriginatorNodeConfiguration(); @MethodSource
config.setOriginatorSource(CUSTOMER_SOURCE); public void givenOriginatorSourceIsEntity_whenOnMsg_thenTellSuccess(String entityNamePattern, TbMsgMetaData metaData, String data) throws TbNodeException {
TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); config.setOriginatorSource(ENTITY);
config.setEntityType(EntityType.ASSET.name());
config.setEntityNamePattern(entityNamePattern);
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, data);
TbMsg expectedMsg = TbMsg.transformMsgOriginator(msg, ASSET_ID);
given(ctxMock.getDbCallbackExecutor()).willReturn(dbExecutor);
given(ctxMock.getAssetService()).willReturn(assetServiceMock);
given(ctxMock.getTenantId()).willReturn(TENANT_ID);
given(assetServiceMock.findAssetByTenantIdAndName(any(TenantId.class), any(String.class))).willReturn(new Asset(ASSET_ID));
given(ctxMock.transformMsgOriginator(any(TbMsg.class), any(EntityId.class))).willReturn(expectedMsg);
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
node.onMsg(ctxMock, msg);
String expectedEntityName = TbNodeUtils.processPattern(entityNamePattern, msg);
then(assetServiceMock).should().findAssetByTenantIdAndName(TENANT_ID, expectedEntityName);
then(ctxMock).should().transformMsgOriginator(msg, ASSET_ID);
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
then(ctxMock).should().tellSuccess(actualMsg.capture());
assertThat(actualMsg.getValue()).usingRecursiveComparison().ignoringFields("ctx").isEqualTo(expectedMsg);
}
when(ctx.getDbCallbackExecutor()).thenReturn(dbExecutor); private static Stream<Arguments> givenOriginatorSourceIsEntity_whenOnMsg_thenTellSuccess() {
return Stream.of(
Arguments.of("test-asset", TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT),
Arguments.of("${md-name-pattern}", new TbMsgMetaData(Map.of("md-name-pattern", "md-test-asset")), TbMsg.EMPTY_JSON_OBJECT),
Arguments.of("${msg-name-pattern}", TbMsgMetaData.EMPTY, "{\"msg-name-pattern\":\"msg-test-asset\"}")
);
}
node = new TbChangeOriginatorNode(); @Test
node.init(null, nodeConfiguration); public void givenOriginatorSourceIsEntityAndEntityCouldNotFound_whenOnMsg_thenTellFailure() throws TbNodeException {
config.setOriginatorSource(ENTITY);
config.setEntityType(EntityType.ASSET.name());
config.setEntityNamePattern("${md-name-pattern}");
TbMsgMetaData metaData = new TbMsgMetaData();
metaData.putValue("md-name-pattern", "test-asset");
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, TbMsg.EMPTY_JSON_OBJECT);
given(ctxMock.getDbCallbackExecutor()).willReturn(dbExecutor);
given(ctxMock.getAssetService()).willReturn(assetServiceMock);
given(ctxMock.getTenantId()).willReturn(TENANT_ID);
given(assetServiceMock.findAssetByTenantIdAndName(any(TenantId.class), any(String.class))).willReturn(null);
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
node.onMsg(ctxMock, msg);
ArgumentCaptor<Throwable> throwable = ArgumentCaptor.forClass(Throwable.class);
then(ctxMock).should().tellFailure(eq(msg), throwable.capture());
assertThat(throwable.getValue()).isInstanceOf(IllegalStateException.class).hasMessage("Failed to find asset with name 'test-asset'!");
} }
} }

Write Preview
Loading…
Cancel
Save