diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java index 5f43aa7e00..7e31704436 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java @@ -92,10 +92,10 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt " }\n" + "}"; - private static final String CLIENT_ATTRIBUTES_PAYLOAD = "{\"clientStr\":\"value1\",\"clientBool\":true,\"clientDbl\":42.0,\"clientLong\":73," + + protected static final String CLIENT_ATTRIBUTES_PAYLOAD = "{\"clientStr\":\"value1\",\"clientBool\":true,\"clientDbl\":42.0,\"clientLong\":73," + "\"clientJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}"; - private static final String SHARED_ATTRIBUTES_PAYLOAD = "{\"sharedStr\":\"value1\",\"sharedBool\":true,\"sharedDbl\":42.0,\"sharedLong\":73," + + protected static final String SHARED_ATTRIBUTES_PAYLOAD = "{\"sharedStr\":\"value1\",\"sharedBool\":true,\"sharedDbl\":42.0,\"sharedLong\":73," + "\"sharedJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}"; private static final String SHARED_ATTRIBUTES_DELETED_RESPONSE = "{\"deleted\":[\"sharedJson\"]}"; @@ -369,6 +369,32 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt client.disconnect(); } + protected void processJsonTestRequestAttributesWithPayload(String attrPubTopic, String attrSubTopic, String attrReqTopicPrefix, + String requestPayload, String expectedResponse) throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(accessToken); + SingleEntityFilter dtf = new SingleEntityFilter(); + dtf.setSingleEntity(AliasEntityId.fromEntityId(savedDevice.getId())); + String clientKeysStr = "clientStr,clientBool,clientDbl,clientLong,clientJson"; + String sharedKeysStr = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; + List keys = new ArrayList<>(); + keys.addAll(getEntityKeys(List.of(clientKeysStr.split(",")), CLIENT_ATTRIBUTE)); + keys.addAll(getEntityKeys(List.of(sharedKeysStr.split(",")), SHARED_ATTRIBUTE)); + getWsClient().subscribeLatestUpdate(keys, dtf); + getWsClient().registerWaitForUpdate(2); + doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", + SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); + client.publishAndWait(attrPubTopic, CLIENT_ATTRIBUTES_PAYLOAD.getBytes()); + client.subscribeAndWait(attrSubTopic, MqttQoS.AT_MOST_ONCE); + String update = getWsClient().waitForUpdate(); + assertThat(update).as("ws update received").isNotBlank(); + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(attrSubTopic.replace("+", "1")); + client.setCallback(callback); + client.publishAndWait(attrReqTopicPrefix + "1", requestPayload.getBytes()); + validateJsonResponse(callback, expectedResponse); + client.disconnect(); + } + protected void processProtoTestRequestAttributesValuesFromTheServer(String attrPubTopic, String attrSubTopic, String attrReqTopicPrefix) throws Exception { MqttTestClient client = new MqttTestClient(); client.connectAndWait(accessToken); @@ -468,6 +494,56 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt client.disconnect(); } + protected void processJsonTestGatewayRequestAttributesSeparated(String requestPayloadSuffix, String expectedBody) throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(gatewayAccessToken); + String deviceName = "Gateway Device Request Attributes Separated"; + String postClientAttributes = "{\"" + deviceName + "\":" + CLIENT_ATTRIBUTES_PAYLOAD + "}"; + client.publishAndWait(GATEWAY_ATTRIBUTES_TOPIC, postClientAttributes.getBytes()); + + Device device = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class), + 20, 100); + assertNotNull(device); + + String clientKeysStr = "clientStr,clientBool,clientDbl,clientLong,clientJson"; + String attributeValuesUrl = "/api/plugins/telemetry/DEVICE/" + device.getId() + "/values/attributes/CLIENT_SCOPE?keys=" + clientKeysStr; + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .ignoreExceptions() + .until(() -> { + List> attributes = doGetAsyncTyped(attributeValuesUrl, new TypeReference<>() { + }); + return attributes.size() == 5; + }); + + SingleEntityFilter dtf = new SingleEntityFilter(); + dtf.setSingleEntity(AliasEntityId.fromEntityId(device.getId())); + String sharedKeysStr = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; + List keys = new ArrayList<>(); + keys.addAll(getEntityKeys(List.of(clientKeysStr.split(",")), CLIENT_ATTRIBUTE)); + keys.addAll(getEntityKeys(List.of(sharedKeysStr.split(",")), SHARED_ATTRIBUTE)); + EntityDataUpdate initUpdate = getWsClient().subscribeLatestUpdate(keys, dtf); + assertNotNull(initUpdate); + assertFalse(initUpdate.getData().getData().isEmpty()); + getWsClient().registerWaitForUpdate(); + + doPostAsync("/api/plugins/telemetry/DEVICE/" + device.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); + String update = getWsClient().waitForUpdate(); + assertThat(update).as("ws update received").isNotBlank(); + + client.subscribeAndWait(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, MqttQoS.AT_LEAST_ONCE); + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC); + client.setCallback(callback); + String requestPayloadStr = "{\"id\": 1, \"device\": \"" + deviceName + "\"" + requestPayloadSuffix + "}"; + client.publishAndWait(GATEWAY_ATTRIBUTES_REQUEST_TOPIC, requestPayloadStr.getBytes()); + + assertThat(callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)).as("await callback").isTrue(); + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), callback.getMessageArrivedQoS()); + String expected = "{\"id\":1,\"device\":\"" + deviceName + "\"" + expectedBody + "}"; + assertEquals(JacksonUtil.toJsonNode(expected), JacksonUtil.fromBytes(callback.getPayloadBytes())); + client.disconnect(); + } + protected void processProtoTestGatewayRequestAttributesValuesFromTheServer() throws Exception { MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java index 3786a133ef..77e2df7b5c 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java @@ -57,4 +57,47 @@ public class MqttAttributesRequestJsonIntegrationTest extends AbstractMqttAttrib public void testRequestAttributesValuesFromTheServerGateway() throws Exception { processJsonTestGatewayRequestAttributesValuesFromTheServer(); } + + @Test + public void testRequestAllSharedAttributesViaEmptyValue() throws Exception { + // sharedKeys present + empty => all shared; clientKeys absent => no client + processJsonTestRequestAttributesWithPayload( + MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX, + "{\"sharedKeys\":\"\"}", + "{\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testRequestAllClientAttributesViaEmptyValue() throws Exception { + // clientKeys present + empty => all client; sharedKeys absent => no shared + processJsonTestRequestAttributesWithPayload( + MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX, + "{\"clientKeys\":\"\"}", + "{\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testRequestAllAttributesViaEmptyObject() throws Exception { + // both fields absent => fetch everything (both scopes) + processJsonTestRequestAttributesWithPayload( + MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX, + "{}", + "{\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testGatewayRequestAllAttributesSeparated() throws Exception { + // new format: empty clientKeys + empty sharedKeys => all both, scope-separated response + processJsonTestGatewayRequestAttributesSeparated( + ", \"clientKeys\": \"\", \"sharedKeys\": \"\"", + ",\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD); + } + + @Test + public void testGatewayRequestAllSharedSeparated() throws Exception { + // new format: empty sharedKeys only => all shared, no client in the separated response + processJsonTestGatewayRequestAttributesSeparated( + ", \"sharedKeys\": \"\"", + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD); + } }