diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sns/TbSnsNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sns/TbSnsNodeTest.java index 2b67187687..09787d3da8 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sns/TbSnsNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sns/TbSnsNodeTest.java @@ -37,6 +37,7 @@ import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.msg.TbMsgType; +import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -47,10 +48,12 @@ import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.BDDMockito.mock; +import static org.mockito.BDDMockito.never; +import static org.mockito.BDDMockito.verifyNoMoreInteractions; @ExtendWith(MockitoExtension.class) class TbSnsNodeTest { @@ -88,7 +91,8 @@ class TbSnsNodeTest { @ParameterizedTest @MethodSource - void givenTopicNamePattern_whenOnMsg_thenTellSuccess(String topicName, TbMsgMetaData metaData, String data) { + void givenForceAckIsTrueAndTopicNamePattern_whenOnMsg_thenEnqueueForTellNext(String topicName, TbMsgMetaData metaData, String data) { + ReflectionTestUtils.setField(node, "forceAck", true); config.setAccessKeyId("accessKeyId"); config.setSecretAccessKey("secretAccessKey"); config.setTopicArnPattern(topicName); @@ -104,19 +108,25 @@ class TbSnsNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, data); node.onMsg(ctxMock, msg); + then(ctxMock).should().ack(msg); PublishRequest publishRequest = new PublishRequest() .withTopicArn(TbNodeUtils.processPattern(topicName, msg)) .withMessage(data); then(snsClientMock).should().publish(publishRequest); - ArgumentCaptor msgArgumentCaptor = ArgumentCaptor.forClass(TbMsg.class); - then(ctxMock).should().tellSuccess(msgArgumentCaptor.capture()); - assertThat(msgArgumentCaptor.getValue().getMetaData().getData()) + ArgumentCaptor actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should().enqueueForTellNext(actualMsgCaptor.capture(), eq(TbNodeConnectionType.SUCCESS)); + TbMsg actualMsg = actualMsgCaptor.getValue(); + assertThat(actualMsg) + .usingRecursiveComparison() + .ignoringFields("metaData", "ctx") + .isEqualTo(msg); + assertThat(actualMsg.getMetaData().getData()) .hasFieldOrPropertyWithValue("messageId", messageId) .hasFieldOrPropertyWithValue("requestId", requestId); verifyNoMoreInteractions(ctxMock, snsClientMock, publishResultMock, responseMetadataMock); } - private static Stream givenTopicNamePattern_whenOnMsg_thenTellSuccess() { + private static Stream givenForceAckIsTrueAndTopicNamePattern_whenOnMsg_thenEnqueueForTellNext() { return Stream.of( Arguments.of("arn:aws:sns:us-east-1:123456789012:NewTopic", TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT), Arguments.of("arn:aws:sns:us-east-1:123456789012:$[msgTopicName]", TbMsgMetaData.EMPTY, "{\"msgTopicName\":\"msg-topic-name\"}"), @@ -125,7 +135,8 @@ class TbSnsNodeTest { } @Test - void givenErrorOccursDuringProcessingRequest_whenOnMsg_thenTellFailure() { + void givenForceAckIsFalseAndErrorOccursDuringProcessingRequest_whenOnMsg_thenTellFailure() { + ReflectionTestUtils.setField(node, "forceAck", false); ListeningExecutor listeningExecutor = mock(ListeningExecutor.class); given(ctxMock.getExternalCallExecutor()).willReturn(listeningExecutor); String errorMsg = "Something went wrong"; @@ -135,12 +146,31 @@ class TbSnsNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); - ArgumentCaptor msgArgumentCaptor = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should(never()).enqueueForTellNext(any(), any(String.class)); + ArgumentCaptor actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); - then(ctxMock).should().tellFailure(msgArgumentCaptor.capture(), throwableCaptor.capture()); - assertThat(msgArgumentCaptor.getValue().getMetaData().getData()) + then(ctxMock).should().tellFailure(actualMsgCaptor.capture(), throwableCaptor.capture()); + TbMsg actualMsg = actualMsgCaptor.getValue(); + assertThat(actualMsg) + .usingRecursiveComparison() + .ignoringFields("metaData", "ctx") + .isEqualTo(msg); + assertThat(actualMsg.getMetaData().getData()) .hasFieldOrPropertyWithValue("error", RuntimeException.class + ": " + errorMsg); assertThat(throwableCaptor.getValue()).isInstanceOf(RuntimeException.class).hasMessage(errorMsg); verifyNoMoreInteractions(ctxMock, snsClientMock); } + + @Test + void givenSnsClientIsNotNull_whenDestroy_thenShutdown() { + node.destroy(); + then(snsClientMock).should().shutdown(); + } + + @Test + void givenSnsClientIsNull_whenDestroy_thenVerifyNoInteractions() { + ReflectionTestUtils.setField(node, "snsClient", null); + node.destroy(); + then(snsClientMock).shouldHaveNoInteractions(); + } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNodeTest.java index b2b80b19d8..48f42edb97 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNodeTest.java @@ -39,6 +39,7 @@ import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.rule.engine.aws.sqs.TbSqsNodeConfiguration.QueueType; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.msg.TbMsgType; +import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -51,10 +52,12 @@ import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.BDDMockito.mock; +import static org.mockito.BDDMockito.never; +import static org.mockito.BDDMockito.verifyNoMoreInteractions; @ExtendWith(MockitoExtension.class) class TbSqsNodeTest { @@ -172,7 +175,8 @@ class TbSqsNodeTest { } @Test - void givenMsgResultContainsBodyAndAttributesAndNumber_whenOnMsg_thenTellSuccess() { + void givenForceAckIsTrueAndMsgResultContainsBodyAndAttributesAndNumber_whenOnMsg_thenEnqueueForTellNext() { + ReflectionTestUtils.setField(node, "forceAck", true); String messageBodyMd5 = "msgBodyMd5-55fb8ba2-2b71-4673-a82a-969756764761"; String messageAttributesMd5 = "msgAttrMd5-e3ba3eef-52ae-436a-bec1-0c2c2252d1f1"; String sequenceNumber = "seqNum-bb5ddce0-cf4e-4295-b015-524bdb6a332f"; @@ -185,15 +189,20 @@ class TbSqsNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); + then(ctxMock).should().ack(msg); SendMessageRequest sendMsgRequest = new SendMessageRequest() .withQueueUrl(TbNodeUtils.processPattern(config.getQueueUrlPattern(), msg)) .withMessageBody(msg.getData()) .withDelaySeconds(config.getDelaySeconds()); then(sqsClientMock).should().sendMessage(sendMsgRequest); - ArgumentCaptor msgArgumentCaptor = ArgumentCaptor.forClass(TbMsg.class); - then(ctxMock).should().tellSuccess(msgArgumentCaptor.capture()); - Map metadata = msgArgumentCaptor.getValue().getMetaData().getData(); - assertThat(metadata) + ArgumentCaptor actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should().enqueueForTellNext(actualMsgCaptor.capture(), eq(TbNodeConnectionType.SUCCESS)); + TbMsg actualMsg = actualMsgCaptor.getValue(); + assertThat(actualMsg) + .usingRecursiveComparison() + .ignoringFields("metaData", "ctx") + .isEqualTo(msg); + assertThat(actualMsg.getMetaData().getData()) .hasFieldOrPropertyWithValue("messageId", messageId) .hasFieldOrPropertyWithValue("requestId", requestId) .hasFieldOrPropertyWithValue("messageBodyMd5", messageBodyMd5) @@ -203,7 +212,8 @@ class TbSqsNodeTest { } @Test - void givenErrorOccursDuringProcessingRequest_whenOnMsg_thenTellFailure() { + void givenForceAckIsFalseAndErrorOccursDuringProcessingRequest_whenOnMsg_thenTellFailure() { + ReflectionTestUtils.setField(node, "forceAck", false); ListeningExecutor listeningExecutor = mock(ListeningExecutor.class); given(ctxMock.getExternalCallExecutor()).willReturn(listeningExecutor); String errorMsg = "Something went wrong"; @@ -214,15 +224,34 @@ class TbSqsNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); - ArgumentCaptor msgArgumentCaptor = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should(never()).enqueueForTellNext(any(), any(String.class)); + ArgumentCaptor actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); - then(ctxMock).should().tellFailure(msgArgumentCaptor.capture(), throwableCaptor.capture()); - assertThat(msgArgumentCaptor.getValue().getMetaData().getData()) + then(ctxMock).should().tellFailure(actualMsgCaptor.capture(), throwableCaptor.capture()); + TbMsg actualMsg = actualMsgCaptor.getValue(); + assertThat(actualMsg) + .usingRecursiveComparison() + .ignoringFields("metaData", "ctx") + .isEqualTo(msg); + assertThat(actualMsg.getMetaData().getData()) .hasFieldOrPropertyWithValue("error", RuntimeException.class + ": " + errorMsg); assertThat(throwableCaptor.getValue()).isInstanceOf(RuntimeException.class).hasMessage(errorMsg); verifyNoMoreInteractions(ctxMock, sqsClientMock); } + @Test + void givenSqsClientIsNotNull_whenDestroy_thenShutdown() { + node.destroy(); + then(sqsClientMock).should().shutdown(); + } + + @Test + void givenSqsClientIsNull_whenDestroy_thenVerifyNoInteractions() { + ReflectionTestUtils.setField(node, "sqsClient", null); + node.destroy(); + then(sqsClientMock).shouldHaveNoInteractions(); + } + private void mockSendingMsgRequest() { given(ctxMock.getExternalCallExecutor()).willReturn(executor); given(sqsClientMock.sendMessage(any(SendMessageRequest.class))).willReturn(sendMessageResultMock);