added tests for azure iot hub node

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java

@@ -115,7 +115,7 @@ public class TbMqttNode extends TbAbstractExternalNode {
         return "Tenant[" + ctx.getTenantId().getId() + "]RuleNode[" + ctx.getSelf().getId().getId() + "]";
     }
 
-    protected MqttClient initClient(TbContext ctx) throws Exception {
+    public MqttClient initClient(TbContext ctx) throws Exception {
         MqttClientConfig config = new MqttClientConfig(getSslContext());
         config.setOwnerId(getOwnerId(ctx));
         if (!StringUtils.isEmpty(this.mqttNodeConfiguration.getClientId())) {

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNode.java

@@ -32,8 +32,6 @@ import org.thingsboard.rule.engine.mqtt.TbMqttNodeConfiguration;
 import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 
-import javax.net.ssl.SSLException;
-
 @Slf4j
 @RuleNode(
         type = ComponentType.EXTERNAL,
@@ -74,4 +72,8 @@ public class TbAzureIotHubNode extends TbMqttNode {
             config.setPassword(AzureIotHubUtil.buildSasToken(mqttNodeConfiguration.getHost(), ((AzureIotHubSasCredentials) credentials).getSasKey()));
         }
     }
+
+    protected TbMqttNodeConfiguration getMqttNodeConfiguration() {
+        return this.mqttNodeConfiguration;
+    }
 }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeTest.java

@@ -30,7 +30,6 @@ import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
-import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.thingsboard.common.util.JacksonUtil;
@@ -73,38 +72,37 @@ import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.BDDMockito.willThrow;
 
 @ExtendWith(MockitoExtension.class)
-class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
+public class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
 
     private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("09115d92-d333-432a-868c-ccd6e89c9287"));
 
-    @Spy
+    protected TbMqttNode mqttNode;
-    private TbMqttNode node;
+    protected TbMqttNodeConfiguration mqttNodeConfig;
-    private TbMqttNodeConfiguration config;
 
     @Mock
-    private TbContext ctxMock;
+    protected TbContext ctxMock;
     @Mock
-    private MqttClient clientMock;
+    protected MqttClient mqttClientMock;
 
     @BeforeEach
-    public void setUp() throws Exception {
+    protected void setUp() {
-        node = spy(new TbMqttNode());
+        mqttNode = spy(new TbMqttNode());
-        config = new TbMqttNodeConfiguration().defaultConfiguration();
+        mqttNodeConfig = new TbMqttNodeConfiguration().defaultConfiguration();
     }
 
     @Test
     public void verifyDefaultConfig() {
-        assertThat(config.getTopicPattern()).isEqualTo("my-topic");
+        assertThat(mqttNodeConfig.getTopicPattern()).isEqualTo("my-topic");
-        assertThat(config.getHost()).isNull();
+        assertThat(mqttNodeConfig.getHost()).isNull();
-        assertThat(config.getPort()).isEqualTo(1883);
+        assertThat(mqttNodeConfig.getPort()).isEqualTo(1883);
-        assertThat(config.getConnectTimeoutSec()).isEqualTo(10);
+        assertThat(mqttNodeConfig.getConnectTimeoutSec()).isEqualTo(10);
-        assertThat(config.getClientId()).isNull();
+        assertThat(mqttNodeConfig.getClientId()).isNull();
-        assertThat(config.isAppendClientIdSuffix()).isFalse();
+        assertThat(mqttNodeConfig.isAppendClientIdSuffix()).isFalse();
-        assertThat(config.isRetainedMessage()).isFalse();
+        assertThat(mqttNodeConfig.isRetainedMessage()).isFalse();
-        assertThat(config.isCleanSession()).isTrue();
+        assertThat(mqttNodeConfig.isCleanSession()).isTrue();
-        assertThat(config.isSsl()).isFalse();
+        assertThat(mqttNodeConfig.isSsl()).isFalse();
-        assertThat(config.isParseToPlainText()).isFalse();
+        assertThat(mqttNodeConfig.isParseToPlainText()).isFalse();
-        assertThat(config.getCredentials()).isInstanceOf(AnonymousCredentials.class);
+        assertThat(mqttNodeConfig.getCredentials()).isInstanceOf(AnonymousCredentials.class);
     }
 
     @Test
@@ -115,7 +113,7 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
         given(ctxMock.getTenantId()).willReturn(TenantId.fromUUID(UUID.fromString(tenantIdStr)));
         given(ctxMock.getSelf()).willReturn(ruleNode);
 
-        String actualOwnerIdStr = node.getOwnerId(ctxMock);
+        String actualOwnerIdStr = mqttNode.getOwnerId(ctxMock);
         String expectedOwnerIdStr = "Tenant[" + tenantIdStr + "]RuleNode[" + ruleNodeIdStr + "]";
         assertThat(actualOwnerIdStr).isEqualTo(expectedOwnerIdStr);
     }
@@ -125,11 +123,11 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
         BasicCredentials credentials = new BasicCredentials();
         credentials.setUsername("test_username");
         credentials.setPassword("test_password");
-        config.setCredentials(credentials);
+        mqttNodeConfig.setCredentials(credentials);
-        ReflectionTestUtils.setField(node, "mqttNodeConfiguration", config);
+        ReflectionTestUtils.setField(mqttNode, "mqttNodeConfiguration", mqttNodeConfig);
-        MqttClientConfig mqttClientConfig = new MqttClientConfig(node.getSslContext());
+        MqttClientConfig mqttClientConfig = new MqttClientConfig(mqttNode.getSslContext());
 
-        node.prepareMqttClientConfig(mqttClientConfig);
+        mqttNode.prepareMqttClientConfig(mqttClientConfig);
 
         assertThat(mqttClientConfig)
                 .hasFieldOrPropertyWithValue("username", "test_username")
@@ -139,11 +137,11 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
     @ParameterizedTest
     @MethodSource
     public void verifyGetSslContextMethod(boolean ssl, ClientCredentials credentials, SslContext expectedSslContext) throws SSLException {
-        config.setSsl(ssl);
+        mqttNodeConfig.setSsl(ssl);
-        config.setCredentials(credentials);
+        mqttNodeConfig.setCredentials(credentials);
-        ReflectionTestUtils.setField(node, "mqttNodeConfiguration", config);
+        ReflectionTestUtils.setField(mqttNode, "mqttNodeConfiguration", mqttNodeConfig);
 
-        SslContext actualSslContext = node.getSslContext();
+        SslContext actualSslContext = mqttNode.getSslContext();
         assertThat(actualSslContext)
                 .usingRecursiveComparison()
                 .ignoringFields("ctx", "ctxLock", "sessionContext.context.ctx", "sessionContext.context.ctxLock")
@@ -160,10 +158,10 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
     @Test
     public void givenFailedToInitializeMqttClient_whenInit_thenThrowsException() throws Exception {
         String errorMsg = "Failed to connect to MQTT broker!";
-        willThrow(new RuntimeException(errorMsg)).given(node).initClient(any());
+        willThrow(new RuntimeException(errorMsg)).given(mqttNode).initClient(any());
 
-        var configuration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
+        var configuration = new TbNodeConfiguration(JacksonUtil.valueToTree(mqttNodeConfig));
-        assertThatThrownBy(() -> node.init(ctxMock, configuration))
+        assertThatThrownBy(() -> mqttNode.init(ctxMock, configuration))
                 .isInstanceOf(TbNodeException.class)
                 .hasMessage(RuntimeException.class.getName() + ": " + errorMsg);
     }
@@ -171,25 +169,25 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
     @ParameterizedTest
     @MethodSource
     public void givenTopicPatternAndIsRetainedMsgIsTrue_whenOnMsg_thenTellSuccess(String topicPattern, TbMsgMetaData metaData, String data) throws Exception {
-        config.setRetainedMessage(true);
+        mqttNodeConfig.setRetainedMessage(true);
-        config.setTopicPattern(topicPattern);
+        mqttNodeConfig.setTopicPattern(topicPattern);
 
-        willReturn(clientMock).given(node).initClient(any());
+        willReturn(mqttClientMock).given(mqttNode).initClient(any());
         Future<Void> future = mock(Future.class);
         given(future.isSuccess()).willReturn(true);
-        given(clientMock.publish(any(String.class), any(ByteBuf.class), any(MqttQoS.class), anyBoolean())).willReturn(future);
+        given(mqttClientMock.publish(any(String.class), any(ByteBuf.class), any(MqttQoS.class), anyBoolean())).willReturn(future);
         willAnswer(invocation-> {
             GenericFutureListener<Future<Void>> listener = invocation.getArgument(0);
             listener.operationComplete(future);
             return null;
         }).given(future).addListener(any());
 
-        node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
+        mqttNode.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(mqttNodeConfig)));
         TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, data);
