From 947a72b2c912558d08a30c557bcc6e75f923c065 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Fri, 11 Mar 2022 13:58:12 +0200 Subject: [PATCH] added mqtt tests with malformed publish payload --- .../mqtt/AbstractMqttIntegrationTest.java | 24 ++++++-- ...tBackwardCompatibilityIntegrationTest.java | 8 +-- ...AttributesRequestProtoIntegrationTest.java | 6 +- .../AbstractMqttProvisionJsonDeviceTest.java | 14 ++--- .../AbstractMqttProvisionProtoDeviceTest.java | 14 ++--- ...cBackwardCompatibilityIntegrationTest.java | 20 +++--- ...MqttServerSideRpcProtoIntegrationTest.java | 2 +- ...AbstractMqttTimeseriesIntegrationTest.java | 33 ++++++++++ ...ractMqttTimeseriesJsonIntegrationTest.java | 42 +++++++++++-- ...actMqttTimeseriesProtoIntegrationTest.java | 61 +++++++++++++++++-- 10 files changed, 178 insertions(+), 46 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java index 88c1b5f5a1..2dde81a6a2 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java @@ -62,11 +62,19 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg protected DeviceProfile deviceProfile; protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic) throws Exception { - this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); } protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic, boolean enableCompatibilityWithJsonPayloadFormat, boolean useJsonPayloadFormatForDefaultDownlinkTopics) throws Exception { - this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics); + this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics, false); + } + + protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic, boolean enableCompatibilityWithJsonPayloadFormat, boolean useJsonPayloadFormatForDefaultDownlinkTopics, boolean sendPubAckOnValidationException) throws Exception { + this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics, sendPubAckOnValidationException); + } + + protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic, boolean sendPubAckOnValidationException) throws Exception { + this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, sendPubAckOnValidationException); } protected void processBeforeTest(String deviceName, @@ -82,7 +90,8 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg String provisionSecret, DeviceProfileProvisionType provisionType, boolean enableCompatibilityWithJsonPayloadFormat, - boolean useJsonPayloadFormatForDefaultDownlinkTopics) throws Exception { + boolean useJsonPayloadFormatForDefaultDownlinkTopics, + boolean sendPubAckOnValidationException) throws Exception { loginSysAdmin(); Tenant tenant = new Tenant(); @@ -111,7 +120,10 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg gateway.setAdditionalInfo(additionalInfo); if (payloadType != null) { - DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic, telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema, provisionKey, provisionSecret, provisionType, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics); + DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic, + telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema, + provisionKey, provisionSecret, provisionType, enableCompatibilityWithJsonPayloadFormat, + useJsonPayloadFormatForDefaultDownlinkTopics, sendPubAckOnValidationException); deviceProfile = doPost("/api/deviceProfile", mqttDeviceProfile, DeviceProfile.class); device.setType(deviceProfile.getName()); device.setDeviceProfileId(deviceProfile.getId()); @@ -169,7 +181,8 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg String provisionKey, String provisionSecret, DeviceProfileProvisionType provisionType, boolean enableCompatibilityWithJsonPayloadFormat, - boolean useJsonPayloadFormatForDefaultDownlinkTopics) { + boolean useJsonPayloadFormatForDefaultDownlinkTopics, + boolean sendPubAckOnValidationException) { DeviceProfile deviceProfile = new DeviceProfile(); deviceProfile.setName(transportPayloadType.name()); deviceProfile.setType(DeviceProfileType.DEFAULT); @@ -186,6 +199,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg if (!StringUtils.isEmpty(attributesTopic)) { mqttDeviceProfileTransportConfiguration.setDeviceAttributesTopic(attributesTopic); } + mqttDeviceProfileTransportConfiguration.setSendPubAckOnValidationException(sendPubAckOnValidationException); TransportPayloadTypeConfiguration transportPayloadTypeConfiguration; if (TransportPayloadType.JSON.equals(transportPayloadType)) { transportPayloadTypeConfiguration = new JsonTransportPayloadConfiguration(); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestBackwardCompatibilityIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestBackwardCompatibilityIntegrationTest.java index b7878154be..415cd8533b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestBackwardCompatibilityIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestBackwardCompatibilityIntegrationTest.java @@ -38,28 +38,28 @@ public abstract class AbstractMqttAttributesRequestBackwardCompatibilityIntegrat @Test public void testRequestAttributesValuesFromTheServerWithEnabledJsonCompatibility() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, false); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, false, false); processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX); } @Test public void testRequestAttributesValuesFromTheServerWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processJsonTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX); } @Test public void testRequestAttributesValuesFromTheServerOnShortTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX); } @Test public void testRequestAttributesValuesFromTheServerOnShortProtoTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java index 4c88cbbaec..bbd736f9ab 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java @@ -38,21 +38,21 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends @Test public void testRequestAttributesValuesFromTheServer() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX); } @Test public void testRequestAttributesValuesFromTheServerOnShortTopic() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX); } @Test public void testRequestAttributesValuesFromTheServerOnShortProtoTopic() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java index 876b40992a..43b6deb12a 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java @@ -94,7 +94,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningDisabledDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString()); @@ -103,7 +103,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -119,7 +119,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithAccessToken() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); String requestCredentials = ",\"credentialsType\": \"ACCESS_TOKEN\",\"token\": \"test_token\""; byte[] result = createMqttClientAndPublish(requestCredentials).getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -138,7 +138,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithCert() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); String requestCredentials = ",\"credentialsType\": \"X509_CERTIFICATE\",\"hash\": \"testHash\""; byte[] result = createMqttClientAndPublish(requestCredentials).getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -163,7 +163,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithMqttBasic() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); String requestCredentials = ",\"credentialsType\": \"MQTT_BASIC\",\"clientId\": \"test_clientId\",\"username\": \"test_username\",\"password\": \"test_password\""; byte[] result = createMqttClientAndPublish(requestCredentials).getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -188,7 +188,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn } protected void processTestProvisioningCheckPreProvisionedDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -199,7 +199,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn } protected void processTestProvisioningWithBadKeyDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString()); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java index e05b5a82b3..6d16096e4c 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java @@ -101,14 +101,14 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI protected void processTestProvisioningDisabledDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); ProvisionDeviceResponseMsg result = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); Assert.assertNotNull(result); Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), result.getStatus().toString()); } protected void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); Device createdDevice = deviceService.findDeviceByTenantIdAndName(savedTenant.getTenantId(), "Test Provision device"); @@ -122,7 +122,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCreateNewDeviceWithAccessToken() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null,null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null,null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateDeviceTokenRequestMsg(ValidateDeviceTokenRequestMsg.newBuilder().setToken("test_token").build()).build(); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish(createTestsProvisionMessage(CredentialsType.ACCESS_TOKEN, requestCredentials)).getPayloadBytes()); @@ -140,7 +140,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCreateNewDeviceWithCert() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateDeviceX509CertRequestMsg(ValidateDeviceX509CertRequestMsg.newBuilder().setHash("testHash").build()).build(); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish(createTestsProvisionMessage(CredentialsType.X509_CERTIFICATE, requestCredentials)).getPayloadBytes()); @@ -164,7 +164,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCreateNewDeviceWithMqttBasic() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateBasicMqttCredRequestMsg( ValidateBasicMqttCredRequestMsg.newBuilder() .setClientId("test_clientId") @@ -195,7 +195,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCheckPreProvisionedDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(savedTenant.getTenantId(), savedDevice.getId()); @@ -205,7 +205,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningWithBadKeyDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), response.getStatus().toString()); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcBackwardCompatibilityIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcBackwardCompatibilityIntegrationTest.java index 82757801aa..003ced8ea5 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcBackwardCompatibilityIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcBackwardCompatibilityIntegrationTest.java @@ -33,61 +33,61 @@ public abstract class AbstractMqttServerSideRpcBackwardCompatibilityIntegrationT @Test public void testServerMqttOneWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC); } @Test public void testServerMqttOneWayRpcOnShortTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC); } @Test public void testServerMqttOneWayRpcOnShortProtoTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC); } @Test public void testServerMqttTwoWayRpcWithEnabledJsonCompatibility() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, false); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, false, false); processProtoTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC); } @Test public void testServerMqttTwoWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processJsonTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC); } @Test public void testServerMqttTwoWayRpcOnShortTopic() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processProtoTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC); } @Test public void testServerMqttTwoWayRpcOnShortProtoTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processProtoTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC); } @Test public void testServerMqttTwoWayRpcOnShortJsonTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processJsonTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC); } @Test public void testGatewayServerMqttOneWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processProtoOneWayRpcTestGateway("Gateway Device OneWay RPC Proto"); } @Test public void testGatewayServerMqttTwoWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception { - super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true); + super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false); processProtoTwoWayRpcTestGateway("Gateway Device TwoWay RPC Proto"); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java index db70e99b2f..aaae2c2099 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java @@ -28,7 +28,7 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst @Before public void beforeTest() throws Exception { - processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, false, false); + processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); } @After diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 57f3ee9701..761407c92c 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -23,6 +23,7 @@ import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.junit.After; import org.junit.Before; @@ -50,6 +51,9 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt protected static final String PAYLOAD_VALUES_STR = "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4," + " \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; + protected static final String MALFORMED_JSON_PAYLOAD = "{\"key1\":, \"key2\":true, \"key3\": 3.0, \"key4\": 4," + + " \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; + @Before public void beforeTest() throws Exception { processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", null, null, null); @@ -368,5 +372,34 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt } } + public static class TestMqttPublishCallback implements MqttCallback { + + private final CountDownLatch latch; + private boolean pubAckReceived; + + public boolean isPubAckReceived() { + return pubAckReceived; + } + + public TestMqttPublishCallback(CountDownLatch latch) { + this.latch = latch; + } + + @Override + public void connectionLost(Throwable throwable) { + latch.countDown(); + } + + @Override + public void messageArrived(String s, MqttMessage mqttMessage) throws Exception { + } + + @Override + public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { + pubAckReceived = iMqttDeliveryToken.getResponse().getType() == MqttWireMessage.MESSAGE_TYPE_PUBACK; + latch.countDown(); + } + } + } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java index 85fb96bfec..cf2c751264 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java @@ -20,14 +20,15 @@ import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.TransportPayloadType; -import org.thingsboard.server.common.data.device.profile.MqttTopics; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; -import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; @Slf4j public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends AbstractMqttTimeseriesIntegrationTest { @@ -36,7 +37,7 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract @Before public void beforeTest() throws Exception { - processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); + //do nothing, processBeforeTest will be invoked in particular test methods with different parameters } @After @@ -46,12 +47,14 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract @Test public void testPushTelemetry() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); processJsonPayloadTelemetryTest(POST_DATA_TELEMETRY_TOPIC, expectedKeys, PAYLOAD_VALUES_STR.getBytes(), false); } @Test public void testPushTelemetryWithTs() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); String payloadStr = "{\"ts\": 10000, \"values\": " + PAYLOAD_VALUES_STR + "}"; List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); processJsonPayloadTelemetryTest(POST_DATA_TELEMETRY_TOPIC, expectedKeys, payloadStr.getBytes(), true); @@ -59,21 +62,50 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract @Test public void testPushTelemetryOnShortTopic() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); super.testPushTelemetryOnShortTopic(); } @Test - public void testPushTelemetryWithTsOnShortJsonTopic() throws Exception { + public void testPushTelemetryOnShortJsonTopic() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); super.testPushTelemetryOnShortJsonTopic(); } @Test public void testPushTelemetryGateway() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); super.testPushTelemetryGateway(); } @Test public void testGatewayConnect() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null); super.testGatewayConnect(); } + + @Test + public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorEnabled() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null, true); + CountDownLatch latch = new CountDownLatch(1); + MqttAsyncClient client = getMqttAsyncClient(accessToken); + TestMqttPublishCallback callback = new TestMqttPublishCallback(latch); + client.setCallback(callback); + publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC); + latch.await(3, TimeUnit.SECONDS); + assertTrue(callback.isPubAckReceived()); + } + + @Test + public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorDisabled() throws Exception { + processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null, false); + CountDownLatch latch = new CountDownLatch(1); + MqttAsyncClient client = getMqttAsyncClient(accessToken); + TestMqttPublishCallback callback = new TestMqttPublishCallback(latch); + client.setCallback(callback); + publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC); + latch.await(3, TimeUnit.SECONDS); + assertFalse(callback.isPubAckReceived()); + } + } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java index ecb2233332..fb6a8e1f9b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java @@ -36,7 +36,10 @@ import org.thingsboard.server.gen.transport.TransportProtos; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -44,6 +47,7 @@ import static org.junit.Assert.assertTrue; public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends AbstractMqttTimeseriesIntegrationTest { private static final String POST_DATA_TELEMETRY_TOPIC = "proto/telemetry"; + private static final String MALFORMED_PROTO_PAYLOAD = "invalid proto payload str"; @Before @Override @@ -91,7 +95,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac " }\n" + " }\n" + "}"; - processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, schemaStr, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, schemaStr, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); DynamicSchema telemetrySchema = getDynamicSchema(schemaStr); DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject"); @@ -193,7 +197,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac " }\n" + " }\n" + "}"; - processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, schemaStr, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, schemaStr, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); DynamicSchema telemetrySchema = getDynamicSchema(schemaStr); DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject"); @@ -253,7 +257,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac @Test public void testPushTelemetryGateway() throws Exception { - processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); TransportApiProtos.GatewayTelemetryMsg.Builder gatewayTelemetryMsgProtoBuilder = TransportApiProtos.GatewayTelemetryMsg.newBuilder(); List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); String deviceName1 = "Device A"; @@ -267,7 +271,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac @Test public void testGatewayConnect() throws Exception { - processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false); + processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false); String deviceName = "Device A"; TransportApiProtos.ConnectMsg connectMsgProto = getConnectProto(deviceName); MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken); @@ -280,6 +284,55 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac assertNotNull(device); } + + @Test + public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorEnabled() throws Exception { + processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true); + CountDownLatch latch = new CountDownLatch(1); + MqttAsyncClient client = getMqttAsyncClient(accessToken); + TestMqttPublishCallback callback = new TestMqttPublishCallback(latch); + client.setCallback(callback); + publishMqttMsg(client, MALFORMED_PROTO_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC); + latch.await(3, TimeUnit.SECONDS); + assertTrue(callback.isPubAckReceived()); + } + + @Test + public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorDisabled() throws Exception { + processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, false); + CountDownLatch latch = new CountDownLatch(1); + MqttAsyncClient client = getMqttAsyncClient(accessToken); + TestMqttPublishCallback callback = new TestMqttPublishCallback(latch); + client.setCallback(callback); + publishMqttMsg(client, MALFORMED_PROTO_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC); + latch.await(3, TimeUnit.SECONDS); + assertFalse(callback.isPubAckReceived()); + } + + @Test + public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorEnabledAndBackwardCompatibilityEnabled() throws Exception { + processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, true); + CountDownLatch latch = new CountDownLatch(1); + MqttAsyncClient client = getMqttAsyncClient(accessToken); + TestMqttPublishCallback callback = new TestMqttPublishCallback(latch); + client.setCallback(callback); + publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC); + latch.await(3, TimeUnit.SECONDS); + assertTrue(callback.isPubAckReceived()); + } + + @Test + public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorDisabledAndBackwardCompatibilityEnabled() throws Exception { + processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, false); + CountDownLatch latch = new CountDownLatch(1); + MqttAsyncClient client = getMqttAsyncClient(accessToken); + TestMqttPublishCallback callback = new TestMqttPublishCallback(latch); + client.setCallback(callback); + publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC); + latch.await(3, TimeUnit.SECONDS); + assertFalse(callback.isPubAckReceived()); + } + private DynamicSchema getDynamicSchema(String deviceTelemetryProtoSchema) { DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);