diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbAckNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbAckNodeTest.java index 909f866493..49e92f2d74 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbAckNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbAckNodeTest.java @@ -43,7 +43,6 @@ public class TbAckNodeTest { private EmptyNodeConfiguration config; private TbNodeConfiguration nodeConfiguration; - @Mock private TbContext ctxMock; diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeTest.java index 9b6e86b9a6..e596c71d1e 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeTest.java @@ -22,6 +22,7 @@ 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.ValueSource; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.thingsboard.common.util.JacksonUtil; @@ -49,12 +50,13 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.spy; import static org.mockito.BDDMockito.then; -import static org.mockito.BDDMockito.willAnswer; @Slf4j @ExtendWith(MockitoExtension.class) public class TbCheckpointNodeTest extends AbstractRuleNodeUpgradeTest { + private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("37840655-b7dc-4f49-8da3-9429159e0970")); + private TbCheckpointNode node; private EmptyNodeConfiguration config; private TbNodeConfiguration nodeConfiguration; @@ -80,24 +82,37 @@ public class TbCheckpointNodeTest extends AbstractRuleNodeUpgradeTest { } @ParameterizedTest - @ValueSource(strings = {DataConstants.MAIN_QUEUE_NAME, DataConstants.HP_QUEUE_NAME, DataConstants.HP_QUEUE_NAME, "Custom queue"}) + @ValueSource(strings = {DataConstants.MAIN_QUEUE_NAME, DataConstants.HP_QUEUE_NAME, DataConstants.SQ_QUEUE_NAME, "Custom queue"}) public void givenQueueName_whenOnMsg_thenTransfersMsgToDefinedQueue(String queueName) throws TbNodeException { given(ctxMock.getQueueName()).willReturn(queueName); - willAnswer(invocationOnMock -> { - Runnable onSuccess = invocationOnMock.getArgument(3); - onSuccess.run(); - return null; - }).given(ctxMock).enqueueForTellNext(any(TbMsg.class), any(String.class), any(String.class), any(Runnable.class), any(Consumer.class)); node.init(ctxMock, nodeConfiguration); - DeviceId deviceId = new DeviceId(UUID.fromString("2cd04871-7f07-41d1-b850-95dd444a6506")); - TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, deviceId, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); - then(ctxMock).should().enqueueForTellNext(eq(msg), eq(queueName), eq(TbNodeConnectionType.SUCCESS), any(), any()); + ArgumentCaptor onSuccess = ArgumentCaptor.forClass(Runnable.class); + then(ctxMock).should().enqueueForTellNext(eq(msg), eq(queueName), eq(TbNodeConnectionType.SUCCESS), onSuccess.capture(), any()); + onSuccess.getValue().run(); then(ctxMock).should().ack(msg); } + @Test + public void givenErrorDuringTransfer_whenOnMsg_thenTellFailure() throws TbNodeException { + given(ctxMock.getQueueName()).willReturn(DataConstants.HP_QUEUE_NAME); + + node.init(ctxMock, nodeConfiguration); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); + node.onMsg(ctxMock, msg); + + ArgumentCaptor> onFailure = ArgumentCaptor.forClass(Consumer.class); + then(ctxMock).should().enqueueForTellNext(eq(msg), eq(DataConstants.HP_QUEUE_NAME), eq(TbNodeConnectionType.SUCCESS), any(), onFailure.capture()); + String errorMsg = "Something went wrong."; + onFailure.getValue().accept(new RuntimeException(errorMsg)); + ArgumentCaptor throwable = ArgumentCaptor.forClass(Throwable.class); + then(ctxMock).should().tellFailure(eq(msg), throwable.capture()); + assertThat(throwable.getValue()).isInstanceOf(RuntimeException.class).hasMessage(errorMsg); + } + // Rule nodes upgrade private static Stream givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { return Stream.of(