-        node.onMsg(ctxMock, msg);
+        mqttNode.onMsg(ctxMock, msg);
 
-        String expectedTopic = TbNodeUtils.processPattern(config.getTopicPattern(), msg);
+        String expectedTopic = TbNodeUtils.processPattern(mqttNodeConfig.getTopicPattern(), msg);
-        then(clientMock).should().publish(expectedTopic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE, true);
+        then(mqttClientMock).should().publish(expectedTopic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE, true);
         then(ctxMock).should().tellSuccess(msg);
     }
 
@@ -203,11 +201,11 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
 
     @Test
     public void givenParseToPlainTextIsTrueAndMsgPublishingFailed_whenOnMsg_thenTellFailure() throws Exception {
-        config.setParseToPlainText(true);
+        mqttNodeConfig.setParseToPlainText(true);
 
-        willReturn(clientMock).given(node).initClient(any());
+        willReturn(mqttClientMock).given(mqttNode).initClient(any());
         Future<Void> future = mock(Future.class);
-        given(clientMock.publish(any(String.class), any(ByteBuf.class), any(MqttQoS.class), anyBoolean())).willReturn(future);
+        given(mqttClientMock.publish(any(String.class), any(ByteBuf.class), any(MqttQoS.class), anyBoolean())).willReturn(future);
         given(future.isSuccess()).willReturn(false);
         String errorMsg = "Message publishing was failed!";
         Throwable exception = new RuntimeException(errorMsg);
@@ -218,12 +216,12 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
             return null;
         }).given(future).addListener(any());
 
