tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #1183 from vkukhtyn/master 

Cover MQTT API with black box tests
pull/1211/head
Igor Kulikov 8 years ago
committed by GitHub
parent
13893856c3 32dd7f0437
commit
000741444a
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 991 additions and 1 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 21
      msa/black-box-tests/README.md
  2. 104
      msa/black-box-tests/pom.xml
  3. 175
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java
  4. 39
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java
  5. 76
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/WsClient.java
  6. 57
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java
  7. 392
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java
  8. 26
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java
  9. 40
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java
  10. 59
      msa/black-box-tests/src/test/resources/RpcResponseRuleChainMetadata.json
  11. 3
      msa/pom.xml

21
msa/black-box-tests/README.md
View File

@ -0,0 +1,21 @@
## Black box tests execution
To run the black box tests with using Docker, the local Docker images of Thingsboard's microservices should be built. <br />
- Build the local Docker images in the directory with the Thingsboard's main [pom.xml](./../../pom.xml):
mvn clean install -Ddockerfile.skip=false
- Verify that the new local images were built:
docker image ls
As result, in REPOSITORY column, next images should be present:
thingsboard/tb-coap-transport
thingsboard/tb-http-transport
thingsboard/tb-mqtt-transport
thingsboard/tb-node
thingsboard/tb-web-ui
thingsboard/tb-js-executor
- Run the black box tests in the [msa/black-box-tests](../black-box-tests) directory:
mvn clean install -DblackBoxTests.skip=false

104
msa/black-box-tests/pom.xml
View File

