Browse Source

test(attributes): MQTT integration tests for unified request

Device: empty-value all-shared/all-client, empty-object all-both.
Gateway: new clientKeys/sharedKeys format -> scope-separated {client,shared} response
(all-both and all-shared). Existing legacy tests untouched.
pull/15865/head
dshvaika 3 months ago
parent
commit
dfa4d1b286
  1. 80
      application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java
  2. 43
      application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java

80
application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java

@ -92,10 +92,10 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
" }\n" + " }\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\"}}}"; "\"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\"}}}"; "\"sharedJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}";
private static final String SHARED_ATTRIBUTES_DELETED_RESPONSE = "{\"deleted\":[\"sharedJson\"]}"; private static final String SHARED_ATTRIBUTES_DELETED_RESPONSE = "{\"deleted\":[\"sharedJson\"]}";
@ -369,6 +369,32 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
client.disconnect(); 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<EntityKey> 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 { protected void processProtoTestRequestAttributesValuesFromTheServer(String attrPubTopic, String attrSubTopic, String attrReqTopicPrefix) throws Exception {
MqttTestClient client = new MqttTestClient(); MqttTestClient client = new MqttTestClient();
client.connectAndWait(accessToken); client.connectAndWait(accessToken);
@ -468,6 +494,56 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
client.disconnect(); 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<Map<String, Object>> 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<EntityKey> 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 { protected void processProtoTestGatewayRequestAttributesValuesFromTheServer() throws Exception {
MqttTestClient client = new MqttTestClient(); MqttTestClient client = new MqttTestClient();
client.connectAndWait(gatewayAccessToken); client.connectAndWait(gatewayAccessToken);

43
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 { public void testRequestAttributesValuesFromTheServerGateway() throws Exception {
processJsonTestGatewayRequestAttributesValuesFromTheServer(); 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);
}
} }

Loading…
Cancel
Save