subState = new HashMap<>(keys.size());
keys.forEach(key -> subState.put(key, startTs));
data.forEach(v -> subState.put(v.getKey(), v.getTs()));
+ Lock subLock = new ReentrantLock();
TbTimeseriesSubscription sub = TbTimeseriesSubscription.builder()
.serviceId(serviceId)
.sessionId(sessionId)
.subscriptionId(cmd.getCmdId())
.tenantId(sessionRef.getSecurityCtx().getTenantId())
.entityId(entityId)
- .updateConsumer(DefaultTelemetryWebSocketService.this::sendWsMsg)
+ .updateConsumer((sessionId, update) -> {
+ subLock.lock();
+ try {
+ sendWsMsg(sessionId, update);
+ } finally {
+ subLock.unlock();
+ }
+ })
.allKeys(false)
.keyStates(subState).build();
- oldSubService.addSubscription(sub);
+
+ subLock.lock();
+ try{
+ oldSubService.addSubscription(sub);
+ sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), data));
+ } finally {
+ subLock.unlock();
+ }
}
@Override
@@ -793,7 +852,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, int cmdId, Object update) {
try {
- String msg = jsonMapper.writeValueAsString(update);
+ String msg = JacksonUtil.OBJECT_MAPPER.writeValueAsString(update);
executor.submit(() -> {
try {
msgEndpoint.send(sessionRef, cmdId, msg);
diff --git a/application/src/main/java/org/thingsboard/server/utils/EventDeduplicationExecutor.java b/application/src/main/java/org/thingsboard/server/utils/EventDeduplicationExecutor.java
deleted file mode 100644
index 7ce958ef2d..0000000000
--- a/application/src/main/java/org/thingsboard/server/utils/EventDeduplicationExecutor.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/**
- * Copyright © 2016-2023 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.utils;
-
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.function.Consumer;
-
-/**
- * This class deduplicate executions of the specified function.
- * Useful in cluster mode, when you get event about partition change multiple times.
- * Assuming that the function execution is expensive, we should execute it immediately when first time event occurs and
- * later, once the processing of first event is done, process last pending task.
- *
- * @param parameters of the function
- */
-@Slf4j
-public class EventDeduplicationExecutor
{
- private final String name;
- private final ExecutorService executor;
- private final Consumer
function;
- private P pendingTask;
- private boolean busy;
-
- public EventDeduplicationExecutor(String name, ExecutorService executor, Consumer
function) {
- this.name = name;
- this.executor = executor;
- this.function = function;
- }
-
- public void submit(P params) {
- log.info("[{}] Going to submit: {}", name, params);
- synchronized (EventDeduplicationExecutor.this) {
- if (!busy) {
- busy = true;
- pendingTask = null;
- try {
- log.info("[{}] Submitting task: {}", name, params);
- executor.submit(() -> {
- try {
- log.info("[{}] Executing task: {}", name, params);
- function.accept(params);
- } catch (Throwable e) {
- log.warn("[{}] Failed to process task with parameters: {}", name, params, e);
- throw e;
- } finally {
- unlockAndProcessIfAny();
- }
- });
- } catch (Throwable e) {
- log.warn("[{}] Failed to submit task with parameters: {}", name, params, e);
- unlockAndProcessIfAny();
- throw e;
- }
- } else {
- log.info("[{}] Task is already in progress. {} pending task: {}", name, pendingTask == null ? "adding" : "updating", params);
- pendingTask = params;
- }
- }
- }
-
- private void unlockAndProcessIfAny() {
- synchronized (EventDeduplicationExecutor.this) {
- busy = false;
- if (pendingTask != null) {
- submit(pendingTask);
- }
- }
- }
-}
diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index 5549861950..10c4c9edac 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@@ -73,6 +73,8 @@ server:
min_timeout: "${MIN_SERVER_SIDE_RPC_TIMEOUT:5000}"
# Default value of the server side RPC timeout.
default_timeout: "${DEFAULT_SERVER_SIDE_RPC_TIMEOUT:10000}"
+ rate_limits:
+ reset_password_per_user: "${RESET_PASSWORD_PER_USER_RATE_LIMIT_CONFIGURATION:5:3600}"
# Application info
app:
@@ -571,6 +573,7 @@ spring:
password: "${SPRING_DATASOURCE_PASSWORD:postgres}"
hikari:
maximumPoolSize: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:16}"
+ registerMbeans: "${SPRING_DATASOURCE_HIKARI_REGISTER_MBEANS:false}" # true - enable MBean to diagnose pools state via JMX
# Audit log parameters
audit-log:
@@ -1209,3 +1212,4 @@ management:
exposure:
# Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
include: '${METRICS_ENDPOINTS_EXPOSE:info}'
+
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java
index 06cee7b71d..32d1f65765 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java
@@ -52,7 +52,7 @@ public abstract class AbstractControllerTest extends AbstractNotifyEntityTest {
@LocalServerPort
protected int wsPort;
- private TbTestWebSocketClient wsClient; // lazy
+ private volatile TbTestWebSocketClient wsClient; // lazy
public TbTestWebSocketClient getWsClient() {
if (wsClient == null) {
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
index 24ce820760..4afb12e638 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
@@ -29,14 +29,19 @@ import lombok.extern.slf4j.Slf4j;
import org.hamcrest.Matcher;
import org.hibernate.exception.ConstraintViolationException;
import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.rules.TestRule;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.converter.HttpMessageConverter;
@@ -52,6 +57,9 @@ import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilde
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.context.WebApplicationContext;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.MailService;
+import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.DeviceProfileType;
@@ -69,6 +77,7 @@ import org.thingsboard.server.common.data.device.profile.MqttTopics;
import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration;
import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration;
import org.thingsboard.server.common.data.edge.Edge;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.HasId;
@@ -83,7 +92,6 @@ import org.thingsboard.server.common.data.security.Authority;
import org.thingsboard.server.config.ThingsboardSecurityConfiguration;
import org.thingsboard.server.dao.Dao;
import org.thingsboard.server.dao.tenant.TenantProfileService;
-import org.thingsboard.server.service.mail.TestMailService;
import org.thingsboard.server.service.security.auth.jwt.RefreshTokenRequest;
import org.thingsboard.server.service.security.auth.rest.LoginRequest;
@@ -98,6 +106,7 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete;
@@ -144,6 +153,9 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
protected MockMvc mockMvc;
+ protected String currentActivateToken;
+ protected String currentResetPasswordToken;
+
protected String token;
protected String refreshToken;
protected String username;
@@ -168,6 +180,9 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
@Autowired
private TenantProfileService tenantProfileService;
+ @SpyBean
+ protected MailService mailService;
+
@Rule
public TestRule watcher = new TestWatcher() {
protected void starting(Description description) {
@@ -198,7 +213,9 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
@Before
public void setupWebTest() throws Exception {
- log.info("Executing web test setup");
+ log.debug("Executing web test setup");
+
+ setupMailServiceMock();
if (this.mockMvc == null) {
this.mockMvc = webAppContextSetup(webApplicationContext)
@@ -238,12 +255,33 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
resetTokens();
- log.info("Executed web test setup");
+ log.debug("Executed web test setup");
+ }
+
+ private void setupMailServiceMock() throws ThingsboardException {
+ Mockito.doNothing().when(mailService).sendAccountActivatedEmail(anyString(), anyString());
+ Mockito.doAnswer(new Answer() {
+ public Void answer(InvocationOnMock invocation) {
+ Object[] args = invocation.getArguments();
+ String activationLink = (String) args[0];
+ currentActivateToken = activationLink.split("=")[1];
+ return null;
+ }
+ }).when(mailService).sendActivationEmail(anyString(), anyString());
+
+ Mockito.doAnswer(new Answer() {
+ public Void answer(InvocationOnMock invocation) {
+ Object[] args = invocation.getArguments();
+ String passwordResetLink = (String) args[0];
+ currentResetPasswordToken = passwordResetLink.split("=")[1];
+ return null;
+ }
+ }).when(mailService).sendResetPasswordEmailAsync(anyString(), anyString());
}
@After
public void teardownWebTest() throws Exception {
- log.info("Executing web test teardown");
+ log.debug("Executing web test teardown");
loginSysAdmin();
doDelete("/api/tenant/" + tenantId.getId().toString())
@@ -368,11 +406,11 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
}
private JsonNode getActivateRequest(String password) throws Exception {
- doGet("/api/noauth/activate?activateToken={activateToken}", TestMailService.currentActivateToken)
+ doGet("/api/noauth/activate?activateToken={activateToken}", this.currentActivateToken)
.andExpect(status().isSeeOther())
- .andExpect(header().string(HttpHeaders.LOCATION, "/login/createPassword?activateToken=" + TestMailService.currentActivateToken));
+ .andExpect(header().string(HttpHeaders.LOCATION, "/login/createPassword?activateToken=" + this.currentActivateToken));
return new ObjectMapper().createObjectNode()
- .put("activateToken", TestMailService.currentActivateToken)
+ .put("activateToken", this.currentActivateToken)
.put("password", password);
}
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseAdminControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseAdminControllerTest.java
index 021171f1af..e6c27aa4a8 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseAdminControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseAdminControllerTest.java
@@ -21,12 +21,9 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.Test;
import org.mockito.Mockito;
-import org.springframework.beans.factory.annotation.Autowired;
import org.thingsboard.common.util.JacksonUtil;
-import org.thingsboard.rule.engine.api.MailService;
import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.security.model.JwtSettings;
-import org.thingsboard.server.service.mail.DefaultMailService;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
@@ -35,6 +32,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -43,12 +42,6 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
public abstract class BaseAdminControllerTest extends AbstractControllerTest {
final JwtSettings defaultJwtSettings = new JwtSettings(9000, 604800, "thingsboard.io", "thingsboardDefaultSigningKey");
- @Autowired
- MailService mailService;
-
- @Autowired
- DefaultMailService defaultMailService;
-
@Test
public void testFindAdminSettingsByKey() throws Exception {
loginSysAdmin();
@@ -118,10 +111,12 @@ public abstract class BaseAdminControllerTest extends AbstractControllerTest {
@Test
public void testSendTestMail() throws Exception {
+ Mockito.doNothing().when(mailService).sendTestMail(any(), anyString());
loginSysAdmin();
AdminSettings adminSettings = doGet("/api/admin/settings/mail", AdminSettings.class);
doPost("/api/admin/settings/testMail", adminSettings)
.andExpect(status().isOk());
+ Mockito.verify(mailService).sendTestMail(Mockito.any(), Mockito.anyString());
}
@Test
@@ -137,15 +132,8 @@ public abstract class BaseAdminControllerTest extends AbstractControllerTest {
adminSettings.setJsonValue(objectNode);
- Mockito.doAnswer((invocations) -> {
- var jsonConfig = (JsonNode) invocations.getArgument(0);
- var email = (String) invocations.getArgument(1);
-
- defaultMailService.sendTestMail(jsonConfig, email);
- return null;
- }).when(mailService).sendTestMail(Mockito.any(), Mockito.anyString());
doPost("/api/admin/settings/testMail", adminSettings).andExpect(status().is5xxServerError());
- Mockito.doNothing().when(mailService).sendTestMail(Mockito.any(), Mockito.any());
+ Mockito.verify(mailService).sendTestMail(Mockito.any(), Mockito.anyString());
}
void resetJwtSettingsToDefault() throws Exception {
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseUserControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseUserControllerTest.java
index f289c8066f..7149f52630 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseUserControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseUserControllerTest.java
@@ -44,7 +44,6 @@ import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.security.Authority;
import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.dao.user.UserDao;
-import org.thingsboard.server.service.mail.TestMailService;
import java.util.ArrayList;
import java.util.Collections;
@@ -54,6 +53,8 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.is;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -106,12 +107,12 @@ public abstract class BaseUserControllerTest extends AbstractControllerTest {
Mockito.reset(tbClusterService, auditLogService);
resetTokens();
- doGet("/api/noauth/activate?activateToken={activateToken}", TestMailService.currentActivateToken)
+ doGet("/api/noauth/activate?activateToken={activateToken}", this.currentActivateToken)
.andExpect(status().isSeeOther())
- .andExpect(header().string(HttpHeaders.LOCATION, "/login/createPassword?activateToken=" + TestMailService.currentActivateToken));
+ .andExpect(header().string(HttpHeaders.LOCATION, "/login/createPassword?activateToken=" + this.currentActivateToken));
JsonNode activateRequest = new ObjectMapper().createObjectNode()
- .put("activateToken", TestMailService.currentActivateToken)
+ .put("activateToken", this.currentActivateToken)
.put("password", "testPassword");
JsonNode tokenInfo = readResponse(doPost("/api/noauth/activate", activateRequest).andExpect(status().isOk()), JsonNode.class);
@@ -208,17 +209,19 @@ public abstract class BaseUserControllerTest extends AbstractControllerTest {
doPost("/api/noauth/resetPasswordByEmail", resetPasswordByEmailRequest)
.andExpect(status().isOk());
Thread.sleep(1000);
- doGet("/api/noauth/resetPassword?resetToken={resetToken}", TestMailService.currentResetPasswordToken)
+ doGet("/api/noauth/resetPassword?resetToken={resetToken}", this.currentResetPasswordToken)
.andExpect(status().isSeeOther())
- .andExpect(header().string(HttpHeaders.LOCATION, "/login/resetPassword?resetToken=" + TestMailService.currentResetPasswordToken));
+ .andExpect(header().string(HttpHeaders.LOCATION, "/login/resetPassword?resetToken=" + this.currentResetPasswordToken));
JsonNode resetPasswordRequest = new ObjectMapper().createObjectNode()
- .put("resetToken", TestMailService.currentResetPasswordToken)
+ .put("resetToken", this.currentResetPasswordToken)
.put("password", "testPassword2");
+ Mockito.doNothing().when(mailService).sendPasswordWasResetEmail(anyString(), anyString());
JsonNode tokenInfo = readResponse(
doPost("/api/noauth/resetPassword", resetPasswordRequest)
.andExpect(status().isOk()), JsonNode.class);
+ Mockito.verify(mailService).sendPasswordWasResetEmail(anyString(), anyString());
validateAndSetJwtToken(tokenInfo, email);
doGet("/api/auth/user")
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java
index dbadcb271b..5494fee4a2 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java
@@ -550,7 +550,7 @@ public abstract class BaseWebsocketApiTest extends AbstractControllerTest {
SingleEntityFilter entityFilter = new SingleEntityFilter();
entityFilter.setSingleEntity(tenantId);
- assertThatNoException().isThrownBy(() -> {
+ assertThatNoException().as("subscribeForAttributes").isThrownBy(() -> {
JsonNode update = getWsClient().subscribeForAttributes(tenantId, TbAttributeSubscriptionScope.SERVER_SCOPE.name(), List.of("attr"));
assertThat(update.get("errorMsg").isNull()).isTrue();
assertThat(update.get("errorCode").asInt()).isEqualTo(SubscriptionErrorCode.NO_ERROR.getCode());
@@ -562,7 +562,7 @@ public abstract class BaseWebsocketApiTest extends AbstractControllerTest {
new BaseAttributeKvEntry(System.currentTimeMillis(), new StringDataEntry("attr", expectedAttrValue))
));
JsonNode update = JacksonUtil.toJsonNode(getWsClient().waitForUpdate());
- assertThat(update).isNotNull();
+ assertThat(update).as("waitForUpdate").isNotNull();
assertThat(update.get("data").get("attr").get(0).get(1).asText()).isEqualTo(expectedAttrValue);
}
@@ -599,15 +599,17 @@ public abstract class BaseWebsocketApiTest extends AbstractControllerTest {
tsService.saveAndNotify(device.getTenantId(), null, device.getId(), tsData, 0, new FutureCallback() {
@Override
public void onSuccess(@Nullable Void result) {
+ log.debug("sendTelemetry callback onSuccess");
latch.countDown();
}
@Override
public void onFailure(Throwable t) {
+ log.error("Failed to send telemetry", t);
latch.countDown();
}
});
- latch.await(3, TimeUnit.SECONDS);
+ assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendTelemetry callback");
}
private void sendAttributes(Device device, TbAttributeSubscriptionScope scope, List attrData) throws InterruptedException {
@@ -619,14 +621,16 @@ public abstract class BaseWebsocketApiTest extends AbstractControllerTest {
tsService.saveAndNotify(tenantId, entityId, scope.name(), attrData, new FutureCallback() {
@Override
public void onSuccess(@Nullable Void result) {
+ log.debug("sendAttributes callback onSuccess");
latch.countDown();
}
@Override
public void onFailure(Throwable t) {
+ log.error("Failed to sendAttributes", t);
latch.countDown();
}
});
- latch.await(3, TimeUnit.SECONDS);
+ assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendAttributes callback").isTrue();
}
}
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseWidgetsBundleControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseWidgetsBundleControllerTest.java
index 56df8513fe..828d279e93 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseWidgetsBundleControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseWidgetsBundleControllerTest.java
@@ -192,7 +192,7 @@ public abstract class BaseWidgetsBundleControllerTest extends AbstractController
WidgetsBundle savedWidgetsBundle = doPost("/api/widgetsBundle", widgetsBundle, WidgetsBundle.class);
savedWidgetsBundle.setAlias("new_alias");
- Mockito.reset(tbClusterService);
+ Mockito.clearInvocations(tbClusterService);
doPost("/api/widgetsBundle", savedWidgetsBundle)
.andExpect(status().isBadRequest())
diff --git a/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java b/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java
index db26e9f6df..1f9ac4ea69 100644
--- a/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java
+++ b/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java
@@ -47,6 +47,7 @@ import java.util.concurrent.TimeUnit;
@Slf4j
public class TbTestWebSocketClient extends WebSocketClient {
+ private static final long TIMEOUT = TimeUnit.SECONDS.toMillis(30);
private volatile String lastMsg;
private volatile CountDownLatch reply;
private volatile CountDownLatch update;
@@ -87,12 +88,14 @@ public class TbTestWebSocketClient extends WebSocketClient {
}
public void registerWaitForUpdate(int count) {
+ log.debug("registerWaitForUpdate [{}]", count);
lastMsg = null;
update = new CountDownLatch(count);
}
@Override
public void send(String text) throws NotYetConnectedException {
+ log.debug("send [{}]", text);
reply = new CountDownLatch(1);
super.send(text);
}
@@ -110,21 +113,31 @@ public class TbTestWebSocketClient extends WebSocketClient {
}
public String waitForUpdate() {
- return waitForUpdate(TimeUnit.SECONDS.toMillis(3));
+ return waitForUpdate(TIMEOUT);
}
public String waitForUpdate(long ms) {
+ log.debug("waitForUpdate [{}]", ms);
try {
- update.await(ms, TimeUnit.MILLISECONDS);
+ if (!update.await(ms, TimeUnit.MILLISECONDS)) {
+ log.warn("Failed to await update (waiting time [{}]ms elapsed)", ms, new RuntimeException("stacktrace"));
+ }
} catch (InterruptedException e) {
- log.warn("Failed to await reply", e);
+ log.warn("Failed to await update", e);
}
return lastMsg;
}
public String waitForReply() {
+ return waitForReply(TIMEOUT);
+ }
+
+ public String waitForReply(long ms) {
+ log.debug("waitForReply [{}]", ms);
try {
- reply.await(3, TimeUnit.SECONDS);
+ if (!reply.await(ms, TimeUnit.MILLISECONDS)) {
+ log.warn("Failed to await reply (waiting time [{}]ms elapsed)", ms, new RuntimeException("stacktrace"));
+ }
} catch (InterruptedException e) {
log.warn("Failed to await reply", e);
}
diff --git a/application/src/test/java/org/thingsboard/server/service/mail/TestMailService.java b/application/src/test/java/org/thingsboard/server/service/mail/TestMailService.java
deleted file mode 100644
index 7c4c45c951..0000000000
--- a/application/src/test/java/org/thingsboard/server/service/mail/TestMailService.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/**
- * Copyright © 2016-2023 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.service.mail;
-
-import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-import org.springframework.context.annotation.Profile;
-import org.thingsboard.rule.engine.api.MailService;
-import org.thingsboard.server.common.data.exception.ThingsboardException;
-
-@Profile("test")
-@Configuration
-public class TestMailService {
-
- public static String currentActivateToken;
- public static String currentResetPasswordToken;
-
- @Bean
- @Primary
- public MailService mailService() throws ThingsboardException {
- MailService mailService = Mockito.mock(MailService.class);
- Mockito.doAnswer(new Answer() {
- public Void answer(InvocationOnMock invocation) {
- Object[] args = invocation.getArguments();
- String activationLink = (String) args[0];
- currentActivateToken = activationLink.split("=")[1];
- return null;
- }
- }).when(mailService).sendActivationEmail(Mockito.anyString(), Mockito.anyString());
- Mockito.doAnswer(new Answer() {
- public Void answer(InvocationOnMock invocation) {
- Object[] args = invocation.getArguments();
- String passwordResetLink = (String) args[0];
- currentResetPasswordToken = passwordResetLink.split("=")[1];
- return null;
- }
- }).when(mailService).sendResetPasswordEmailAsync(Mockito.anyString(), Mockito.anyString());
- return mailService;
- }
-
-}
diff --git a/application/src/test/java/org/thingsboard/server/service/script/NashornJsInvokeServiceTest.java b/application/src/test/java/org/thingsboard/server/service/script/NashornJsInvokeServiceTest.java
index 08cd031a52..4373a4c108 100644
--- a/application/src/test/java/org/thingsboard/server/service/script/NashornJsInvokeServiceTest.java
+++ b/application/src/test/java/org/thingsboard/server/service/script/NashornJsInvokeServiceTest.java
@@ -36,6 +36,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
@DaoSqlTest
@TestPropertySource(properties = {
+ "js.evaluator=local",
"js.max_script_body_size=50",
"js.max_total_args_size=50",
"js.max_result_size=50",
diff --git a/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java
index cf4de21cc2..89a4b707bf 100644
--- a/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java
@@ -28,7 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
public abstract class AbstractTransportIntegrationTest extends AbstractControllerTest {
- protected static final int DEFAULT_WAIT_TIMEOUT_SECONDS = 10;
+ protected static final int DEFAULT_WAIT_TIMEOUT_SECONDS = 30;
protected static final String MQTT_URL = "tcp://localhost:1883";
protected static final String COAP_BASE_URL = "coap://localhost:5683/api/v1/";
diff --git a/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java
index 41cc4c2b76..ab2f6b9bca 100644
--- a/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java
+++ b/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java
@@ -15,30 +15,14 @@
*/
package org.thingsboard.server.transport;
-import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
import org.junit.extensions.cpsuite.ClasspathSuite;
import org.junit.runner.RunWith;
-import org.thingsboard.server.dao.CustomCassandraCQLUnit;
-import org.thingsboard.server.queue.memory.InMemoryStorage;
-
-import java.util.Arrays;
+import org.thingsboard.server.dao.AbstractNoSqlContainer;
@RunWith(ClasspathSuite.class)
@ClasspathSuite.ClassnameFilters({
"org.thingsboard.server.transport.*.telemetry.timeseries.nosql.*Test",
})
-public class TransportNoSqlTestSuite {
-
- @ClassRule
- public static CustomCassandraCQLUnit cassandraUnit =
- new CustomCassandraCQLUnit(
- Arrays.asList(
- new ClassPathCQLDataSet("cassandra/schema-keyspace.cql", false, false),
- new ClassPathCQLDataSet("cassandra/schema-ts.cql", false, false),
- new ClassPathCQLDataSet("cassandra/schema-ts-latest.cql", false, false)
- ),
- "cassandra-test.yaml", 30000l);
+public class TransportNoSqlTestSuite extends AbstractNoSqlContainer {
}
diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java
index d90af7b6ba..1266ef2c29 100644
--- a/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java
@@ -45,6 +45,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@TestPropertySource(properties = {
+ "coap.enabled=true",
+ "service.integrations.supported=ALL",
"transport.coap.enabled=true",
})
@Slf4j
diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java
index 0f1815b46b..6cb6e49e79 100644
--- a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java
@@ -46,8 +46,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@TestPropertySource(properties = {
+ "service.integrations.supported=ALL",
"transport.mqtt.enabled=true",
- "js.evaluator=mock",
})
@Slf4j
public abstract class AbstractMqttIntegrationTest extends AbstractTransportIntegrationTest {
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 60af3b66f7..c37f22de36 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
@@ -15,6 +15,7 @@
*/
package org.thingsboard.server.transport.mqtt.mqttv3.attributes;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.github.os72.protobuf.dynamic.DynamicSchema;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
@@ -22,6 +23,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
import com.squareup.wire.schema.internal.parser.ProtoFileElement;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
+import org.awaitility.Awaitility;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DynamicProtoUtils;
@@ -45,6 +47,7 @@ import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -125,14 +128,16 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
client.subscribeAndWait(attrSubTopic, MqttQoS.AT_MOST_ONCE);
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
- onUpdateCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(onUpdateCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await onUpdateCallback").isTrue();
validateUpdateAttributesJsonResponse(onUpdateCallback, SHARED_ATTRIBUTES_PAYLOAD);
MqttTestCallback onDeleteCallback = new MqttTestCallback();
client.setCallback(onDeleteCallback);
doDelete("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/SHARED_SCOPE?keys=sharedJson", String.class);
- onDeleteCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(onDeleteCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await onDeleteCallback").isTrue();
validateUpdateAttributesJsonResponse(onDeleteCallback, SHARED_ATTRIBUTES_DELETED_RESPONSE);
client.disconnect();
}
@@ -145,13 +150,15 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
client.subscribeAndWait(attrSubTopic, MqttQoS.AT_MOST_ONCE);
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
- onUpdateCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(onUpdateCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await onUpdateCallback").isTrue();
validateUpdateAttributesProtoResponse(onUpdateCallback);
MqttTestCallback onDeleteCallback = new MqttTestCallback();
client.setCallback(onDeleteCallback);
doDelete("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/SHARED_SCOPE?keys=sharedJson", String.class);
- onDeleteCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(onDeleteCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await onDeleteCallback").isTrue();
validateDeleteAttributesProtoResponse(onDeleteCallback);
client.disconnect();
}
@@ -162,7 +169,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
}
protected void validateUpdateAttributesProtoResponse(MqttTestCallback callback) throws InvalidProtocolBufferException {
- assertNotNull(callback.getPayloadBytes());
+ assertThat(callback.getPayloadBytes()).as("callback payload non-null").isNotNull();
TransportProtos.AttributeUpdateNotificationMsg.Builder attributeUpdateNotificationMsgBuilder = TransportProtos.AttributeUpdateNotificationMsg.newBuilder();
List tsKvProtoList = getTsKvProtoList("shared");
attributeUpdateNotificationMsgBuilder.addAllSharedUpdated(tsKvProtoList);
@@ -178,7 +185,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
}
protected void validateDeleteAttributesProtoResponse(MqttTestCallback callback) throws InvalidProtocolBufferException {
- assertNotNull(callback.getPayloadBytes());
+ assertThat(callback.getPayloadBytes()).as("callback payload non-null").isNotNull();
TransportProtos.AttributeUpdateNotificationMsg.Builder attributeUpdateNotificationMsgBuilder = TransportProtos.AttributeUpdateNotificationMsg.newBuilder();
attributeUpdateNotificationMsgBuilder.addSharedDeleted("sharedJson");
@@ -209,7 +216,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
client.subscribeAndWait(GATEWAY_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE);
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
- onUpdateCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(onUpdateCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await onUpdateCallback").isTrue();
validateJsonGatewayUpdateAttributesResponse(onUpdateCallback, deviceName, SHARED_ATTRIBUTES_PAYLOAD);
@@ -217,7 +225,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
client.setCallback(onDeleteCallback);
doDelete("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/SHARED_SCOPE?keys=sharedJson", String.class);
- onDeleteCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(onDeleteCallback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await onDeleteCallback").isTrue();
validateJsonGatewayUpdateAttributesResponse(onDeleteCallback, deviceName, SHARED_ATTRIBUTES_DELETED_RESPONSE);
client.disconnect();
@@ -246,7 +255,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
}
protected void validateJsonGatewayUpdateAttributesResponse(MqttTestCallback callback, String deviceName, String expectResultData) {
- assertNotNull(callback.getPayloadBytes());
+ assertThat(callback.getPayloadBytes()).as("callback payload non-null").isNotNull();
assertEquals(JacksonUtil.toJsonNode(getGatewayAttributesResponseJson(deviceName, expectResultData)), JacksonUtil.fromBytes(callback.getPayloadBytes()));
}
@@ -260,8 +269,9 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
}
protected void validateProtoGatewayUpdateAttributesResponse(MqttTestCallback callback, String deviceName) throws InvalidProtocolBufferException, InterruptedException {
- callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- assertNotNull(callback.getPayloadBytes());
+ assertThat(callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await callback").isTrue();
+ assertThat(callback.getPayloadBytes()).as("callback payload non-null").isNotNull();
TransportProtos.AttributeUpdateNotificationMsg.Builder attributeUpdateNotificationMsgBuilder = TransportProtos.AttributeUpdateNotificationMsg.newBuilder();
List tsKvProtoList = getTsKvProtoList("shared");
@@ -285,8 +295,9 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
}
protected void validateProtoGatewayDeleteAttributesResponse(MqttTestCallback callback, String deviceName) throws InvalidProtocolBufferException, InterruptedException {
- callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- assertNotNull(callback.getPayloadBytes());
+ assertThat(callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .as("await callback").isTrue();
+ assertThat(callback.getPayloadBytes()).as("callback payload non-null").isNotNull();
TransportProtos.AttributeUpdateNotificationMsg.Builder attributeUpdateNotificationMsgBuilder = TransportProtos.AttributeUpdateNotificationMsg.newBuilder();
attributeUpdateNotificationMsgBuilder.addSharedDeleted("sharedJson");
TransportProtos.AttributeUpdateNotificationMsg attributeUpdateNotificationMsg = attributeUpdateNotificationMsgBuilder.build();
@@ -391,9 +402,19 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
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)
+ .until(() -> {
+ List