Merge remote-tracking branch 'upstream/master' into feature/edge-json-converter-impl

application/src/test/java/org/thingsboard/server/controller/EdgeControllerTest.java

@@ -23,6 +23,8 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.AbstractMessage;
+import lombok.extern.slf4j.Slf4j;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -84,9 +86,10 @@ import org.thingsboard.server.gen.edge.v1.UserUpdateMsg;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.hamcrest.Matchers.containsString;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID;
@@ -97,6 +100,7 @@ import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID;
 })
 @ContextConfiguration(classes = {EdgeControllerTest.Config.class})
 @DaoSqlTest
+@Slf4j
 public class EdgeControllerTest extends AbstractControllerTest {
 
     public static final String EDGE_HOST = "localhost";
@@ -865,10 +869,7 @@ public class EdgeControllerTest extends AbstractControllerTest {
 
         Edge edge = doPost("/api/edge", constructEdge("Test Sync Edge", "test"), Edge.class);
 
-        // simulate edge activation
+        simulateEdgeActivation(edge);
-        ObjectNode attributes = JacksonUtil.newObjectNode();
-        attributes.put("active", true);
-        doPost("/api/plugins/telemetry/EDGE/" + edge.getId() + "/attributes/" + DataConstants.SERVER_SCOPE, attributes);
 
         doPost("/api/edge/" + edge.getId().getId().toString()
                 + "/device/" + savedDevice.getId().getId().toString(), Device.class);
@@ -880,7 +881,7 @@ public class EdgeControllerTest extends AbstractControllerTest {
 
         edgeImitator.expectMessageAmount(24);
         edgeImitator.connect();
-        assertThat(edgeImitator.waitForMessages()).as("await for messages on first connect").isTrue();
+        waitForMessages(edgeImitator);
 
         verifyFetchersMsgs(edgeImitator);
         // verify queue msgs
@@ -892,7 +893,7 @@ public class EdgeControllerTest extends AbstractControllerTest {
 
         edgeImitator.expectMessageAmount(20);
         doPost("/api/edge/sync/" + edge.getId());
-        assertThat(edgeImitator.waitForMessages()).as("await for messages after edge sync rest api call").isTrue();
+        waitForMessages(edgeImitator);
 
         verifyFetchersMsgs(edgeImitator);
         Assert.assertTrue(edgeImitator.getDownlinkMsgs().isEmpty());
@@ -911,6 +912,35 @@ public class EdgeControllerTest extends AbstractControllerTest {
                 .andExpect(status().isOk());
     }
 
+    private void simulateEdgeActivation(Edge edge) throws Exception {
+        ObjectNode attributes = JacksonUtil.newObjectNode();
+        attributes.put("active", true);
+        doPost("/api/plugins/telemetry/EDGE/" + edge.getId() + "/attributes/" + DataConstants.SERVER_SCOPE, attributes);
+        Awaitility.await()
+                .atMost(30, TimeUnit.SECONDS)
+                .until(() -> {
+                    List<Map<String, Object>> values = doGetAsyncTyped("/api/plugins/telemetry/EDGE/" + edge.getId() +
+                            "/values/attributes/SERVER_SCOPE", new TypeReference<>() {});
+                    Optional<Map<String, Object>> activeAttrOpt = values.stream().filter(att -> att.get("key").equals("active")).findFirst();
+                    if (activeAttrOpt.isEmpty()) {
+                        return false;
+                    }
+                    Map<String, Object> activeAttr = activeAttrOpt.get();
+                    return "true".equals(activeAttr.get("value").toString());
+                });
+    }
+
+    private void waitForMessages(EdgeImitator edgeImitator) throws Exception {
+        boolean success = edgeImitator.waitForMessages();
+        if (!success) {
+            List<AbstractMessage> downlinkMsgs = edgeImitator.getDownlinkMsgs();
+            for (AbstractMessage downlinkMsg : downlinkMsgs) {
+                log.error("{}\n{}", downlinkMsg.getClass(), downlinkMsg);
+            }
+            Assert.fail("Await for messages was not successful!");
+        }
+    }
+
     private void verifyFetchersMsgs(EdgeImitator edgeImitator) {
         Assert.assertTrue(popQueueMsg(edgeImitator.getDownlinkMsgs(), UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, "Main"));
         Assert.assertTrue(popRuleChainMsg(edgeImitator.getDownlinkMsgs(), UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, "Edge Root Rule Chain"));