-        node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
+        mqttNode.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(mqttNodeConfig)));
         TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, "\"string\"");
-        node.onMsg(ctxMock, msg);
+        mqttNode.onMsg(ctxMock, msg);
 
         String expectedData = JacksonUtil.toPlainText(msg.getData());
-        then(clientMock).should().publish(config.getTopicPattern(), Unpooled.wrappedBuffer(expectedData.getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE, false);
+        then(mqttClientMock).should().publish(mqttNodeConfig.getTopicPattern(), Unpooled.wrappedBuffer(expectedData.getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE, false);
         TbMsgMetaData metaData = new TbMsgMetaData();
         metaData.putValue("error", RuntimeException.class + ": " + errorMsg);
         TbMsg expectedMsg = TbMsg.transformMsgMetadata(msg, metaData);
@@ -251,6 +249,6 @@ class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest {
 
     @Override
     protected TbNode getTestNode() {
-        return node;
+        return mqttNode;
     }
 }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNodeTest.java

@@ -0,0 +1,113 @@
+/**
+ * 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.mqtt.azure;
+
+import io.netty.handler.codec.mqtt.MqttVersion;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.common.util.AzureIotHubUtil;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.mqtt.MqttClientConfig;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.rule.engine.credentials.CertPemCredentials;
+import org.thingsboard.rule.engine.mqtt.TbMqttNodeConfiguration;
+import org.thingsboard.rule.engine.mqtt.TbMqttNodeTest;
+
+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.BDDMockito.spy;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.BDDMockito.willThrow;
+
+public class TbAzureIotHubNodeTest extends TbMqttNodeTest {
+
+    private TbAzureIotHubNode azureIotHubNode;
+    private TbAzureIotHubNodeConfiguration azureIotHubNodeConfig;
+
+    @BeforeEach
+    public void setUp() {
+        super.setUp();
+        azureIotHubNode = spy(new TbAzureIotHubNode());
+        azureIotHubNodeConfig = new TbAzureIotHubNodeConfiguration().defaultConfiguration();
+    }
+
+    @Test
+    public void verifyDefaultConfig() {
+        assertThat(azureIotHubNodeConfig.getTopicPattern()).isEqualTo("devices/<device_id>/messages/events/");
+        assertThat(azureIotHubNodeConfig.getHost()).isEqualTo("<iot-hub-name>.azure-devices.net");
+        assertThat(azureIotHubNodeConfig.getPort()).isEqualTo(8883);
+        assertThat(azureIotHubNodeConfig.getConnectTimeoutSec()).isEqualTo(10);
+        assertThat(azureIotHubNodeConfig.isCleanSession()).isTrue();
+        assertThat(azureIotHubNodeConfig.isSsl()).isTrue();
+        assertThat(azureIotHubNodeConfig.getCredentials()).isInstanceOf(AzureIotHubSasCredentials.class);
+    }
+
+    @Test
+    public void verifyPrepareMqttClientConfigMethodWithAzureIotHubSasCredentials() throws TbNodeException {
+        AzureIotHubSasCredentials credentials = new AzureIotHubSasCredentials();
+        credentials.setSasKey("testSasKey");
+        credentials.setCaCert("test-ca-cert.pem");
+        azureIotHubNodeConfig.setCredentials(credentials);
+        TbNodeConfiguration configuration = new TbNodeConfiguration(JacksonUtil.valueToTree(azureIotHubNodeConfig));
+        mqttNodeConfig = TbNodeUtils.convert(configuration, TbMqttNodeConfiguration.class);
+        ReflectionTestUtils.setField(azureIotHubNode, "mqttNodeConfiguration", mqttNodeConfig);
+        MqttClientConfig mqttClientConfig = new MqttClientConfig();
+
+        azureIotHubNode.prepareMqttClientConfig(mqttClientConfig);
+
+        assertThat(mqttClientConfig)
+                .hasFieldOrPropertyWithValue("protocolVersion", MqttVersion.MQTT_3_1_1)
+                .hasFieldOrPropertyWithValue("username", AzureIotHubUtil.buildUsername(mqttNodeConfig.getHost(), mqttClientConfig.getClientId()))
+                .hasFieldOrPropertyWithValue("password", AzureIotHubUtil.buildSasToken(mqttNodeConfig.getHost(), credentials.getSasKey()));
+    }
+
+    @Test
+    public void givenPemCredentialsAndSuccessfulInitClient_whenInit_thenOk() throws Exception {
+        CertPemCredentials credentials = new CertPemCredentials();
+        credentials.setCaCert("test-ca-cert.pem");
+        credentials.setPassword("test-password");
+        azureIotHubNodeConfig.setCredentials(credentials);
+        var configuration = new TbNodeConfiguration(JacksonUtil.valueToTree(azureIotHubNodeConfig));
+
+        willReturn(mqttClientMock).given(azureIotHubNode).initClient(any());
+
+        azureIotHubNode.init(ctxMock, configuration);
+
+        assertThat(azureIotHubNode.getMqttNodeConfiguration())
+                .hasFieldOrPropertyWithValue("port", 8883)
+                .hasFieldOrPropertyWithValue("cleanSession", true);
+    }
+
+    @Test
+    public void givenAzureIotHubSasCredentialsAndFailedInitClient_whenInit_thenThrowsException() throws Exception {
+        AzureIotHubSasCredentials credentials = new AzureIotHubSasCredentials();
+        credentials.setSasKey("testSasKey");
+        credentials.setCaCert("test-ca-cert.pem");
+        azureIotHubNodeConfig.setCredentials(credentials);
+
+        String errorMsg = "Failed to connect to MQTT broker!";
+        willThrow(new RuntimeException(errorMsg)).given(azureIotHubNode).initClient(any());
+
+        var configuration = new TbNodeConfiguration(JacksonUtil.valueToTree(azureIotHubNodeConfig));
+        assertThatThrownBy(() -> azureIotHubNode.init(ctxMock, configuration))
+                .isInstanceOf(TbNodeException.class)
+                .hasMessage(RuntimeException.class.getName() + ": " + errorMsg);
+    }
+}