@ -0,0 +1,104 @@
<!--
Copyright © 2016-2018 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.thingsboard</groupId>
<version>2.2.0-SNAPSHOT</version>
<artifactId>msa</artifactId>
</parent>
<groupId>org.thingsboard.msa</groupId>
<artifactId>black-box-tests</artifactId>
<name>ThingsBoard Black Box Tests</name>
<url>https://thingsboard.io</url>
<description>Project for ThingsBoard black box testing with using Docker</description>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<main.dir>${basedir}/../..</main.dir>
<blackBoxTests.skip>true</blackBoxTests.skip>
<testcontainers.version>1.9.1</testcontainers.version>
<java-websocket.version>1.3.9</java-websocket.version>
<httpclient.version>4.5.6</httpclient.version>
</properties>
<dependencies>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>${testcontainers.version}</version>
</dependency>
<dependency>
<groupId>org.java-websocket</groupId>
<artifactId>Java-WebSocket</artifactId>
<version>${java-websocket.version}</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>${httpclient.version}</version>
</dependency>
<dependency>
<groupId>io.takari.junit</groupId>
<artifactId>takari-cpsuite</artifactId>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>
<dependency>
<groupId>org.thingsboard</groupId>
<artifactId>netty-mqtt</artifactId>
</dependency>
<dependency>
<groupId>org.thingsboard</groupId>
<artifactId>tools</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<includes>
<include>**/*TestSuite.java</include>
</includes>
<skipTests>${blackBoxTests.skip}</skipTests>
</configuration>
</plugin>
</plugins>
</build>
</project>

175
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java
View File

@ -0,0 +1,175 @@
/**
* Copyright © 2016-2018 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.server.msa;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustStrategy;
import org.apache.http.conn.ssl.X509HostnameVerifier;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.SSLContexts;
import org.junit.*;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.thingsboard.client.tools.RestClient;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
import javax.net.ssl.*;
import java.net.URI;
import java.security.cert.X509Certificate;
import java.util.List;
import java.util.Map;
import java.util.Random;
@Slf4j
public abstract class AbstractContainerTest {
protected static final String HTTPS_URL = "https://localhost";
protected static final String WSS_URL = "wss://localhost";
protected static RestClient restClient;
protected ObjectMapper mapper = new ObjectMapper();
@BeforeClass
public static void before() throws Exception {
restClient = new RestClient(HTTPS_URL);
restClient.getRestTemplate().setRequestFactory(getRequestFactoryForSelfSignedCert());
}
protected Device createDevice(String name) {
return restClient.createDevice(name + RandomStringUtils.randomAlphanumeric(7), "DEFAULT");
}
protected WsClient subscribeToWebSocket(DeviceId deviceId, String scope, CmdsType property) throws Exception {
WsClient wsClient = new WsClient(new URI(WSS_URL + "/api/ws/plugins/telemetry?token=" + restClient.getToken()));
SSLContextBuilder builder = SSLContexts.custom();
builder.loadTrustMaterial(null, (TrustStrategy) (chain, authType) -> true);
wsClient.setSocket(builder.build().getSocketFactory().createSocket());
wsClient.connectBlocking();
JsonObject cmdsObject = new JsonObject();
cmdsObject.addProperty("entityType", EntityType.DEVICE.name());
cmdsObject.addProperty("entityId", deviceId.toString());
cmdsObject.addProperty("scope", scope);
cmdsObject.addProperty("cmdId", new Random().nextInt(100));
JsonArray cmd = new JsonArray();
cmd.add(cmdsObject);
JsonObject wsRequest = new JsonObject();
wsRequest.add(property.toString(), cmd);
wsClient.send(wsRequest.toString());
return wsClient;
}
protected Map<String, Long> getExpectedLatestValues(long ts) {
return ImmutableMap.<String, Long>builder()
.put("booleanKey", ts)
.put("stringKey", ts)
.put("doubleKey", ts)
.put("longKey", ts)
.build();
}
protected boolean verify(WsTelemetryResponse wsTelemetryResponse, String key, Long expectedTs, String expectedValue) {
List<Object> list = wsTelemetryResponse.getDataValuesByKey(key);
return expectedTs.equals(list.get(0)) && expectedValue.equals(list.get(1));
}
protected boolean verify(WsTelemetryResponse wsTelemetryResponse, String key, String expectedValue) {
List<Object> list = wsTelemetryResponse.getDataValuesByKey(key);
return expectedValue.equals(list.get(1));
}
protected JsonObject createPayload(long ts) {
JsonObject values = createPayload();
JsonObject payload = new JsonObject();
payload.addProperty("ts", ts);
payload.add("values", values);
return payload;
}
protected JsonObject createPayload() {
JsonObject values = new JsonObject();
values.addProperty("stringKey", "value1");
values.addProperty("booleanKey", true);
values.addProperty("doubleKey", 42.0);
values.addProperty("longKey", 73L);
return values;
}
protected enum CmdsType {
TS_SUB_CMDS("tsSubCmds"),
HISTORY_CMDS("historyCmds"),
ATTR_SUB_CMDS("attrSubCmds");
private final String text;
CmdsType(final String text) {
this.text = text;
}
@Override
public String toString() {
return text;
}
}
private static HttpComponentsClientHttpRequestFactory getRequestFactoryForSelfSignedCert() throws Exception {
SSLContextBuilder builder = SSLContexts.custom();
builder.loadTrustMaterial(null, (TrustStrategy) (chain, authType) -> true);
SSLContext sslContext = builder.build();
SSLConnectionSocketFactory sslSelfSigned = new SSLConnectionSocketFactory(sslContext, new X509HostnameVerifier() {
@Override
public void verify(String host, SSLSocket ssl) {
}
@Override
public void verify(String host, X509Certificate cert) {
}
@Override
public void verify(String host, String[] cns, String[] subjectAlts) {
}
@Override
public boolean verify(String s, SSLSession sslSession) {
return true;
}
});
Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder
.<ConnectionSocketFactory>create()
.register("https", sslSelfSigned)
.build();
PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(socketFactoryRegistry);
CloseableHttpClient httpClient = HttpClients.custom().setConnectionManager(cm).build();
return new HttpComponentsClientHttpRequestFactory(httpClient);
}
}

39
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java
View File

@ -0,0 +1,39 @@
/**
* Copyright © 2016-2018 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.server.msa;
import org.junit.ClassRule;
import org.junit.extensions.cpsuite.ClasspathSuite;
import org.junit.runner.RunWith;
import org.testcontainers.containers.DockerComposeContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import java.io.File;
import java.time.Duration;
@RunWith(ClasspathSuite.class)
@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*Test"})
public class ContainerTestSuite {
@ClassRule
public static DockerComposeContainer composeContainer = new DockerComposeContainer(
new File("./../../docker/docker-compose.yml"),
new File("./../../docker/docker-compose.postgres.yml"))
.withPull(false)
.withLocalCompose(true)
.withTailChildContainers(true)
.withExposedService("tb-web-ui1", 8080, Wait.forHttp("/login").withStartupTimeout(Duration.ofSeconds(120)));
}

76
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/WsClient.java
View File

@ -0,0 +1,76 @@
/**
* Copyright © 2016-2018 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.server.msa;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@Slf4j
public class WsClient extends WebSocketClient {
private static final ObjectMapper mapper = new ObjectMapper();
private WsTelemetryResponse message;
private CountDownLatch latch = new CountDownLatch(1);;
public WsClient(URI serverUri) {
super(serverUri);
}
@Override
public void onOpen(ServerHandshake serverHandshake) {
}
@Override
public void onMessage(String message) {
try {
WsTelemetryResponse response = mapper.readValue(message, WsTelemetryResponse.class);
if (!response.getData().isEmpty()) {
this.message = response;
latch.countDown();
}
} catch (IOException e) {
log.error("ws message can't be read");
}
}
@Override
public void onClose(int code, String reason, boolean remote) {
log.info("ws is closed, due to [{}]", reason);
}
@Override
public void onError(Exception ex) {
ex.printStackTrace();
}
public WsTelemetryResponse getLastMessage() {
try {
latch.await(10, TimeUnit.SECONDS);
return this.message;
} catch (InterruptedException e) {
log.error("Timeout, ws message wasn't received");
}
return null;
}
}

57
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java
View File

@ -0,0 +1,57 @@
/**
* Copyright © 2016-2018 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.server.msa.connectivity;
import com.google.common.collect.Sets;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.http.ResponseEntity;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.msa.AbstractContainerTest;
import org.thingsboard.server.msa.WsClient;
import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
public class HttpClientTest extends AbstractContainerTest {
@Test
public void telemetryUpload() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("http_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS);
ResponseEntity deviceTelemetryResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/v1/{credentialsId}/telemetry",
mapper.readTree(createPayload().toString()),
ResponseEntity.class,
deviceCredentials.getCredentialsId());
Assert.assertTrue(deviceTelemetryResponse.getStatusCode().is2xxSuccessful());
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage();
wsClient.closeBlocking();
Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"),
actualLatestTelemetry.getLatestValues().keySet());
Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString()));
Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1"));
Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0)));
Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73)));
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
}

392
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java
View File

@ -0,0 +1,392 @@
/**
* Copyright © 2016-2018 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.server.msa.connectivity;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.gson.JsonObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.*;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.thingsboard.mqtt.MqttClient;
import org.thingsboard.mqtt.MqttClientConfig;
import org.thingsboard.mqtt.MqttHandler;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.page.TextPageData;
import org.thingsboard.server.common.data.rule.NodeConnectionInfo;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainMetaData;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.msa.AbstractContainerTest;
import org.thingsboard.server.msa.WsClient;
import org.thingsboard.server.msa.mapper.AttributesResponse;
import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
@Slf4j
public class MqttClientTest extends AbstractContainerTest {
@Test
public void telemetryUpload() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS);
MqttClient mqttClient = getMqttClient(deviceCredentials, null);
mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload().toString().getBytes()));
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage();
wsClient.closeBlocking();
Assert.assertEquals(4, actualLatestTelemetry.getData().size());
Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"),
actualLatestTelemetry.getLatestValues().keySet());
Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString()));
Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1"));
Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0)));
Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73)));
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
@Test
public void telemetryUploadWithTs() throws Exception {
long ts = 1451649600512L;
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS);
MqttClient mqttClient = getMqttClient(deviceCredentials, null);
mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload(ts).toString().getBytes()));
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage();
wsClient.closeBlocking();
Assert.assertEquals(4, actualLatestTelemetry.getData().size());
Assert.assertEquals(getExpectedLatestValues(ts), actualLatestTelemetry.getLatestValues());
Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", ts, Boolean.TRUE.toString()));
Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", ts, "value1"));
Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", ts, Double.toString(42.0)));
Assert.assertTrue(verify(actualLatestTelemetry, "longKey", ts, Long.toString(73)));
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
@Test
public void publishAttributeUpdateToServer() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
WsClient wsClient = subscribeToWebSocket(device.getId(), "CLIENT_SCOPE", CmdsType.ATTR_SUB_CMDS);
MqttMessageListener listener = new MqttMessageListener();
MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
JsonObject clientAttributes = new JsonObject();
clientAttributes.addProperty("attr1", "value1");
clientAttributes.addProperty("attr2", true);
clientAttributes.addProperty("attr3", 42.0);
clientAttributes.addProperty("attr4", 73);
mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes()));
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage();
wsClient.closeBlocking();
Assert.assertEquals(4, actualLatestTelemetry.getData().size());
Assert.assertEquals(Sets.newHashSet("attr1", "attr2", "attr3", "attr4"),
actualLatestTelemetry.getLatestValues().keySet());
Assert.assertTrue(verify(actualLatestTelemetry, "attr1", "value1"));
Assert.assertTrue(verify(actualLatestTelemetry, "attr2", Boolean.TRUE.toString()));
Assert.assertTrue(verify(actualLatestTelemetry, "attr3", Double.toString(42.0)));
Assert.assertTrue(verify(actualLatestTelemetry, "attr4", Long.toString(73)));
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
@Test
public void requestAttributeValuesFromServer() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
MqttMessageListener listener = new MqttMessageListener();
MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
// Add a new client attribute
JsonObject clientAttributes = new JsonObject();
String clientAttributeValue = RandomStringUtils.randomAlphanumeric(8);
clientAttributes.addProperty("clientAttr", clientAttributeValue);
mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes()));
// Add a new shared attribute
JsonObject sharedAttributes = new JsonObject();
String sharedAttributeValue = RandomStringUtils.randomAlphanumeric(8);
sharedAttributes.addProperty("sharedAttr", sharedAttributeValue);
ResponseEntity sharedAttributesResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
device.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
// Subscribe to attributes response
mqttClient.on("v1/devices/me/attributes/response/+", listener, MqttQoS.AT_LEAST_ONCE);
// Request attributes
JsonObject request = new JsonObject();
request.addProperty("clientKeys", "clientAttr");
request.addProperty("sharedKeys", "sharedAttr");
mqttClient.publish("v1/devices/me/attributes/request/" + new Random().nextInt(100), Unpooled.wrappedBuffer(request.toString().getBytes()));
MqttEvent event = listener.getEvents().poll(10, TimeUnit.SECONDS);
AttributesResponse attributes = mapper.readValue(Objects.requireNonNull(event).getMessage(), AttributesResponse.class);
Assert.assertEquals(1, attributes.getClient().size());
Assert.assertEquals(clientAttributeValue, attributes.getClient().get("clientAttr"));
Assert.assertEquals(1, attributes.getShared().size());
Assert.assertEquals(sharedAttributeValue, attributes.getShared().get("sharedAttr"));
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
@Test
public void subscribeToAttributeUpdatesFromServer() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
MqttMessageListener listener = new MqttMessageListener();
MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
mqttClient.on("v1/devices/me/attributes", listener, MqttQoS.AT_LEAST_ONCE);
String sharedAttributeName = "sharedAttr";
// Add a new shared attribute
JsonObject sharedAttributes = new JsonObject();
String sharedAttributeValue = RandomStringUtils.randomAlphanumeric(8);
sharedAttributes.addProperty(sharedAttributeName, sharedAttributeValue);
ResponseEntity sharedAttributesResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
device.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
MqttEvent event = listener.getEvents().poll(10, TimeUnit.SECONDS);
Assert.assertEquals(sharedAttributeValue,
mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText());
// Update the shared attribute value
JsonObject updatedSharedAttributes = new JsonObject();
String updatedSharedAttributeValue = RandomStringUtils.randomAlphanumeric(8);
updatedSharedAttributes.addProperty(sharedAttributeName, updatedSharedAttributeValue);
ResponseEntity updatedSharedAttributesResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
mapper.readTree(updatedSharedAttributes.toString()), ResponseEntity.class,
device.getId());
Assert.assertTrue(updatedSharedAttributesResponse.getStatusCode().is2xxSuccessful());
event = listener.getEvents().poll(10, TimeUnit.SECONDS);
Assert.assertEquals(updatedSharedAttributeValue,
mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText());
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
@Test
public void serverSideRpc() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
MqttMessageListener listener = new MqttMessageListener();
MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
mqttClient.on("v1/devices/me/rpc/request/+", listener, MqttQoS.AT_LEAST_ONCE);
// Send an RPC from the server
JsonObject serverRpcPayload = new JsonObject();
serverRpcPayload.addProperty("method", "getValue");
serverRpcPayload.addProperty("params", true);
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
ListenableFuture<ResponseEntity> future = service.submit(() -> {
try {
return restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/plugins/rpc/twoway/{deviceId}",
mapper.readTree(serverRpcPayload.toString()), String.class,
device.getId());
} catch (IOException e) {
return ResponseEntity.badRequest().build();
}
});
// Wait for RPC call from the server and send the response
MqttEvent requestFromServer = listener.getEvents().poll(10, TimeUnit.SECONDS);
Assert.assertEquals("{\"method\":\"getValue\",\"params\":true}", Objects.requireNonNull(requestFromServer).getMessage());
Integer requestId = Integer.valueOf(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length()));
JsonObject clientResponse = new JsonObject();
clientResponse.addProperty("response", "someResponse");
// Send a response to the server's RPC request
mqttClient.publish("v1/devices/me/rpc/response/" + requestId, Unpooled.wrappedBuffer(clientResponse.toString().getBytes()));
ResponseEntity serverResponse = future.get(5, TimeUnit.SECONDS);
Assert.assertTrue(serverResponse.getStatusCode().is2xxSuccessful());
Assert.assertEquals(clientResponse.toString(), serverResponse.getBody());
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
@Test
public void clientSideRpc() throws Exception {
restClient.login("tenant@thingsboard.org", "tenant");
Device device = createDevice("mqtt_");
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
MqttMessageListener listener = new MqttMessageListener();
MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
mqttClient.on("v1/devices/me/rpc/request/+", listener, MqttQoS.AT_LEAST_ONCE);
// Get the default rule chain id to make it root again after test finished
RuleChainId defaultRuleChainId = getDefaultRuleChainId();
// Create a new root rule chain
RuleChainId ruleChainId = createRootRuleChainForRpcResponse();
// Send the request to the server
JsonObject clientRequest = new JsonObject();
clientRequest.addProperty("method", "getResponse");
clientRequest.addProperty("params", true);
Integer requestId = 42;
mqttClient.publish("v1/devices/me/rpc/request/" + requestId, Unpooled.wrappedBuffer(clientRequest.toString().getBytes()));
// Check the response from the server
TimeUnit.SECONDS.sleep(1);
MqttEvent responseFromServer = listener.getEvents().poll(1, TimeUnit.SECONDS);
Integer responseId = Integer.valueOf(Objects.requireNonNull(responseFromServer).getTopic().substring("v1/devices/me/rpc/response/".length()));
Assert.assertEquals(requestId, responseId);
Assert.assertEquals("requestReceived", mapper.readTree(responseFromServer.getMessage()).get("response").asText());
// Make the default rule chain a root again
ResponseEntity<RuleChain> rootRuleChainResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/ruleChain/{ruleChainId}/root",
null,
RuleChain.class,
defaultRuleChainId);
Assert.assertTrue(rootRuleChainResponse.getStatusCode().is2xxSuccessful());
// Delete the created rule chain
restClient.getRestTemplate().delete(HTTPS_URL + "/api/ruleChain/{ruleChainId}", ruleChainId);
restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
}
private RuleChainId createRootRuleChainForRpcResponse() throws Exception {
RuleChain newRuleChain = new RuleChain();
newRuleChain.setName("testRuleChain");
ResponseEntity<RuleChain> ruleChainResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/ruleChain",
newRuleChain,
RuleChain.class);
Assert.assertTrue(ruleChainResponse.getStatusCode().is2xxSuccessful());
RuleChain ruleChain = ruleChainResponse.getBody();
JsonNode configuration = mapper.readTree(this.getClass().getClassLoader().getResourceAsStream("RpcResponseRuleChainMetadata.json"));
RuleChainMetaData ruleChainMetaData = new RuleChainMetaData();
ruleChainMetaData.setRuleChainId(ruleChain.getId());
ruleChainMetaData.setFirstNodeIndex(configuration.get("firstNodeIndex").asInt());
ruleChainMetaData.setNodes(Arrays.asList(mapper.treeToValue(configuration.get("nodes"), RuleNode[].class)));
ruleChainMetaData.setConnections(Arrays.asList(mapper.treeToValue(configuration.get("connections"), NodeConnectionInfo[].class)));
ResponseEntity<RuleChainMetaData> ruleChainMetadataResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/ruleChain/metadata",
ruleChainMetaData,
RuleChainMetaData.class);
Assert.assertTrue(ruleChainMetadataResponse.getStatusCode().is2xxSuccessful());
// Set a new rule chain as root
ResponseEntity<RuleChain> rootRuleChainResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/ruleChain/{ruleChainId}/root",
null,
RuleChain.class,
ruleChain.getId());
Assert.assertTrue(rootRuleChainResponse.getStatusCode().is2xxSuccessful());
return ruleChain.getId();
}
private RuleChainId getDefaultRuleChainId() {
ResponseEntity<TextPageData<RuleChain>> ruleChains = restClient.getRestTemplate().exchange(
HTTPS_URL + "/api/ruleChains?limit=40&textSearch=",
HttpMethod.GET,
null,
new ParameterizedTypeReference<TextPageData<RuleChain>>() {
});
Optional<RuleChain> defaultRuleChain = ruleChains.getBody().getData()
.stream()
.filter(RuleChain::isRoot)
.findFirst();
if (!defaultRuleChain.isPresent()) {
Assert.fail("Root rule chain wasn't found");
}
return defaultRuleChain.get().getId();
}
private MqttClient getMqttClient(DeviceCredentials deviceCredentials, MqttMessageListener listener) throws InterruptedException {
MqttClientConfig clientConfig = new MqttClientConfig();
clientConfig.setClientId("MQTT client from test");
clientConfig.setUsername(deviceCredentials.getCredentialsId());
MqttClient mqttClient = MqttClient.create(clientConfig, listener);
mqttClient.connect("localhost", 1883).sync();
return mqttClient;
}
@Data
private class MqttMessageListener implements MqttHandler {
private final BlockingQueue<MqttEvent> events;
private MqttMessageListener() {
events = new ArrayBlockingQueue<>(100);
}
@Override
public void onMessage(String topic, ByteBuf message) {
log.info("MQTT message [{}], topic [{}]", message.toString(StandardCharsets.UTF_8), topic);
events.add(new MqttEvent(topic, message.toString(StandardCharsets.UTF_8)));
}
}
@Data
private class MqttEvent {
private final String topic;
private final String message;
}
}

26
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java
View File

@ -0,0 +1,26 @@
/**
* Copyright © 2016-2018 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.server.msa.mapper;
import lombok.Data;
import java.util.Map;
@Data
public class AttributesResponse {
private Map<String, Object> client;
private Map<String, Object> shared;
}

40
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java
View File

@ -0,0 +1,40 @@
/**
* Copyright © 2016-2018 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.server.msa.mapper;
import lombok.Data;
import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@Data
public class WsTelemetryResponse implements Serializable {
private int subscriptionId;
private int errorCode;
private String errorMsg;
private Map<String, List<List<Object>>> data;
private Map<String, Object> latestValues;
public List<Object> getDataValuesByKey(String key) {
return data.entrySet().stream()
.filter(e -> e.getKey().equals(key))
.flatMap(e -> e.getValue().stream().flatMap(Collection::stream))
.collect(Collectors.toList());
}
}

59
msa/black-box-tests/src/test/resources/RpcResponseRuleChainMetadata.json
View File

@ -0,0 +1,59 @@
{
"firstNodeIndex": 0,
"nodes": [
{
"additionalInfo": {
"layoutX": 325,
"layoutY": 150
},
"type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode",
"name": "msgTypeSwitch",
"debugMode": true,
"configuration": {
"version": 0
}
},
{
"additionalInfo": {
"layoutX": 60,
"layoutY": 300
},
"type": "org.thingsboard.rule.engine.transform.TbTransformMsgNode",
"name": "formResponse",
"debugMode": true,
"configuration": {
"jsScript": "if (msg.method == \"getResponse\") {\n return {msg: {\"response\": \"requestReceived\"}, metadata: metadata, msgType: msgType};\n}\n\nreturn {msg: msg, metadata: metadata, msgType: msgType};"
}
},
{
"additionalInfo": {
"layoutX": 450,
"layoutY": 300
},
"type": "org.thingsboard.rule.engine.rpc.TbSendRPCReplyNode",
"name": "rpcReply",
"debugMode": true,
"configuration": {
"requestIdMetaDataAttribute": "requestId"
}
}
],
"connections": [
{
"fromIndex": 0,
"toIndex": 1,
"type": "RPC Request from Device"
},
{
"fromIndex": 1,
"toIndex": 2,
"type": "Success"
},
{
"fromIndex": 1,
"toIndex": 2,
"type": "Failure"
}
],
"ruleChainConnections": null
}

3
msa/pom.xml
View File

@ -16,7 +16,7 @@
--> -->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<parent> <parent>
<groupId>org.thingsboard</groupId> <groupId>org.thingsboard</groupId>
@ -41,6 +41,7 @@
<module>web-ui</module> <module>web-ui</module>
<module>tb-node</module> <module>tb-node</module>
<module>transport</module> <module>transport</module>
<module>black-box-tests</module>
</modules> </modules>
<build> <build>

Write Preview
Loading…
Cancel
Save