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

Code review changes

pull/6933/head
Volodymyr Babak 4 years ago
parent
8fc5fa7526
commit
58313cd664
5 changed files with 38 additions and 34 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 7
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java
  2. 9
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/WsClient.java
  3. 2
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java
  4. 24
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java
  5. 30
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java

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

@ -70,7 +70,7 @@ public abstract class AbstractContainerTest {
protected static String TB_TOKEN;
protected static RestClient restClient;
protected static boolean slowQueue;
protected static long timeoutMultiplier = 1;
protected ObjectMapper mapper = new ObjectMapper();
protected JsonParser jsonParser = new JsonParser();
@ -80,8 +80,9 @@ public abstract class AbstractContainerTest {
restClient = new RestClient(HTTPS_URL);
restClient.getRestTemplate().setRequestFactory(getRequestFactoryForSelfSignedCert());
String QUEUE_TYPE = System.getProperty("blackBoxTests.queue", "kafka");
slowQueue = !"kafka".equals(QUEUE_TYPE);
if (!"kafka".equals(System.getProperty("blackBoxTests.queue", "kafka"))) {
timeoutMultiplier = 10;
}
}
@Rule

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

@ -36,8 +36,11 @@ public class WsClient extends WebSocketClient {
private CountDownLatch firstReply = new CountDownLatch(1);
private CountDownLatch latch = new CountDownLatch(1);
WsClient(URI serverUri) {
private final long timeoutMultiplier;
WsClient(URI serverUri, long timeoutMultiplier) {
super(serverUri);
this.timeoutMultiplier = timeoutMultiplier;
}
@Override
@ -74,7 +77,7 @@ public class WsClient extends WebSocketClient {
public WsTelemetryResponse getLastMessage() {
try {
boolean result = latch.await(120, TimeUnit.SECONDS);
boolean result = latch.await(10 * timeoutMultiplier, TimeUnit.SECONDS);
if (result) {
return this.message;
} else {
@ -89,7 +92,7 @@ public class WsClient extends WebSocketClient {
void waitForFirstReply() {
try {
boolean result = firstReply.await(120, TimeUnit.SECONDS);
boolean result = firstReply.await(10 * timeoutMultiplier, TimeUnit.SECONDS);
if (!result) {
log.error("Timeout, ws message wasn't received");
throw new RuntimeException("Timeout, ws message wasn't received");

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

@ -90,7 +90,7 @@ public class HttpClientTest extends AbstractContainerTest {
Assert.assertTrue(deviceClientsAttributes.getStatusCode().is2xxSuccessful());
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
@SuppressWarnings("deprecation")
Optional<JsonNode> allOptional = restClient.getAttributes(accessToken, null, null);

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

@ -178,14 +178,14 @@ public class MqttClientTest extends AbstractContainerTest {
mqttClient.on("v1/devices/me/attributes/response/+", listener, MqttQoS.AT_LEAST_ONCE).get();
// Wait until subscription is processed
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
// 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())).get();
MqttEvent event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
AttributesResponse attributes = mapper.readValue(Objects.requireNonNull(event).getMessage(), AttributesResponse.class);
log.info("Received telemetry: {}", attributes);
@ -209,7 +209,7 @@ public class MqttClientTest extends AbstractContainerTest {
mqttClient.on("v1/devices/me/attributes", listener, MqttQoS.AT_LEAST_ONCE).get();
// Wait until subscription is processed
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
String sharedAttributeName = "sharedAttr";
@ -223,7 +223,7 @@ public class MqttClientTest extends AbstractContainerTest {
device.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
MqttEvent event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertEquals(sharedAttributeValue,
mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText());
@ -237,7 +237,7 @@ public class MqttClientTest extends AbstractContainerTest {
device.getId());
Assert.assertTrue(updatedSharedAttributesResponse.getStatusCode().is2xxSuccessful());
event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertEquals(updatedSharedAttributeValue,
mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText());
@ -255,7 +255,7 @@ public class MqttClientTest extends AbstractContainerTest {
mqttClient.on("v1/devices/me/rpc/request/+", listener, MqttQoS.AT_LEAST_ONCE).get();
// Wait until subscription is processed
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
// Send an RPC from the server
JsonObject serverRpcPayload = new JsonObject();
@ -274,7 +274,7 @@ public class MqttClientTest extends AbstractContainerTest {
});
// Wait for RPC call from the server and send the response
MqttEvent requestFromServer = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent requestFromServer = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertEquals("{\"method\":\"getValue\",\"params\":true}", Objects.requireNonNull(requestFromServer).getMessage());
@ -284,7 +284,7 @@ public class MqttClientTest extends AbstractContainerTest {
// Send a response to the server's RPC request
mqttClient.publish("v1/devices/me/rpc/response/" + requestId, Unpooled.wrappedBuffer(clientResponse.toString().getBytes())).get();
ResponseEntity serverResponse = future.get(slowQueue ? 30 : 5, TimeUnit.SECONDS);
ResponseEntity serverResponse = future.get(5 * timeoutMultiplier, TimeUnit.SECONDS);
service.shutdownNow();
Assert.assertTrue(serverResponse.getStatusCode().is2xxSuccessful());
Assert.assertEquals(clientResponse.toString(), serverResponse.getBody());
@ -308,7 +308,7 @@ public class MqttClientTest extends AbstractContainerTest {
// Create a new root rule chain
RuleChainId ruleChainId = createRootRuleChainForRpcResponse();
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
// Send the request to the server
JsonObject clientRequest = new JsonObject();
clientRequest.addProperty("method", "getResponse");
@ -317,8 +317,8 @@ public class MqttClientTest extends AbstractContainerTest {
mqttClient.publish("v1/devices/me/rpc/request/" + requestId, Unpooled.wrappedBuffer(clientRequest.toString().getBytes())).get();
// Check the response from the server
TimeUnit.SECONDS.sleep(slowQueue ? 10 : 1);
MqttEvent responseFromServer = listener.getEvents().poll(slowQueue ? 10 : 1, TimeUnit.SECONDS);
TimeUnit.SECONDS.sleep(1 * timeoutMultiplier);
MqttEvent responseFromServer = listener.getEvents().poll(1 * timeoutMultiplier, 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());
@ -347,7 +347,7 @@ public class MqttClientTest extends AbstractContainerTest {
MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
restClient.deleteDevice(device.getId());
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
Assert.assertFalse(mqttClient.isConnected());
}

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

@ -168,7 +168,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
createdDevice.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
var event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
var event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
JsonObject requestData = new JsonObject();
requestData.addProperty("id", 1);
@ -178,7 +178,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get();
event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
JsonObject responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject();
Assert.assertTrue(responseData.has("value"));
@ -195,7 +195,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get();
event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject();
Assert.assertTrue(responseData.has("values"));
@ -213,7 +213,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get();
event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject();
Assert.assertTrue(responseData.has("values"));
@ -256,7 +256,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
createdDevice.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
MqttEvent sharedAttributeEvent = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent sharedAttributeEvent = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
// Catch attribute update event
Assert.assertNotNull(sharedAttributeEvent);
@ -266,7 +266,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
// Wait until subscription is processed
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
checkAttribute(true, clientAttributeValue);
checkAttribute(false, sharedAttributeValue);
@ -276,7 +276,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
public void subscribeToAttributeUpdatesFromServer() throws Exception {
mqttClient.on("v1/gateway/attributes", listener, MqttQoS.AT_LEAST_ONCE).get();
// Wait until subscription is processed
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
String sharedAttributeName = "sharedAttr";
// Add a new shared attribute
@ -294,7 +294,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
createdDevice.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
MqttEvent event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertEquals(sharedAttributeValue,
mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get("data").get(sharedAttributeName).asText());
@ -313,7 +313,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
createdDevice.getId());
Assert.assertTrue(updatedSharedAttributesResponse.getStatusCode().is2xxSuccessful());
event = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertEquals(updatedSharedAttributeValue,
mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get("data").get(sharedAttributeName).asText());
}
@ -324,7 +324,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
mqttClient.on(gatewayRpcTopic, listener, MqttQoS.AT_LEAST_ONCE).get();
// Wait until subscription is processed
TimeUnit.SECONDS.sleep(slowQueue ? 30 : 3);
TimeUnit.SECONDS.sleep(3 * timeoutMultiplier);
// Send an RPC from the server
JsonObject serverRpcPayload = new JsonObject();
@ -343,7 +343,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
});
// Wait for RPC call from the server and send the response
MqttEvent requestFromServer = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent requestFromServer = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
service.shutdownNow();
Assert.assertNotNull(requestFromServer);
@ -369,7 +369,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
// Send a response to the server's RPC request
mqttClient.publish(gatewayRpcTopic, Unpooled.wrappedBuffer(gatewayResponse.toString().getBytes())).get();
ResponseEntity serverResponse = future.get(slowQueue ? 30 : 5, TimeUnit.SECONDS);
ResponseEntity serverResponse = future.get(5 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertTrue(serverResponse.getStatusCode().is2xxSuccessful());
Assert.assertEquals(clientResponse.toString(), serverResponse.getBody());
}
@ -396,7 +396,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
gatewayAttributesRequest.addProperty("key", attributeName);
log.info(gatewayAttributesRequest.toString());
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(gatewayAttributesRequest.toString().getBytes())).get();
MqttEvent clientAttributeEvent = listener.getEvents().poll(slowQueue ? 60 : 10, TimeUnit.SECONDS);
MqttEvent clientAttributeEvent = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS);
Assert.assertNotNull(clientAttributeEvent);
JsonObject responseMessage = new JsonParser().parse(Objects.requireNonNull(clientAttributeEvent).getMessage()).getAsJsonObject();
@ -407,14 +407,14 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
}
private Device createDeviceThroughGateway(MqttClient mqttClient, Device gatewayDevice) throws Exception {
if (slowQueue) {
if (timeoutMultiplier > 1) {
TimeUnit.SECONDS.sleep(30);
}
String deviceName = "mqtt_device";
mqttClient.publish("v1/gateway/connect", Unpooled.wrappedBuffer(createGatewayConnectPayload(deviceName).toString().getBytes()), MqttQoS.AT_LEAST_ONCE).get();
if (slowQueue) {
if (timeoutMultiplier > 1) {
TimeUnit.SECONDS.sleep(30);
}

Write Preview
Loading…
Cancel
Save