Igor Kulikov
4 months ago
83 changed files with 2797 additions and 481 deletions
@ -0,0 +1,299 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.edge.rpc; |
|||
|
|||
import io.grpc.stub.StreamObserver; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.gen.edge.v1.ResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; |
|||
import org.thingsboard.server.queue.discovery.TopicService; |
|||
import org.thingsboard.server.queue.kafka.KafkaAdmin; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Queue; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentLinkedQueue; |
|||
import java.util.concurrent.CopyOnWriteArrayList; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.BooleanSupplier; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
class KafkaEdgeGrpcSessionTest { |
|||
|
|||
private static final long POLL_INTERVAL_MS = 20L; |
|||
|
|||
private EdgeContextComponent ctx; |
|||
private TbCoreQueueFactory tbCoreQueueFactory; |
|||
private KafkaEdgeGrpcSession session; |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
ctx = mock(EdgeContextComponent.class); |
|||
TopicService topicService = mock(TopicService.class); |
|||
tbCoreQueueFactory = mock(TbCoreQueueFactory.class); |
|||
KafkaAdmin kafkaAdmin = mock(KafkaAdmin.class); |
|||
@SuppressWarnings("unchecked") |
|||
StreamObserver<ResponseMsg> outputStream = mock(StreamObserver.class); |
|||
|
|||
session = new KafkaEdgeGrpcSession(ctx, topicService, tbCoreQueueFactory, kafkaAdmin, outputStream, |
|||
(edgeId, s) -> {}, (edge, uuid) -> {}, null, 0, 0); |
|||
|
|||
ReflectionTestUtils.setField(session, "edge", new Edge(new EdgeId(UUID.randomUUID()))); |
|||
ReflectionTestUtils.setField(session, "tenantId", TenantId.fromUUID(UUID.randomUUID())); |
|||
} |
|||
|
|||
@AfterEach |
|||
void tearDown() { |
|||
if (session != null) { |
|||
session.destroy(); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
void readyOnlyWhenConnectedNotSyncingNotHighPriority() { |
|||
setReadinessFlags(true, false, false); |
|||
assertThat(isReadyToProcessGeneralEvents()) |
|||
.as("connected, not syncing, no high-priority work -> ready") |
|||
.isTrue(); |
|||
} |
|||
|
|||
@Test |
|||
void notReadyWhenDisconnected() { |
|||
setReadinessFlags(false, false, false); |
|||
assertThat(isReadyToProcessGeneralEvents()) |
|||
.as("disconnected -> not ready") |
|||
.isFalse(); |
|||
} |
|||
|
|||
@Test |
|||
void notReadyWhileSyncInProgress() { |
|||
setReadinessFlags(true, true, false); |
|||
assertThat(isReadyToProcessGeneralEvents()) |
|||
.as("sync in progress -> not ready (this is the window where events were being dropped)") |
|||
.isFalse(); |
|||
} |
|||
|
|||
@Test |
|||
void notReadyWhileHighPriorityProcessing() { |
|||
setReadinessFlags(true, false, true); |
|||
assertThat(isReadyToProcessGeneralEvents()) |
|||
.as("high-priority processing -> not ready") |
|||
.isFalse(); |
|||
} |
|||
|
|||
@Test |
|||
void processEdgeEventsWiresReadinessPredicateIntoConsumerGate() { |
|||
// processEdgeEvents() builds the consumer lazily; stub just enough of that path.
|
|||
EdgeEventStorageSettings storageSettings = new EdgeEventStorageSettings(); |
|||
storageSettings.setNoRecordsSleepInterval(1000L); |
|||
when(ctx.getEdgeEventStorageSettings()).thenReturn(storageSettings); |
|||
|
|||
@SuppressWarnings("unchecked") |
|||
TbQueueConsumer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> queueConsumer = mock(TbQueueConsumer.class); |
|||
// Report stopped so the launched consumer loop exits immediately - this test asserts on wiring, not polling.
|
|||
when(queueConsumer.isStopped()).thenReturn(true); |
|||
when(tbCoreQueueFactory.createEdgeEventMsgConsumer(any(), any())).thenReturn(queueConsumer); |
|||
|
|||
// The consumer is only started when the session is ready.
|
|||
setReadinessFlags(true, false, false); |
|||
session.processEdgeEvents(); |
|||
|
|||
QueueConsumerManager<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> manager = session.getConsumer(); |
|||
assertThat(manager).as("processEdgeEvents must build the consumer when ready").isNotNull(); |
|||
|
|||
BooleanSupplier readinessCheck = (BooleanSupplier) ReflectionTestUtils.getField(manager, "readinessCheck"); |
|||
assertThat(readinessCheck) |
|||
.as("the edge consumer must be wired with a readinessCheck (the .readinessCheck(...) builder line)") |
|||
.isNotNull(); |
|||
|
|||
// It must be the live predicate, not a snapshot: flipping the session's state must flip the gate.
|
|||
assertThat(readinessCheck.getAsBoolean()).as("ready session -> gate open").isTrue(); |
|||
setReadinessFlags(true, true, false); |
|||
assertThat(readinessCheck.getAsBoolean()).as("sync starts -> gate closes, consumer pauses polling").isFalse(); |
|||
} |
|||
|
|||
@Test |
|||
void eventArrivingDuringSyncIsHeldByTheEdgeConsumerUntilSyncCompletes() { |
|||
EdgeEventStorageSettings storageSettings = new EdgeEventStorageSettings(); |
|||
storageSettings.setNoRecordsSleepInterval(POLL_INTERVAL_MS); |
|||
when(ctx.getEdgeEventStorageSettings()).thenReturn(storageSettings); |
|||
|
|||
RecordingEdgeEventConsumer queueConsumer = new RecordingEdgeEventConsumer(); |
|||
when(tbCoreQueueFactory.createEdgeEventMsgConsumer(any(), any())).thenReturn(queueConsumer); |
|||
|
|||
// The consumer is launched only while the session is ready - that is how it starts in production.
|
|||
setReadinessFlags(true, false, false); |
|||
session.processEdgeEvents(); |
|||
|
|||
// Sync starts: the gate closes. Wait until the loop has actually parked on it (poll count stops advancing)
|
|||
// before enqueuing - otherwise we would race an in-flight poll() and the test would be non-deterministic.
|
|||
setReadinessFlags(true, true, false); |
|||
awaitParkedOnClosedGate(queueConsumer); |
|||
|
|||
// An event lands in the edge-event topic during the sync window - exactly the case that used to be dropped.
|
|||
@SuppressWarnings("unchecked") |
|||
TbProtoQueueMsg<ToEdgeEventNotificationMsg> event = mock(TbProtoQueueMsg.class); |
|||
int pollsBeforeEvent = queueConsumer.getPollCount(); |
|||
queueConsumer.enqueue(List.of(event)); |
|||
|
|||
// While sync is in progress the consumer stays parked: it neither polls nor consumes the event,
|
|||
// so the event survives in the queue instead of being read-and-skipped.
|
|||
sleepQuietly(POLL_INTERVAL_MS * 5); |
|||
assertThat(queueConsumer.getPolledEvents()) |
|||
.as("event must not be polled while sync is in progress (it must stay queued, not be dropped)") |
|||
.isEmpty(); |
|||
assertThat(queueConsumer.getPollCount()) |
|||
.as("consumer must not poll at all while the gate is closed") |
|||
.isEqualTo(pollsBeforeEvent); |
|||
|
|||
// Sync completes: the gate opens and the held event is finally picked up by the consumer.
|
|||
// (We assert at the poll boundary - the actual drop point - since the downlink-send path that
|
|||
// processMsgs drives afterwards is not reachable from a unit test.)
|
|||
setReadinessFlags(true, false, false); |
|||
await().atMost(5, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> assertThat(queueConsumer.getPolledEvents()) |
|||
.as("event held during sync must be picked up once sync completes, not lost") |
|||
.hasSize(1)); |
|||
} |
|||
|
|||
private boolean isReadyToProcessGeneralEvents() { |
|||
return Boolean.TRUE.equals(ReflectionTestUtils.invokeMethod(session, "isReadyToProcessGeneralEvents")); |
|||
} |
|||
|
|||
private void setReadinessFlags(boolean connected, boolean syncInProgress, boolean highPriorityProcessing) { |
|||
ReflectionTestUtils.setField(session, "connected", connected); |
|||
ReflectionTestUtils.setField(session, "syncInProgress", syncInProgress); |
|||
ReflectionTestUtils.setField(session, "isHighPriorityProcessing", highPriorityProcessing); |
|||
} |
|||
|
|||
private static void awaitParkedOnClosedGate(RecordingEdgeEventConsumer consumer) { |
|||
await().atMost(5, TimeUnit.SECONDS).until(() -> { |
|||
int before = consumer.getPollCount(); |
|||
// Several poll intervals with no new poll means the loop is parked on the readiness gate rather than
|
|||
// blocked inside poll(), so it is safe to enqueue without racing an in-flight read.
|
|||
sleepQuietly(POLL_INTERVAL_MS * 5); |
|||
return consumer.getPollCount() == before; |
|||
}); |
|||
} |
|||
|
|||
private static void sleepQuietly(long millis) { |
|||
try { |
|||
Thread.sleep(millis); |
|||
} catch (InterruptedException e) { |
|||
Thread.currentThread().interrupt(); |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* Fake edge-event consumer that records what it was actually polled, so a test can prove an event stayed in the |
|||
* queue while the readiness gate was closed and was only read once it opened. |
|||
*/ |
|||
private static class RecordingEdgeEventConsumer implements TbQueueConsumer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> { |
|||
|
|||
private final Queue<List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>>> pending = new ConcurrentLinkedQueue<>(); |
|||
private final List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> polledEvents = new CopyOnWriteArrayList<>(); |
|||
private final AtomicInteger pollCount = new AtomicInteger(); |
|||
private volatile boolean stopped; |
|||
|
|||
void enqueue(List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> batch) { |
|||
pending.add(batch); |
|||
} |
|||
|
|||
int getPollCount() { |
|||
return pollCount.get(); |
|||
} |
|||
|
|||
List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> getPolledEvents() { |
|||
return polledEvents; |
|||
} |
|||
|
|||
@Override |
|||
public List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> poll(long durationInMillis) { |
|||
pollCount.incrementAndGet(); |
|||
List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> batch = pending.poll(); |
|||
if (batch != null) { |
|||
polledEvents.addAll(batch); |
|||
return batch; |
|||
} |
|||
sleepQuietly(durationInMillis); |
|||
return Collections.emptyList(); |
|||
} |
|||
|
|||
@Override |
|||
public String getTopic() { |
|||
return "test-edge-event-topic"; |
|||
} |
|||
|
|||
@Override |
|||
public void subscribe() { |
|||
} |
|||
|
|||
@Override |
|||
public void subscribe(Set<TopicPartitionInfo> partitions) { |
|||
} |
|||
|
|||
@Override |
|||
public void stop() { |
|||
stopped = true; |
|||
} |
|||
|
|||
@Override |
|||
public void unsubscribe() { |
|||
stopped = true; |
|||
} |
|||
|
|||
@Override |
|||
public void commit() { |
|||
} |
|||
|
|||
@Override |
|||
public boolean isStopped() { |
|||
return stopped; |
|||
} |
|||
|
|||
@Override |
|||
public List<String> getFullTopicNames() { |
|||
return Collections.emptyList(); |
|||
} |
|||
|
|||
@Override |
|||
public Set<TopicPartitionInfo> getPartitions() { |
|||
return Collections.emptySet(); |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,141 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.entitiy.queue; |
|||
|
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.cluster.TbClusterService; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.dao.queue.QueueService; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.discovery.TopicService; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.anyBoolean; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.Mockito.never; |
|||
import static org.mockito.Mockito.times; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class DefaultTbQueueServiceTest { |
|||
|
|||
@Mock |
|||
private QueueService queueServiceMock; |
|||
@Mock |
|||
private TbClusterService tbClusterServiceMock; |
|||
@Mock |
|||
private TbQueueAdmin tbQueueAdminMock; |
|||
|
|||
private TopicService topicService; |
|||
private DefaultTbQueueService tbQueueService; |
|||
|
|||
private final TenantId tenantId = TenantId.SYS_TENANT_ID; |
|||
|
|||
@BeforeEach |
|||
public void setUp() { |
|||
topicService = new TopicService(); |
|||
tbQueueService = new DefaultTbQueueService(queueServiceMock, tbClusterServiceMock, tbQueueAdminMock, topicService); |
|||
} |
|||
|
|||
private Queue newQueue(int partitions) { |
|||
Queue queue = new Queue(); |
|||
queue.setTenantId(tenantId); |
|||
queue.setName("testQueue"); |
|||
queue.setTopic("tb_rule_engine.testQueue"); |
|||
queue.setPartitions(partitions); |
|||
return queue; |
|||
} |
|||
|
|||
@Test |
|||
public void givenQueuePrefix_whenSaveQueue_thenCreatesPrefixedTopics() { |
|||
// queue.prefix = "thingsboard" (TB_QUEUE_PREFIX set)
|
|||
ReflectionTestUtils.setField(topicService, "prefix", "thingsboard"); |
|||
|
|||
Queue queue = newQueue(2); |
|||
when(queueServiceMock.saveQueue(queue)).thenReturn(queue); |
|||
|
|||
tbQueueService.saveQueue(queue); |
|||
|
|||
ArgumentCaptor<String> topicCaptor = ArgumentCaptor.forClass(String.class); |
|||
verify(tbQueueAdminMock, times(2)).createTopicIfNotExists(topicCaptor.capture(), any(), anyBoolean()); |
|||
|
|||
// All created topics must carry the prefix - this is the fix.
|
|||
assertThat(topicCaptor.getAllValues()) |
|||
.containsExactlyInAnyOrder( |
|||
"thingsboard.tb_rule_engine.testQueue.0", |
|||
"thingsboard.tb_rule_engine.testQueue.1"); |
|||
// No unprefixed (orphan-prone) topic must ever be created.
|
|||
assertThat(topicCaptor.getAllValues()) |
|||
.noneMatch(topic -> topic.equals("tb_rule_engine.testQueue.0") |
|||
|| topic.equals("tb_rule_engine.testQueue.1")); |
|||
} |
|||
|
|||
@Test |
|||
public void givenNoQueuePrefix_whenSaveQueue_thenCreatesUnprefixedTopics() { |
|||
// queue.prefix blank (TB_QUEUE_PREFIX not set) - default behavior preserved
|
|||
ReflectionTestUtils.setField(topicService, "prefix", ""); |
|||
|
|||
Queue queue = newQueue(2); |
|||
when(queueServiceMock.saveQueue(queue)).thenReturn(queue); |
|||
|
|||
tbQueueService.saveQueue(queue); |
|||
|
|||
ArgumentCaptor<String> topicCaptor = ArgumentCaptor.forClass(String.class); |
|||
verify(tbQueueAdminMock, times(2)).createTopicIfNotExists(topicCaptor.capture(), any(), anyBoolean()); |
|||
|
|||
assertThat(topicCaptor.getAllValues()) |
|||
.containsExactlyInAnyOrder( |
|||
"tb_rule_engine.testQueue.0", |
|||
"tb_rule_engine.testQueue.1"); |
|||
} |
|||
|
|||
@Test |
|||
public void givenQueuePrefix_whenIncreasePartitions_thenOnlyNewPartitionsCreatedPrefixed() { |
|||
ReflectionTestUtils.setField(topicService, "prefix", "thingsboard"); |
|||
|
|||
Queue oldQueue = newQueue(2); |
|||
oldQueue.setId(new QueueId(UUID.randomUUID())); |
|||
Queue updatedQueue = newQueue(4); |
|||
updatedQueue.setId(oldQueue.getId()); |
|||
|
|||
when(queueServiceMock.findQueueById(tenantId, updatedQueue.getId())).thenReturn(oldQueue); |
|||
when(queueServiceMock.saveQueue(updatedQueue)).thenReturn(updatedQueue); |
|||
|
|||
tbQueueService.saveQueue(updatedQueue); |
|||
|
|||
ArgumentCaptor<String> topicCaptor = ArgumentCaptor.forClass(String.class); |
|||
verify(tbQueueAdminMock, times(2)).createTopicIfNotExists(topicCaptor.capture(), any(), anyBoolean()); |
|||
|
|||
assertThat(topicCaptor.getAllValues()) |
|||
.containsExactlyInAnyOrder( |
|||
"thingsboard.tb_rule_engine.testQueue.2", |
|||
"thingsboard.tb_rule_engine.testQueue.3"); |
|||
verify(tbQueueAdminMock, never()).createTopicIfNotExists(eq("thingsboard.tb_rule_engine.testQueue.0"), any(), anyBoolean()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,78 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.queue; |
|||
|
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.common.data.rpc.RpcError; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.BDDMockito.then; |
|||
import static org.mockito.Mockito.doCallRealMethod; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class DefaultTbRuleEngineConsumerServiceTest { |
|||
|
|||
@Mock |
|||
private TbRuleEngineDeviceRpcService tbDeviceRpcServiceMock; |
|||
@Mock |
|||
private TbCallback tbCallbackMock; |
|||
|
|||
@Mock |
|||
private DefaultTbRuleEngineConsumerService defaultTbRuleEngineConsumerServiceMock; |
|||
|
|||
@Test |
|||
public void givenNotFoundErrorAndNoResponse_whenHandleFromDeviceRpcResponse_thenNotFoundAndNullResponseAreRecovered() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbRuleEngineConsumerServiceMock, "tbDeviceRpcService", tbDeviceRpcServiceMock); |
|||
var requestId = UUID.randomUUID(); |
|||
// error = NOT_FOUND.ordinal() (0) and response left unset: the previously broken combination
|
|||
// ('error > 0' dropped NOT_FOUND, proto3 default collapsed a null response to "").
|
|||
var proto = TransportProtos.FromDeviceRPCResponseProto.newBuilder() |
|||
.setRequestIdMSB(requestId.getMostSignificantBits()) |
|||
.setRequestIdLSB(requestId.getLeastSignificantBits()) |
|||
.setError(RpcError.NOT_FOUND.ordinal()) |
|||
.build(); |
|||
var nfMsg = ToRuleEngineNotificationMsg.newBuilder().setFromDeviceRpcResponse(proto).build(); |
|||
var queueMsg = new TbProtoQueueMsg<>(requestId, nfMsg); |
|||
doCallRealMethod().when(defaultTbRuleEngineConsumerServiceMock).handleNotification(requestId, queueMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbRuleEngineConsumerServiceMock.handleNotification(requestId, queueMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
var responseCaptor = ArgumentCaptor.forClass(FromDeviceRpcResponse.class); |
|||
then(tbDeviceRpcServiceMock).should().processRpcResponseFromDevice(responseCaptor.capture()); |
|||
var response = responseCaptor.getValue(); |
|||
assertThat(response.getId()).isEqualTo(requestId); |
|||
assertThat(response.getError()).contains(RpcError.NOT_FOUND); |
|||
assertThat(response.getResponse()).isEmpty(); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,145 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.coapserver; |
|||
|
|||
import org.eclipse.californium.core.CoapServer; |
|||
import org.eclipse.californium.core.network.CoapEndpoint; |
|||
import org.eclipse.californium.core.server.resources.Resource; |
|||
import org.eclipse.californium.scandium.DTLSConnector; |
|||
import org.eclipse.californium.scandium.config.DtlsConnectorConfig; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.Mock; |
|||
import org.mockito.MockedConstruction; |
|||
import org.mockito.MockedStatic; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.common.util.ThingsBoardExecutors; |
|||
|
|||
import java.net.DatagramSocket; |
|||
import java.net.InetAddress; |
|||
import java.net.InetSocketAddress; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.anyString; |
|||
import static org.mockito.Mockito.doThrow; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.mockConstruction; |
|||
import static org.mockito.Mockito.mockStatic; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class DefaultCoapServerServiceTest { |
|||
|
|||
private static final String HOST = "127.0.0.1"; |
|||
|
|||
@Mock |
|||
private CoapServerContext mockCoapServerContext; |
|||
|
|||
private DefaultCoapServerService service; |
|||
private DatagramSocket occupiedSocket; |
|||
private int occupiedPort; |
|||
|
|||
@BeforeEach |
|||
public void setUp() throws Exception { |
|||
occupiedSocket = new DatagramSocket(new InetSocketAddress(InetAddress.getByName(HOST), 0)); |
|||
occupiedPort = occupiedSocket.getLocalPort(); |
|||
|
|||
service = new DefaultCoapServerService(); |
|||
ReflectionTestUtils.setField(service, "coapServerContext", mockCoapServerContext); |
|||
|
|||
when(mockCoapServerContext.getHost()).thenReturn(HOST); |
|||
when(mockCoapServerContext.getPort()).thenReturn(occupiedPort); |
|||
when(mockCoapServerContext.getDtlsSettings()).thenReturn(null); |
|||
} |
|||
|
|||
@AfterEach |
|||
public void tearDown() { |
|||
if (occupiedSocket != null && !occupiedSocket.isClosed()) { |
|||
occupiedSocket.close(); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void whenPlainBindFails_thenInitThrowsAndReleasesCoapServer() { |
|||
assertThatThrownBy(() -> service.init()) |
|||
.isInstanceOf(IllegalStateException.class) |
|||
.hasMessageContaining("None of the server endpoints could be started"); |
|||
|
|||
assertThat(ReflectionTestUtils.getField(service, "server")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "dtlsSessionsExecutor")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "dtlsConnector")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "dtlsCoapEndpoint")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "tbDtlsCertificateVerifier")).isNull(); |
|||
} |
|||
|
|||
@Test |
|||
public void whenDtlsEnabledAndStartFails_thenInitShutsDownDtlsExecutorAndReleasesCoapServer() throws Exception { |
|||
// DTLS enabled: the DTLS endpoint is created and dtlsSessionsExecutor is scheduled before server.start().
|
|||
// This exercises the catch's dtlsSessionsExecutor.shutdownNow() branch, which the plain-bind test does not.
|
|||
TbCoapDtlsSettings mockDtlsSettings = mock(TbCoapDtlsSettings.class); |
|||
when(mockCoapServerContext.getDtlsSettings()).thenReturn(mockDtlsSettings); |
|||
|
|||
DtlsConnectorConfig mockDtlsConfig = mock(DtlsConnectorConfig.class); |
|||
when(mockDtlsConfig.getAddress()).thenReturn(new InetSocketAddress(InetAddress.getByName(HOST), occupiedPort + 1)); |
|||
TbCoapDtlsCertificateVerifier mockVerifier = mock(TbCoapDtlsCertificateVerifier.class); |
|||
when(mockVerifier.getDtlsSessionReportTimeout()).thenReturn(1800000L); |
|||
when(mockDtlsConfig.getAdvancedCertificateVerifier()).thenReturn(mockVerifier); |
|||
when(mockDtlsSettings.dtlsConnectorConfig(any())).thenReturn(mockDtlsConfig); |
|||
|
|||
ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); |
|||
Resource mockRoot = mock(Resource.class); |
|||
|
|||
try (MockedStatic<ThingsBoardExecutors> executorsStatic = mockStatic(ThingsBoardExecutors.class); |
|||
MockedConstruction<CoapServer> serverMock = mockConstruction(CoapServer.class, (server, ctx) -> { |
|||
when(server.getRoot()).thenReturn(mockRoot); |
|||
doThrow(new IllegalStateException("None of the server endpoints could be started")).when(server).start(); |
|||
}); |
|||
MockedConstruction<DTLSConnector> dtlsMock = mockConstruction(DTLSConnector.class); |
|||
MockedConstruction<CoapEndpoint.Builder> builderMock = mockConstruction(CoapEndpoint.Builder.class, (builder, ctx) -> { |
|||
when(builder.setInetSocketAddress(any())).thenReturn(builder); |
|||
when(builder.setConfiguration(any())).thenReturn(builder); |
|||
when(builder.setConnector(any(DTLSConnector.class))).thenReturn(builder); |
|||
when(builder.build()).thenReturn(mock(CoapEndpoint.class)); |
|||
})) { |
|||
|
|||
executorsStatic.when(() -> ThingsBoardExecutors.newSingleThreadScheduledExecutor(anyString())).thenReturn(mockExecutor); |
|||
|
|||
assertThatThrownBy(() -> service.init()) |
|||
.isInstanceOf(IllegalStateException.class) |
|||
.hasMessageContaining("None of the server endpoints could be started"); |
|||
|
|||
// DTLS branch was actually entered and the executor was created...
|
|||
verify(mockDtlsSettings).dtlsConnectorConfig(any()); |
|||
// ...and the cleanup branch shut it down and destroyed the server.
|
|||
verify(mockExecutor).shutdownNow(); |
|||
verify(serverMock.constructed().get(0)).destroy(); |
|||
} |
|||
|
|||
assertThat(ReflectionTestUtils.getField(service, "server")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "dtlsSessionsExecutor")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "dtlsConnector")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "dtlsCoapEndpoint")).isNull(); |
|||
assertThat(ReflectionTestUtils.getField(service, "tbDtlsCertificateVerifier")).isNull(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,251 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.queue.common.consumer; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueMsg; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Queue; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ConcurrentLinkedQueue; |
|||
import java.util.concurrent.CopyOnWriteArrayList; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicBoolean; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.BooleanSupplier; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.Mockito.mock; |
|||
|
|||
@Slf4j |
|||
class QueueConsumerManagerTest { |
|||
|
|||
private static final long POLL_INTERVAL_MS = 20L; |
|||
// Before asserting the consumer never polled we wait until the loop has evaluated the readiness gate at least
|
|||
// this many times. That proves the consumer thread is actually running and deciding not to poll, rather than the
|
|||
// assertion passing vacuously because the thread simply has not started yet.
|
|||
private static final int MIN_READINESS_CHECKS = 3; |
|||
|
|||
private final AtomicBoolean readyToProcess = new AtomicBoolean(false); |
|||
private final AtomicInteger readinessChecks = new AtomicInteger(); |
|||
private final TestQueueConsumer consumer = new TestQueueConsumer(); |
|||
private ExecutorService consumerExecutor; |
|||
private QueueConsumerManager<TbQueueMsg> manager; |
|||
|
|||
@AfterEach |
|||
void tearDown() { |
|||
if (manager != null) { |
|||
manager.stop(); |
|||
} |
|||
if (consumerExecutor != null) { |
|||
consumerExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
void eventQueuedWhileNotReadyIsDeliveredAfterReadinessGateOpensInsteadOfBeingDropped() { |
|||
List<TbQueueMsg> delivered = new CopyOnWriteArrayList<>(); |
|||
|
|||
consumer.enqueue(List.of(mock(TbQueueMsg.class))); |
|||
|
|||
// The processor is unconditional: only the readiness gate may hold the event back, so delivery proves the
|
|||
// gate (not the processor) is what kept the event queued while not ready.
|
|||
manager = launchManager(consumer, countingReadiness(readyToProcess, readinessChecks), (msgs, c) -> { |
|||
delivered.addAll(msgs); |
|||
c.commit(); |
|||
}); |
|||
|
|||
// The loop is running and repeatedly evaluating the gate during the not-ready (sync) window...
|
|||
awaitReadinessGateEvaluated(readinessChecks); |
|||
// ...yet the queued event is neither polled nor delivered - it stays in the queue rather than being dropped.
|
|||
assertThat(consumer.getPollCount()) |
|||
.as("consumer must not poll while not ready") |
|||
.isZero(); |
|||
assertThat(delivered) |
|||
.as("event must not be delivered while not ready") |
|||
.isEmpty(); |
|||
|
|||
// Sync completes - the processor becomes ready.
|
|||
readyToProcess.set(true); |
|||
|
|||
await().atMost(5, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> assertThat(delivered) |
|||
.as("event queued during the not-ready window must be delivered, not dropped") |
|||
.hasSize(1)); |
|||
} |
|||
|
|||
@Test |
|||
void consumerIsNotPolledWhileNotReadyToProcess() { |
|||
manager = launchManager(consumer, countingReadiness(readyToProcess, readinessChecks), (msgs, c) -> c.commit()); |
|||
|
|||
awaitReadinessGateEvaluated(readinessChecks); |
|||
assertThat(consumer.getPollCount()) |
|||
.as("consumer must not be polled while not ready to process") |
|||
.isZero(); |
|||
|
|||
readyToProcess.set(true); |
|||
await().atMost(5, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> assertThat(consumer.getPollCount()) |
|||
.as("consumer resumes polling once ready") |
|||
.isPositive()); |
|||
} |
|||
|
|||
@Test |
|||
void consumerWithoutReadinessCheckPollsAndDeliversImmediately() { |
|||
List<TbQueueMsg> delivered = new CopyOnWriteArrayList<>(); |
|||
|
|||
consumer.enqueue(List.of(mock(TbQueueMsg.class))); |
|||
|
|||
// No readiness gate configured - the consumer must default to "always ready", preserving the behaviour every
|
|||
// consumer that does not opt in relies on.
|
|||
manager = launchManager(consumer, null, (msgs, c) -> { |
|||
delivered.addAll(msgs); |
|||
c.commit(); |
|||
}); |
|||
|
|||
await().atMost(5, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> assertThat(delivered) |
|||
.as("consumer without a readiness gate must poll and deliver immediately") |
|||
.hasSize(1)); |
|||
} |
|||
|
|||
@Test |
|||
void consumerLoopExitsWhenInterruptedWhileNotReady() throws Exception { |
|||
manager = launchManager(consumer, countingReadiness(readyToProcess, readinessChecks), (msgs, c) -> c.commit()); |
|||
|
|||
// The loop is parked in the not-ready wait...
|
|||
awaitReadinessGateEvaluated(readinessChecks); |
|||
|
|||
// ...interrupting the worker (as shutdownNow does on stop) must end the loop, not spin or hang.
|
|||
consumerExecutor.shutdownNow(); |
|||
assertThat(consumerExecutor.awaitTermination(5, TimeUnit.SECONDS)) |
|||
.as("consumer loop must exit when interrupted while waiting to become ready") |
|||
.isTrue(); |
|||
} |
|||
|
|||
private static void awaitReadinessGateEvaluated(AtomicInteger readinessChecks) { |
|||
await().atMost(5, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> assertThat(readinessChecks.get()) |
|||
.as("consumer loop must be running and repeatedly evaluating the readiness gate") |
|||
.isGreaterThanOrEqualTo(MIN_READINESS_CHECKS)); |
|||
} |
|||
|
|||
private static BooleanSupplier countingReadiness(AtomicBoolean ready, AtomicInteger readinessChecks) { |
|||
return () -> { |
|||
readinessChecks.incrementAndGet(); |
|||
return ready.get(); |
|||
}; |
|||
} |
|||
|
|||
private QueueConsumerManager<TbQueueMsg> launchManager(TestQueueConsumer consumer, BooleanSupplier readinessCheck, |
|||
QueueConsumerManager.MsgPackProcessor<TbQueueMsg> processor) { |
|||
consumerExecutor = Executors.newSingleThreadExecutor(); |
|||
QueueConsumerManager<TbQueueMsg> queueConsumerManager = QueueConsumerManager.<TbQueueMsg>builder() |
|||
.name("test-consumer") |
|||
.pollInterval(POLL_INTERVAL_MS) |
|||
.consumerCreator(() -> consumer) |
|||
.consumerExecutor(consumerExecutor) |
|||
.readinessCheck(readinessCheck) |
|||
.msgPackProcessor(processor) |
|||
.build(); |
|||
queueConsumerManager.subscribe(); |
|||
queueConsumerManager.launch(); |
|||
return queueConsumerManager; |
|||
} |
|||
|
|||
private static class TestQueueConsumer implements TbQueueConsumer<TbQueueMsg> { |
|||
|
|||
private final Queue<List<TbQueueMsg>> batches = new ConcurrentLinkedQueue<>(); |
|||
private final AtomicInteger pollCount = new AtomicInteger(); |
|||
private volatile boolean stopped; |
|||
|
|||
void enqueue(List<TbQueueMsg> batch) { |
|||
batches.add(batch); |
|||
} |
|||
|
|||
int getPollCount() { |
|||
return pollCount.get(); |
|||
} |
|||
|
|||
@Override |
|||
public List<TbQueueMsg> poll(long durationInMillis) { |
|||
pollCount.incrementAndGet(); |
|||
List<TbQueueMsg> batch = batches.poll(); |
|||
if (batch != null) { |
|||
return batch; |
|||
} |
|||
try { |
|||
Thread.sleep(durationInMillis); |
|||
} catch (InterruptedException e) { |
|||
Thread.currentThread().interrupt(); |
|||
} |
|||
return Collections.emptyList(); |
|||
} |
|||
|
|||
@Override |
|||
public String getTopic() { |
|||
return "test-topic"; |
|||
} |
|||
|
|||
@Override |
|||
public void subscribe() { |
|||
} |
|||
|
|||
@Override |
|||
public void subscribe(Set<TopicPartitionInfo> partitions) { |
|||
} |
|||
|
|||
@Override |
|||
public void stop() { |
|||
stopped = true; |
|||
} |
|||
|
|||
@Override |
|||
public void unsubscribe() { |
|||
stopped = true; |
|||
} |
|||
|
|||
@Override |
|||
public void commit() { |
|||
} |
|||
|
|||
@Override |
|||
public boolean isStopped() { |
|||
return stopped; |
|||
} |
|||
|
|||
@Override |
|||
public Set<TopicPartitionInfo> getPartitions() { |
|||
return Collections.emptySet(); |
|||
} |
|||
|
|||
@Override |
|||
public List<String> getFullTopicNames() { |
|||
return Collections.emptyList(); |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,115 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.transport.lwm2m.bootstrap; |
|||
|
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.mockito.junit.jupiter.MockitoSettings; |
|||
import org.mockito.quality.Strictness; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.common.transport.TransportService; |
|||
import org.thingsboard.server.transport.lwm2m.bootstrap.secure.TbLwM2MDtlsBootstrapCertificateVerifier; |
|||
import org.thingsboard.server.transport.lwm2m.bootstrap.store.LwM2MBootstrapSecurityStore; |
|||
import org.thingsboard.server.transport.lwm2m.bootstrap.store.LwM2MInMemoryBootstrapConfigStore; |
|||
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportBootstrapConfig; |
|||
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig; |
|||
|
|||
import java.net.DatagramSocket; |
|||
import java.net.InetAddress; |
|||
import java.net.InetSocketAddress; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
@MockitoSettings(strictness = Strictness.LENIENT) |
|||
public class LwM2MTransportBootstrapServiceTest { |
|||
|
|||
private static final String HOST = "127.0.0.1"; |
|||
|
|||
@Mock |
|||
private LwM2MTransportServerConfig serverConfig; |
|||
|
|||
@Mock |
|||
private LwM2MTransportBootstrapConfig bootstrapConfig; |
|||
|
|||
@Mock |
|||
private LwM2MBootstrapSecurityStore lwM2MBootstrapSecurityStore; |
|||
|
|||
@Mock |
|||
private LwM2MInMemoryBootstrapConfigStore lwM2MInMemoryBootstrapConfigStore; |
|||
|
|||
@Mock |
|||
private TransportService transportService; |
|||
|
|||
@Mock |
|||
private TbLwM2MDtlsBootstrapCertificateVerifier certificateVerifier; |
|||
|
|||
private LwM2MTransportBootstrapService service; |
|||
private DatagramSocket occupiedPlain; |
|||
private DatagramSocket occupiedSecure; |
|||
|
|||
@BeforeEach |
|||
public void setUp() throws Exception { |
|||
occupiedPlain = new DatagramSocket(new InetSocketAddress(InetAddress.getByName(HOST), 0)); |
|||
occupiedSecure = new DatagramSocket(new InetSocketAddress(InetAddress.getByName(HOST), 0)); |
|||
|
|||
when(bootstrapConfig.getHost()).thenReturn(HOST); |
|||
when(bootstrapConfig.getPort()).thenReturn(occupiedPlain.getLocalPort()); |
|||
when(bootstrapConfig.getSecureHost()).thenReturn(HOST); |
|||
when(bootstrapConfig.getSecurePort()).thenReturn(occupiedSecure.getLocalPort()); |
|||
when(bootstrapConfig.getSslCredentials()).thenReturn(null); |
|||
|
|||
when(serverConfig.isRecommendedCiphers()).thenReturn(false); |
|||
when(serverConfig.isRecommendedSupportedGroups()).thenReturn(false); |
|||
when(serverConfig.getDtlsRetransmissionTimeout()).thenReturn(9000); |
|||
when(serverConfig.getDtlsCidLength()).thenReturn(null); |
|||
|
|||
service = new LwM2MTransportBootstrapService( |
|||
serverConfig, |
|||
bootstrapConfig, |
|||
lwM2MBootstrapSecurityStore, |
|||
lwM2MInMemoryBootstrapConfigStore, |
|||
transportService, |
|||
certificateVerifier |
|||
); |
|||
} |
|||
|
|||
@AfterEach |
|||
public void tearDown() { |
|||
if (occupiedPlain != null && !occupiedPlain.isClosed()) { |
|||
occupiedPlain.close(); |
|||
} |
|||
if (occupiedSecure != null && !occupiedSecure.isClosed()) { |
|||
occupiedSecure.close(); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void whenEndpointsFailToStart_thenInitThrowsAndReleasesBootstrapServer() { |
|||
assertThatThrownBy(() -> service.init()) |
|||
.isInstanceOf(IllegalStateException.class) |
|||
.hasMessageContaining("None of the server endpoints could be started"); |
|||
|
|||
assertThat(ReflectionTestUtils.getField(service, "server")).isNull(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,202 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.common.transport.limits; |
|||
|
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.params.ParameterizedTest; |
|||
import org.junit.jupiter.params.provider.EnumSource; |
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.TenantProfileId; |
|||
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; |
|||
import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; |
|||
import org.thingsboard.server.common.transport.TransportTenantProfileCache; |
|||
import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; |
|||
|
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.CountDownLatch; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
class DefaultTransportRateLimitServiceTest { |
|||
|
|||
private TransportTenantProfileCache tenantProfileCache; |
|||
private ExecutorService executor; |
|||
|
|||
private final TenantId tenant = TenantId.fromUUID(UUID.randomUUID()); |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
tenantProfileCache = mock(TransportTenantProfileCache.class); |
|||
executor = Executors.newCachedThreadPool(); |
|||
} |
|||
|
|||
@AfterEach |
|||
void tearDown() { |
|||
executor.shutdownNow(); |
|||
} |
|||
|
|||
@Test |
|||
void checkLimitsDoesNotHoldMapBinLockAcrossProfileFetch() throws Exception { |
|||
// Two concurrent rate-limit checks for the SAME tenant must both be able to reach
|
|||
// the (blocking) tenant-profile fetch concurrently. If the blocking fetch runs inside
|
|||
// ConcurrentHashMap.computeIfAbsent, the second caller is stuck on the bin reservation
|
|||
// node and never reaches the fetch -> the latch never reaches zero.
|
|||
CountDownLatch bothCallersReachedFetch = new CountDownLatch(2); |
|||
CountDownLatch releaseFetch = new CountDownLatch(1); |
|||
|
|||
when(tenantProfileCache.get(tenant)).thenAnswer(invocation -> { |
|||
bothCallersReachedFetch.countDown(); |
|||
releaseFetch.await(5, TimeUnit.SECONDS); |
|||
return tenantProfile(); |
|||
}); |
|||
|
|||
DefaultTransportRateLimitService service = new DefaultTransportRateLimitService(tenantProfileCache); |
|||
|
|||
Runnable check = () -> service.checkLimits(tenant, null, null, 1, false); |
|||
executor.submit(check); |
|||
executor.submit(check); |
|||
|
|||
boolean bothReached = bothCallersReachedFetch.await(3, TimeUnit.SECONDS); |
|||
releaseFetch.countDown(); |
|||
|
|||
assertThat(bothReached) |
|||
.as("both checkLimits calls should reach the profile fetch concurrently (no bin lock across I/O)") |
|||
.isTrue(); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@EnumSource(TransportLimitsType.class) |
|||
void eachLimitsTypeReadsItsOwnProfileFields(TransportLimitsType type) { |
|||
// Distinct sentinel per profile field so a transposed method reference (e.g. GATEWAY_DEVICE_LIMITS
|
|||
// wired to the plain gateway getters) resolves to the wrong value and fails the assertion.
|
|||
DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); |
|||
config.setTransportTenantMsgRateLimit("tenant-msg"); |
|||
config.setTransportTenantTelemetryMsgRateLimit("tenant-tele-msg"); |
|||
config.setTransportTenantTelemetryDataPointsRateLimit("tenant-tele-dp"); |
|||
config.setTransportDeviceMsgRateLimit("device-msg"); |
|||
config.setTransportDeviceTelemetryMsgRateLimit("device-tele-msg"); |
|||
config.setTransportDeviceTelemetryDataPointsRateLimit("device-tele-dp"); |
|||
config.setTransportGatewayMsgRateLimit("gateway-msg"); |
|||
config.setTransportGatewayTelemetryMsgRateLimit("gateway-tele-msg"); |
|||
config.setTransportGatewayTelemetryDataPointsRateLimit("gateway-tele-dp"); |
|||
config.setTransportGatewayDeviceMsgRateLimit("gateway-device-msg"); |
|||
config.setTransportGatewayDeviceTelemetryMsgRateLimit("gateway-device-tele-msg"); |
|||
config.setTransportGatewayDeviceTelemetryDataPointsRateLimit("gateway-device-tele-dp"); |
|||
|
|||
String prefix = switch (type) { |
|||
case TENANT_LIMITS -> "tenant"; |
|||
case DEVICE_LIMITS -> "device"; |
|||
case GATEWAY_LIMITS -> "gateway"; |
|||
case GATEWAY_DEVICE_LIMITS -> "gateway-device"; |
|||
}; |
|||
|
|||
assertThat(type.getRegularMsgRateLimit().apply(config)).isEqualTo(prefix + "-msg"); |
|||
assertThat(type.getTelemetryMsgRateLimit().apply(config)).isEqualTo(prefix + "-tele-msg"); |
|||
assertThat(type.getTelemetryDataPointsRateLimit().apply(config)).isEqualTo(prefix + "-tele-dp"); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@EnumSource(EntityLevel.class) |
|||
void profileUpdateReachesEntityTrackedDuringFirstCheck(EntityLevel level) { |
|||
DeviceId entity = new DeviceId(UUID.randomUUID()); |
|||
when(tenantProfileCache.get(tenant)).thenReturn(profileWithRegularMsgLimit(level, "100:600")); |
|||
DefaultTransportRateLimitService service = new DefaultTransportRateLimitService(tenantProfileCache); |
|||
|
|||
// First check resolves the (permissive) limit and must register the entity into the per-tenant
|
|||
// tracking set via the onMiss callback - otherwise a later update(tenantId) can't reach it.
|
|||
assertThat(level.check(service, tenant, entity)) |
|||
.as("permissive limit should allow the first %s check", level).isNull(); |
|||
|
|||
// Tighten the limit to a single message and push a profile update for this tenant.
|
|||
service.update(new TenantProfileUpdateResult(profileWithRegularMsgLimit(level, "1:600"), Set.of(tenant))); |
|||
|
|||
// The freshly merged "1:600" bucket allows exactly one message...
|
|||
assertThat(level.check(service, tenant, entity)).isNull(); |
|||
// ...and blocks the next one. This only happens if update(tenantId) reached the tracked entity.
|
|||
assertThat(level.check(service, tenant, entity)) |
|||
.as("update(tenantId) must reach the tracked %s so the tightened limit applies", level).isNotNull(); |
|||
} |
|||
|
|||
private TenantProfile tenantProfile() { |
|||
return profileWith(new DefaultTenantProfileConfiguration()); |
|||
} |
|||
|
|||
private TenantProfile profileWithRegularMsgLimit(EntityLevel level, String regularMsgRateLimit) { |
|||
DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); |
|||
level.setRegularMsgRateLimit(config, regularMsgRateLimit); |
|||
return profileWith(config); |
|||
} |
|||
|
|||
private TenantProfile profileWith(DefaultTenantProfileConfiguration config) { |
|||
TenantProfile profile = new TenantProfile(new TenantProfileId(UUID.randomUUID())); |
|||
profile.setName("test-profile"); |
|||
TenantProfileData profileData = new TenantProfileData(); |
|||
profileData.setConfiguration(config); |
|||
profile.setProfileData(profileData); |
|||
return profile; |
|||
} |
|||
|
|||
private enum EntityLevel { |
|||
DEVICE { |
|||
@Override |
|||
void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value) { |
|||
config.setTransportDeviceMsgRateLimit(value); |
|||
} |
|||
|
|||
@Override |
|||
Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId) { |
|||
return service.checkLimits(tenantId, null, entityId, 0, false); |
|||
} |
|||
}, |
|||
GATEWAY { |
|||
@Override |
|||
void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value) { |
|||
config.setTransportGatewayMsgRateLimit(value); |
|||
} |
|||
|
|||
@Override |
|||
Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId) { |
|||
return service.checkLimits(tenantId, entityId, null, 0, false); |
|||
} |
|||
}, |
|||
GATEWAY_DEVICE { |
|||
@Override |
|||
void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value) { |
|||
config.setTransportGatewayDeviceMsgRateLimit(value); |
|||
} |
|||
|
|||
@Override |
|||
Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId) { |
|||
return service.checkLimits(tenantId, null, entityId, 0, true); |
|||
} |
|||
}; |
|||
|
|||
abstract void setRegularMsgRateLimit(DefaultTenantProfileConfiguration config, String value); |
|||
|
|||
abstract Object check(DefaultTransportRateLimitService service, TenantId tenantId, DeviceId entityId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,191 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.common.transport.service; |
|||
|
|||
import com.google.common.util.concurrent.Striped; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.thingsboard.server.common.data.ApiUsageState; |
|||
import org.thingsboard.server.common.data.ApiUsageStateValue; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.ApiUsageStateId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.TenantProfileId; |
|||
import org.thingsboard.server.common.transport.TransportService; |
|||
import org.thingsboard.server.common.transport.limits.TransportRateLimitService; |
|||
import org.thingsboard.server.common.util.ProtoUtils; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileResponseMsg; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.CompletableFuture; |
|||
import java.util.concurrent.CountDownLatch; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.Future; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.locks.Lock; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.anyBoolean; |
|||
import static org.mockito.Mockito.doNothing; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.times; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
class DefaultTransportTenantProfileCacheTest { |
|||
|
|||
private DefaultTransportTenantProfileCache cache; |
|||
private TransportService transportService; |
|||
private TransportRateLimitService rateLimitService; |
|||
private ExecutorService executor; |
|||
|
|||
// Must match DefaultTransportTenantProfileCache.TENANT_PROFILE_FETCH_LOCK_STRIPES.
|
|||
private static final int STRIPE_COUNT = 1024; |
|||
|
|||
private final TenantId tenantA = TenantId.fromUUID(UUID.randomUUID()); |
|||
// Deterministically pick a tenant that maps to a DIFFERENT stripe than tenantA, so the cross-tenant
|
|||
// test below cannot flake on the ~1/1024 chance two random UUIDs hash to the same stripe.
|
|||
private final TenantId tenantB = differentStripeFrom(tenantA); |
|||
|
|||
private static TenantId differentStripeFrom(TenantId other) { |
|||
Striped<Lock> probe = Striped.lock(STRIPE_COUNT); |
|||
TenantId candidate = TenantId.fromUUID(UUID.randomUUID()); |
|||
while (probe.get(candidate) == probe.get(other)) { |
|||
candidate = TenantId.fromUUID(UUID.randomUUID()); |
|||
} |
|||
return candidate; |
|||
} |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
cache = new DefaultTransportTenantProfileCache(); |
|||
transportService = mock(TransportService.class); |
|||
rateLimitService = mock(TransportRateLimitService.class); |
|||
doNothing().when(rateLimitService).update(any(TenantId.class), anyBoolean()); |
|||
cache.setTransportService(transportService); |
|||
cache.setRateLimitService(rateLimitService); |
|||
executor = Executors.newCachedThreadPool(); |
|||
} |
|||
|
|||
@AfterEach |
|||
void tearDown() { |
|||
executor.shutdownNow(); |
|||
} |
|||
|
|||
@Test |
|||
void fetchForOneTenantDoesNotBlockResolutionOfAnotherTenant() throws Exception { |
|||
CountDownLatch tenantAFetchStarted = new CountDownLatch(1); |
|||
CountDownLatch releaseTenantA = new CountDownLatch(1); |
|||
|
|||
GetEntityProfileResponseMsg responseA = responseFor(tenantA); |
|||
GetEntityProfileResponseMsg responseB = responseFor(tenantB); |
|||
|
|||
when(transportService.getEntityProfile(any())).thenAnswer(invocation -> { |
|||
GetEntityProfileRequestMsg msg = invocation.getArgument(0); |
|||
TenantId requested = TenantId.fromUUID(new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); |
|||
if (requested.equals(tenantA)) { |
|||
tenantAFetchStarted.countDown(); |
|||
releaseTenantA.await(5, TimeUnit.SECONDS); |
|||
return responseA; |
|||
} |
|||
return responseB; |
|||
}); |
|||
|
|||
// T1 starts fetching tenantA's profile and blocks inside the cross-service round-trip.
|
|||
Future<TenantProfile> tenantAResult = executor.submit(() -> cache.get(tenantA)); |
|||
assertThat(tenantAFetchStarted.await(5, TimeUnit.SECONDS)) |
|||
.as("tenantA fetch should have started").isTrue(); |
|||
|
|||
// T2 resolves a different tenant - it must NOT wait for tenantA's in-flight fetch.
|
|||
// Fails today (single global lock); passes once locking is per-tenant.
|
|||
TenantProfile tenantBProfile = CompletableFuture |
|||
.supplyAsync(() -> cache.get(tenantB), executor) |
|||
.get(2, TimeUnit.SECONDS); |
|||
assertThat(tenantBProfile).isNotNull(); |
|||
|
|||
releaseTenantA.countDown(); |
|||
assertThat(tenantAResult.get(5, TimeUnit.SECONDS)).isNotNull(); |
|||
} |
|||
|
|||
@Test |
|||
void concurrentMissesForSameTenantDedupeToSingleFetch() throws Exception { |
|||
// The per-tenant lock exists precisely so that concurrent cold misses for the SAME tenant collapse
|
|||
// into a single cross-service fetch (the rest are served from cache). Assert that contract directly.
|
|||
int callers = 8; |
|||
CountDownLatch fetchStarted = new CountDownLatch(1); |
|||
CountDownLatch releaseFetch = new CountDownLatch(1); |
|||
|
|||
when(transportService.getEntityProfile(any())).thenAnswer(invocation -> { |
|||
fetchStarted.countDown(); |
|||
// Hold the (single) in-flight fetch open while the other callers pile up on the per-tenant lock.
|
|||
releaseFetch.await(5, TimeUnit.SECONDS); |
|||
return responseFor(tenantA); |
|||
}); |
|||
|
|||
CountDownLatch allSubmitted = new CountDownLatch(callers); |
|||
List<Future<TenantProfile>> results = new ArrayList<>(); |
|||
for (int i = 0; i < callers; i++) { |
|||
results.add(executor.submit(() -> { |
|||
allSubmitted.countDown(); |
|||
return cache.get(tenantA); |
|||
})); |
|||
} |
|||
|
|||
assertThat(allSubmitted.await(5, TimeUnit.SECONDS)).as("all callers should start").isTrue(); |
|||
assertThat(fetchStarted.await(5, TimeUnit.SECONDS)).as("the first fetch should start").isTrue(); |
|||
releaseFetch.countDown(); |
|||
|
|||
for (Future<TenantProfile> result : results) { |
|||
assertThat(result.get(5, TimeUnit.SECONDS)).isNotNull(); |
|||
} |
|||
// All 8 callers resolved the same tenant, but only one of them hit the backend.
|
|||
verify(transportService, times(1)).getEntityProfile(any()); |
|||
} |
|||
|
|||
private GetEntityProfileResponseMsg responseFor(TenantId tenantId) { |
|||
TenantProfile profile = new TenantProfile(new TenantProfileId(UUID.randomUUID())); |
|||
profile.setName("profile-" + tenantId.getId()); |
|||
return GetEntityProfileResponseMsg.newBuilder() |
|||
.setEntityType(EntityType.TENANT.name()) |
|||
.setTenantProfile(ProtoUtils.toProto(profile)) |
|||
.setApiState(ProtoUtils.toProto(enabledApiUsageState(tenantId))) |
|||
.build(); |
|||
} |
|||
|
|||
private ApiUsageState enabledApiUsageState(TenantId tenantId) { |
|||
ApiUsageState state = new ApiUsageState(new ApiUsageStateId(UUID.randomUUID())); |
|||
state.setTenantId(tenantId); |
|||
state.setEntityId(tenantId); |
|||
state.setTransportState(ApiUsageStateValue.ENABLED); |
|||
state.setDbStorageState(ApiUsageStateValue.ENABLED); |
|||
state.setReExecState(ApiUsageStateValue.ENABLED); |
|||
state.setJsExecState(ApiUsageStateValue.ENABLED); |
|||
state.setTbelExecState(ApiUsageStateValue.ENABLED); |
|||
state.setEmailExecState(ApiUsageStateValue.ENABLED); |
|||
state.setSmsExecState(ApiUsageStateValue.ENABLED); |
|||
state.setAlarmExecState(ApiUsageStateValue.ENABLED); |
|||
state.setVersion(1L); |
|||
return state; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,138 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.dao.service.validator; |
|||
|
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.InjectMocks; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.device.credentials.BasicMqttCredentials; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentialsType; |
|||
import org.thingsboard.server.dao.device.DeviceCredentialsDao; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.dao.exception.DeviceCredentialsValidationException; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThatCode; |
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
import static org.mockito.BDDMockito.willReturn; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
class DeviceCredentialsDataValidatorTest { |
|||
|
|||
@Mock |
|||
DeviceCredentialsDao deviceCredentialsDao; |
|||
@Mock |
|||
DeviceService deviceService; |
|||
@InjectMocks |
|||
DeviceCredentialsDataValidator validator; |
|||
|
|||
final TenantId tenantId = TenantId.fromUUID(UUID.fromString("9ef79cdf-37a8-4119-b682-2e7ed4e018da")); |
|||
final DeviceId deviceId = new DeviceId(UUID.fromString("11111111-1111-1111-1111-111111111111")); |
|||
|
|||
@Test |
|||
void rejectsNewlineInAccessToken() { |
|||
DeviceCredentials creds = accessToken("safe_token\nentrypoint: [\"/bin/sh\"]"); |
|||
|
|||
assertThatThrownBy(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.isInstanceOf(DeviceCredentialsValidationException.class) |
|||
.hasMessageContaining("credentialsId") |
|||
.hasMessageContaining("control characters"); |
|||
} |
|||
|
|||
@Test |
|||
void rejectsCarriageReturnInAccessToken() { |
|||
DeviceCredentials creds = accessToken("token\rprivileged: true"); |
|||
|
|||
assertThatThrownBy(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.isInstanceOf(DeviceCredentialsValidationException.class) |
|||
.hasMessageContaining("control characters"); |
|||
} |
|||
|
|||
@Test |
|||
void rejectsNewlineInMqttClientId() { |
|||
DeviceCredentials creds = mqttBasic("cid\nentrypoint: x", "user", "pwd"); |
|||
|
|||
assertThatThrownBy(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.isInstanceOf(DeviceCredentialsValidationException.class) |
|||
.hasMessageContaining("clientId"); |
|||
} |
|||
|
|||
@Test |
|||
void rejectsNewlineInMqttUserName() { |
|||
DeviceCredentials creds = mqttBasic("cid", "user\nprivileged: true", "pwd"); |
|||
|
|||
assertThatThrownBy(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.isInstanceOf(DeviceCredentialsValidationException.class) |
|||
.hasMessageContaining("userName"); |
|||
} |
|||
|
|||
@Test |
|||
void rejectsNewlineInMqttPassword() { |
|||
DeviceCredentials creds = mqttBasic("cid", "user", "pwd\nentrypoint: x"); |
|||
|
|||
assertThatThrownBy(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.isInstanceOf(DeviceCredentialsValidationException.class) |
|||
.hasMessageContaining("password"); |
|||
} |
|||
|
|||
@Test |
|||
void acceptsValidCredentials() { |
|||
willReturn(new Device()).given(deviceService).findDeviceById(tenantId, deviceId); |
|||
DeviceCredentials creds = accessToken("safe_token_123"); |
|||
|
|||
assertThatCode(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.doesNotThrowAnyException(); |
|||
} |
|||
|
|||
@Test |
|||
void acceptsValidMqttBasicCredentials() { |
|||
willReturn(new Device()).given(deviceService).findDeviceById(tenantId, deviceId); |
|||
DeviceCredentials creds = mqttBasic("client-1", "user-1", "pwd-1"); |
|||
|
|||
assertThatCode(() -> validator.validateDataImpl(tenantId, creds)) |
|||
.doesNotThrowAnyException(); |
|||
} |
|||
|
|||
private DeviceCredentials accessToken(String token) { |
|||
DeviceCredentials c = new DeviceCredentials(); |
|||
c.setDeviceId(deviceId); |
|||
c.setCredentialsType(DeviceCredentialsType.ACCESS_TOKEN); |
|||
c.setCredentialsId(token); |
|||
return c; |
|||
} |
|||
|
|||
private DeviceCredentials mqttBasic(String clientId, String userName, String password) { |
|||
BasicMqttCredentials inner = new BasicMqttCredentials(); |
|||
inner.setClientId(clientId); |
|||
inner.setUserName(userName); |
|||
inner.setPassword(password); |
|||
DeviceCredentials c = new DeviceCredentials(); |
|||
c.setDeviceId(deviceId); |
|||
c.setCredentialsType(DeviceCredentialsType.MQTT_BASIC); |
|||
c.setCredentialsId("mqtt-credentials-id"); |
|||
c.setCredentialsValue(JacksonUtil.toString(inner)); |
|||
return c; |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue