Browse Source
# Conflicts: # ui-ngx/src/app/modules/home/components/rule-node/external/rest-api-call-config.component.tspull/14540/head
Dmytro Skarzhynets
8 months ago
1306 changed files with 42814 additions and 15299 deletions
@ -0,0 +1,31 @@ |
|||
|
|||
## Running tests in parallel with a reasonable memory usage |
|||
|
|||
```bash |
|||
export MAVEN_OPTS="-Xmx1024m" |
|||
export NODE_OPTIONS="--max_old_space_size=4096" |
|||
export SUREFIRE_JAVA_OPTS="-Xmx1200m -Xss256k -XX:+ExitOnOutOfMemoryError" |
|||
|
|||
mvn clean install -T6 -DskipTests |
|||
mvn test -pl='!application,!dao,!ui-ngx,!msa/js-executor,!msa/web-ui' -T4 |
|||
mvn test -pl dao -Dparallel=packages -DforkCount=4 |
|||
|
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='org.thingsboard.server.controller.**' -DforkCount=6 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='org.thingsboard.server.edge.**' -DforkCount=4 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='org.thingsboard.server.service.**' -DforkCount=6 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='org.thingsboard.server.transport.mqtt.**' -DforkCount=6 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='org.thingsboard.server.transport.coap.**' -DforkCount=6 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='org.thingsboard.server.transport.lwm2m.**' -DforkCount=6 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dsurefire.excludes='**/nosql/*Test.java' -Dtest='**/*TestSuite.java' -DforkCount=4 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
|
|||
#the rest of application tests |
|||
mvn test -pl application -Dtest=' |
|||
!**/nosql/*Test.java, |
|||
!org.thingsboard.server.controller.**, |
|||
!org.thingsboard.server.edge.**, |
|||
!org.thingsboard.server.service.**, |
|||
!org.thingsboard.server.transport.mqtt.**, |
|||
!org.thingsboard.server.transport.coap.**, |
|||
!org.thingsboard.server.transport.lwm2m.** |
|||
' -DforkCount=6 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
``` |
|||
@ -0,0 +1,29 @@ |
|||
{ |
|||
"fqn": "gateway_widgets.markdown_card", |
|||
"name": "Gateway Markdown/HTML Card", |
|||
"deprecated": false, |
|||
"image": "tb-image;/api/images/system/markdown_html_card_system_widget_image.png", |
|||
"description": "Renders markdown/HTML using the configurable pattern or function with applied attributes or time series values.", |
|||
"descriptor": { |
|||
"type": "latest", |
|||
"sizeX": 5, |
|||
"sizeY": 3.5, |
|||
"resources": [ |
|||
{ |
|||
"url": "tb-resource;/api/resource/js_module/system/gateway-management-extension.js", |
|||
"isModule": true |
|||
} |
|||
], |
|||
"templateHtml": "<tb-markdown-widget \n [ctx]=\"ctx\">\n</tb-markdown-widget>", |
|||
"templateCss": "#container tb-markdown-widget {\n height: 100%;\n display: block;\n}\n\n#container tb-markdown-widget .tb-markdown-view {\n height: 100%;\n overflow: auto;\n}\n", |
|||
"controllerScript": "self.onInit = function() {\n}\n\nself.onDataUpdated = function() {\n self.ctx.$scope.markdownWidget.onDataUpdated();\n}\n\nself.actionSources = function() {\n return {\n 'elementClick': {\n name: 'widget-action.element-click',\n multiple: true\n }\n };\n}\n\nself.typeParameters = function() {\n return {\n dataKeysOptional: true,\n datasourcesOptional: true,\n hasDataPageLink: true\n };\n}\n\nself.onDestroy = function() {\n}\n\n", |
|||
"settingsDirective": "tb-markdown-widget-settings", |
|||
"defaultConfig": "{\"datasources\":[{\"type\":\"function\",\"name\":\"function\",\"dataKeys\":[{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Random\",\"color\":\"#2196f3\",\"settings\":{},\"_hash\":0.15479322438769105,\"funcBody\":\"var value = prevValue + Math.random() * 100 - 50;\\nvar multiplier = Math.pow(10, 2 || 0);\\nvar value = Math.round(value * multiplier) / multiplier;\\nif (value < -1000) {\\n\\tvalue = -1000;\\n} else if (value > 1000) {\\n\\tvalue = 1000;\\n}\\nreturn value;\"}]}],\"timewindow\":{\"realtime\":{\"timewindowMs\":60000}},\"showTitle\":false,\"backgroundColor\":\"#fff\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"0px\",\"settings\":{\"markdownTextPattern\":\"### Markdown/HTML card\\n - **Current entity**: ${entityName}.\\n - **Current value**: ${Random}.\",\"markdownTextFunction\":\"return '# Some title\\\\n - Entity name: ' + data[0]['entityName'];\",\"useMarkdownTextFunction\":false},\"title\":\"Markdown/HTML Card\",\"showTitleIcon\":false,\"iconColor\":\"rgba(0, 0, 0, 0.87)\",\"iconSize\":\"24px\",\"titleTooltip\":\"\",\"dropShadow\":true,\"enableFullscreen\":true,\"widgetStyle\":{},\"titleStyle\":{\"fontSize\":\"16px\",\"fontWeight\":400},\"showLegend\":false}" |
|||
}, |
|||
"resources": null, |
|||
"scada": false, |
|||
"tags": [ |
|||
"web", |
|||
"markup" |
|||
] |
|||
} |
|||
File diff suppressed because it is too large
File diff suppressed because one or more lines are too long
@ -0,0 +1,35 @@ |
|||
/** |
|||
* Copyright © 2016-2025 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.springframework.stereotype.Component; |
|||
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; |
|||
|
|||
public interface TbMsgPackProcessingContextFactory { |
|||
|
|||
TbMsgPackProcessingContext create(String queueName, TbRuleEngineSubmitStrategy submitStrategy, boolean skipTimeouts); |
|||
|
|||
@Component |
|||
class DefaultTbMsgPackProcessingContextFactory implements TbMsgPackProcessingContextFactory { |
|||
|
|||
@Override |
|||
public TbMsgPackProcessingContext create(String queueName, TbRuleEngineSubmitStrategy submitStrategy, boolean skipTimeouts) { |
|||
return new TbMsgPackProcessingContext(queueName, submitStrategy, skipTimeouts); |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,41 @@ |
|||
/** |
|||
* 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.edge; |
|||
|
|||
import org.junit.Assert; |
|||
import org.junit.Test; |
|||
import org.thingsboard.edge.rpc.EdgeVersionComparator; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeVersion; |
|||
|
|||
public class EdgeLatestVersionTest { |
|||
|
|||
@Test |
|||
public void edgeLatestVersionIsSynchronizedTest() { |
|||
EdgeVersion currentHighestEdgeVersion = EdgeVersionComparator.getNewestEdgeVersion(); |
|||
|
|||
String projectVersion = EdgeLatestVersionTest.class.getPackage().getImplementationVersion(); |
|||
if (projectVersion == null || projectVersion.isBlank()) { |
|||
projectVersion = System.getProperty("project.version", "UNKNOWN"); |
|||
} |
|||
|
|||
String projectVersionDigits = projectVersion.replaceAll("\\D", ""); |
|||
String currentHighestEdgeVersionDigits = currentHighestEdgeVersion.name().replaceAll("\\D", ""); |
|||
|
|||
String msg = "EdgeVersion enum in edge.proto is out of sync. Please add respective " + projectVersionDigits + " to EdgeVersion"; |
|||
Assert.assertEquals(msg, projectVersionDigits, currentHighestEdgeVersionDigits); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,281 @@ |
|||
/** |
|||
* Copyright © 2016-2025 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.ruleengine; |
|||
|
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
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.InOrder; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.thingsboard.common.util.ThingsBoardExecutors; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.msg.TbMsgType; |
|||
import org.thingsboard.server.common.data.queue.ProcessingStrategy; |
|||
import org.thingsboard.server.common.data.queue.ProcessingStrategyType; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.common.data.queue.SubmitStrategy; |
|||
import org.thingsboard.server.common.data.queue.SubmitStrategyType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.common.stats.StatsFactory; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueMsg; |
|||
import org.thingsboard.server.queue.common.DefaultTbQueueMsgHeaders; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.QueueKey; |
|||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|||
import org.thingsboard.server.queue.memory.DefaultInMemoryStorage; |
|||
import org.thingsboard.server.queue.memory.InMemoryStorage; |
|||
import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; |
|||
import org.thingsboard.server.queue.provider.TbQueueProducerProvider; |
|||
import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; |
|||
import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; |
|||
import org.thingsboard.server.service.queue.TbMsgPackProcessingContextFactory; |
|||
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; |
|||
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; |
|||
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; |
|||
import org.thingsboard.server.service.stats.RuleEngineStatisticsService; |
|||
|
|||
import java.time.Duration; |
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.anyLong; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.ArgumentMatchers.isNull; |
|||
import static org.mockito.BDDMockito.given; |
|||
import static org.mockito.Mockito.doAnswer; |
|||
import static org.mockito.Mockito.inOrder; |
|||
import static org.mockito.Mockito.spy; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
class RuleEngineConsumerLoopTest { |
|||
|
|||
TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
|
|||
InMemoryStorage storage; |
|||
|
|||
@Mock |
|||
ActorSystemContext actorContext; |
|||
@Mock |
|||
StatsFactory statsFactory; |
|||
@Mock |
|||
TbRuleEngineQueueFactory queueFactory; |
|||
@Mock |
|||
RuleEngineStatisticsService statisticsService; |
|||
@Mock |
|||
TbServiceInfoProvider serviceInfoProvider; |
|||
@Mock |
|||
PartitionService partitionService; |
|||
@Mock |
|||
TbQueueProducerProvider producerProvider; |
|||
@Mock |
|||
TbQueueAdmin queueAdmin; |
|||
@Mock |
|||
TbMsgPackProcessingContextFactory packProcessingContextFactory; |
|||
@Mock |
|||
TbMsgPackProcessingContext packCtx; |
|||
|
|||
Queue mainQueue; |
|||
|
|||
TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer; |
|||
|
|||
TbRuleEngineConsumerContext ruleEngineConsumerContext; |
|||
TbRuleEngineQueueConsumerManager consumerManager; |
|||
|
|||
ExecutorService consumersExecutor; |
|||
ScheduledExecutorService scheduler; |
|||
ExecutorService mgmtExecutor; |
|||
|
|||
@BeforeEach |
|||
void setup() throws InterruptedException { |
|||
consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer")); |
|||
scheduler = ThingsBoardExecutors.newSingleThreadScheduledExecutor("tb-rule-engine-consumer-scheduler"); |
|||
mgmtExecutor = ThingsBoardExecutors.newWorkStealingPool(1, "tb-rule-engine-mgmt"); |
|||
|
|||
mainQueue = new Queue(); |
|||
mainQueue.setTenantId(TenantId.SYS_TENANT_ID); |
|||
mainQueue.setName("Main"); |
|||
mainQueue.setTopic("tb_rule_engine.main"); |
|||
mainQueue.setPollInterval(25); |
|||
mainQueue.setPartitions(1); |
|||
mainQueue.setConsumerPerPartition(false); |
|||
mainQueue.setPackProcessingTimeout(2000L); |
|||
|
|||
var submitStrategy = new SubmitStrategy(); |
|||
submitStrategy.setType(SubmitStrategyType.BURST); |
|||
submitStrategy.setBatchSize(1000); |
|||
mainQueue.setSubmitStrategy(submitStrategy); |
|||
|
|||
var processingStrategy = new ProcessingStrategy(); |
|||
processingStrategy.setType(ProcessingStrategyType.SKIP_ALL_FAILURES); |
|||
processingStrategy.setRetries(3); |
|||
processingStrategy.setFailurePercentage(0.0); |
|||
processingStrategy.setPauseBetweenRetries(3); |
|||
processingStrategy.setMaxPauseBetweenRetries(3); |
|||
mainQueue.setProcessingStrategy(processingStrategy); |
|||
|
|||
storage = new DefaultInMemoryStorage(); |
|||
|
|||
consumer = spy(new InMemoryTbQueueConsumer<>(storage, mainQueue.getTopic())); |
|||
given(queueFactory.createToRuleEngineMsgConsumer(eq(mainQueue), isNull())).willReturn(consumer); |
|||
|
|||
ruleEngineConsumerContext = new TbRuleEngineConsumerContext( |
|||
actorContext, statsFactory, new TbRuleEngineSubmitStrategyFactory(), new TbRuleEngineProcessingStrategyFactory(), |
|||
queueFactory, statisticsService, serviceInfoProvider, partitionService, producerProvider, queueAdmin |
|||
); |
|||
ruleEngineConsumerContext.setPollDuration(25); |
|||
ruleEngineConsumerContext.setPackProcessingTimeout(2000); |
|||
ruleEngineConsumerContext.setStatsEnabled(false); // true by default
|
|||
ruleEngineConsumerContext.setPrometheusStatsEnabled(false); |
|||
ruleEngineConsumerContext.setTopicDeletionDelayInSec(15); |
|||
ruleEngineConsumerContext.setMgmtThreadPoolSize(12); |
|||
|
|||
// Tell the (mock) context factory to return (mock) message pack context
|
|||
given(packProcessingContextFactory.create( |
|||
eq(mainQueue.getName()), |
|||
any(TbRuleEngineSubmitStrategy.class), |
|||
eq(false) |
|||
)).willAnswer(invocation -> { |
|||
TbRuleEngineSubmitStrategy realStrategy = invocation.getArgument(1); |
|||
when(packCtx.getPendingMap()).thenAnswer(i -> realStrategy.getPendingMap()); |
|||
when(packCtx.getFailedMap()).thenReturn(new ConcurrentHashMap<>()); |
|||
return packCtx; |
|||
}); |
|||
|
|||
// Tell the (mock) context's await() to return 'false' (always timeout) immediately
|
|||
given(packCtx.await(anyLong(), any(TimeUnit.class))).willReturn(false); |
|||
|
|||
consumerManager = TbRuleEngineQueueConsumerManager.create() |
|||
.ctx(ruleEngineConsumerContext) |
|||
.queueKey(new QueueKey(ServiceType.TB_RULE_ENGINE, mainQueue)) |
|||
.consumerExecutor(consumersExecutor) |
|||
.scheduler(scheduler) |
|||
.taskExecutor(mgmtExecutor) |
|||
.packProcessingContextFactory(packProcessingContextFactory) |
|||
.build(); |
|||
} |
|||
|
|||
@AfterEach |
|||
void destroy() { |
|||
MoreExecutors.shutdownAndAwaitTermination(scheduler, Duration.ofSeconds(30)); |
|||
MoreExecutors.shutdownAndAwaitTermination(mgmtExecutor, Duration.ofSeconds(30)); |
|||
MoreExecutors.shutdownAndAwaitTermination(consumersExecutor, Duration.ofSeconds(30)); |
|||
} |
|||
|
|||
@Test |
|||
void consumerLoopTest_verifyOperationsOrder() throws InterruptedException { |
|||
// Create partition
|
|||
var partition = TopicPartitionInfo.builder() |
|||
.tenantId(TenantId.SYS_TENANT_ID) |
|||
.topic(mainQueue.getTopic()) |
|||
.partition(0) |
|||
.myPartition(true) |
|||
.useInternalPartition(false) |
|||
.build(); |
|||
|
|||
// Put 10k messages to the queue
|
|||
for (int i = 0; i < 10_000; i++) { |
|||
var tbMsg = TbMsg.newMsg() |
|||
.type(TbMsgType.POST_TELEMETRY_REQUEST) |
|||
.originator(deviceId) |
|||
.data("{\"temperature\":123}") |
|||
.metaData(TbMsgMetaData.EMPTY) |
|||
.build(); |
|||
|
|||
var toRuleEngineMsg = TransportProtos.ToRuleEngineMsg.newBuilder() |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTbMsgProto(TbMsg.toProto(tbMsg)) |
|||
.addAllRelationTypes(Set.of("Success")) |
|||
.build(); |
|||
|
|||
storage.put(partition.getFullTopicName(), new TbProtoQueueMsg<>(UUID.randomUUID(), toRuleEngineMsg, new DefaultTbQueueMsgHeaders())); |
|||
} |
|||
|
|||
// Count how many polls were made
|
|||
var totalPolls = new AtomicInteger(0); |
|||
var emptyPolls = new AtomicInteger(0); |
|||
doAnswer(invocation -> { |
|||
totalPolls.incrementAndGet(); |
|||
@SuppressWarnings("unchecked") |
|||
var messages = (List<TbQueueMsg>) invocation.callRealMethod(); |
|||
if (messages.isEmpty()) { |
|||
emptyPolls.incrementAndGet(); |
|||
} |
|||
return messages; |
|||
}).when(consumer).poll(mainQueue.getPollInterval()); |
|||
|
|||
// Count how many commits were made
|
|||
var totalCommits = new AtomicInteger(0); |
|||
doAnswer(invocation -> { |
|||
totalCommits.incrementAndGet(); |
|||
return invocation.callRealMethod(); |
|||
}).when(consumer).commit(); |
|||
|
|||
// Initialize consumer
|
|||
consumerManager.init(mainQueue); |
|||
|
|||
// Assign partition to the consumer
|
|||
consumerManager.update(Set.of(partition)); |
|||
|
|||
// Give some time for the consumer to get all messages
|
|||
await().atMost(Duration.ofSeconds(10L)).until(() -> storage.getLagTotal() == 0); |
|||
|
|||
// Stop consumer
|
|||
consumerManager.stop(); |
|||
consumerManager.awaitStop(); |
|||
|
|||
// Determine number of non-empty consumer iterations made, since polling does not stop immediately after consuming all messages and may do a few empty polls
|
|||
int nonEmptyPolls = totalPolls.get() - emptyPolls.get(); |
|||
|
|||
// Verify that there is 10 polls and 10 matching commits
|
|||
// Each poll consumes 1k messages and queue has 10k total, so that means 10k total msgs / 1k msgs per poll = 10 polls
|
|||
assertThat(nonEmptyPolls).isEqualTo(10).isEqualTo(totalCommits.get()); |
|||
|
|||
// Verify that poll-await-commit cycle happened in order with correct await timeout
|
|||
InOrder inOrder = inOrder(consumer, packCtx); |
|||
for (int i = 0; i < nonEmptyPolls; i++) { |
|||
inOrder.verify(consumer).poll(mainQueue.getPollInterval()); |
|||
inOrder.verify(packCtx).await(mainQueue.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); |
|||
inOrder.verify(consumer).commit(); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,204 @@ |
|||
/** |
|||
* 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.security.system; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.junit.runner.RunWith; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.MockitoJUnitRunner; |
|||
import org.springframework.security.authentication.BadCredentialsException; |
|||
import org.springframework.security.authentication.DisabledException; |
|||
import org.springframework.security.authentication.LockedException; |
|||
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder; |
|||
import org.thingsboard.rule.engine.api.MailService; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.security.UserCredentials; |
|||
import org.thingsboard.server.common.data.security.model.SecuritySettings; |
|||
import org.thingsboard.server.common.data.security.model.UserPasswordPolicy; |
|||
import org.thingsboard.server.dao.audit.AuditLogService; |
|||
import org.thingsboard.server.dao.settings.AdminSettingsService; |
|||
import org.thingsboard.server.dao.settings.SecuritySettingsService; |
|||
import org.thingsboard.server.dao.user.UserService; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.Mockito.never; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@RunWith(MockitoJUnitRunner.class) |
|||
public class DefaultSystemSecurityServiceTest { |
|||
|
|||
@Mock |
|||
private AdminSettingsService adminSettingsService; |
|||
@Mock |
|||
private BCryptPasswordEncoder encoder; |
|||
@Mock |
|||
private UserService userService; |
|||
@Mock |
|||
private MailService mailService; |
|||
@Mock |
|||
private AuditLogService auditLogService; |
|||
@Mock |
|||
private SecuritySettingsService securitySettingsService; |
|||
|
|||
private DefaultSystemSecurityService systemSecurityService; |
|||
|
|||
private TenantId tenantId; |
|||
private UserId userId; |
|||
private UserCredentials userCredentials; |
|||
private SecuritySettings securitySettings; |
|||
private String username; |
|||
private String password; |
|||
private String encodedPassword; |
|||
|
|||
@Before |
|||
public void setUp() { |
|||
systemSecurityService = new DefaultSystemSecurityService(adminSettingsService, encoder, userService, mailService, auditLogService, securitySettingsService); |
|||
|
|||
tenantId = TenantId.fromUUID(UUID.randomUUID()); |
|||
userId = new UserId(UUID.randomUUID()); |
|||
username = "tenant@example.com"; |
|||
password = "correctPassword"; |
|||
encodedPassword = "$2a$10$encodedPasswordHash"; |
|||
|
|||
userCredentials = new UserCredentials(); |
|||
userCredentials.setUserId(userId); |
|||
userCredentials.setEnabled(true); |
|||
userCredentials.setPassword(encodedPassword); |
|||
userCredentials.setCreatedTime(System.currentTimeMillis()); |
|||
|
|||
securitySettings = new SecuritySettings(); |
|||
securitySettings.setMaxFailedLoginAttempts(5); |
|||
securitySettings.setPasswordPolicy(new UserPasswordPolicy()); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_successfulLogin() { |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(true); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password); |
|||
|
|||
verify(encoder).matches(password, encodedPassword); |
|||
verify(userService).resetFailedLoginAttempts(tenantId, userId); |
|||
verify(userService, never()).increaseFailedLoginAttempts(any(), any()); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_wrongPassword_incrementsFailedAttempts() { |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(false); |
|||
when(userService.increaseFailedLoginAttempts(tenantId, userId)).thenReturn(3); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(BadCredentialsException.class) |
|||
.hasMessageContaining("Authentication Failed"); |
|||
|
|||
verify(userService).increaseFailedLoginAttempts(tenantId, userId); |
|||
verify(userService, never()).setUserCredentialsEnabled(any(), any(), eq(false)); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_wrongPassword_accountLocked() { |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(false); |
|||
when(userService.increaseFailedLoginAttempts(tenantId, userId)).thenReturn(6); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(LockedException.class) |
|||
.hasMessageContaining("locked due to security policy"); |
|||
|
|||
verify(userService).increaseFailedLoginAttempts(tenantId, userId); |
|||
verify(userService).setUserCredentialsEnabled(TenantId.SYS_TENANT_ID, userId, false); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_wrongPassword_exactlyAtThreshold_noLock() { |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(false); |
|||
when(userService.increaseFailedLoginAttempts(tenantId, userId)).thenReturn(5); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(BadCredentialsException.class) |
|||
.hasMessageContaining("Authentication Failed"); |
|||
|
|||
verify(userService).increaseFailedLoginAttempts(tenantId, userId); |
|||
verify(userService, never()).setUserCredentialsEnabled(any(), any(), eq(false)); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_correctPassword_disabledUser_throwsDisabledException() { |
|||
userCredentials.setEnabled(false); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(DisabledException.class) |
|||
.hasMessage("User is not active"); |
|||
|
|||
verify(encoder, never()).matches(any(), any()); |
|||
verify(userService, never()).increaseFailedLoginAttempts(any(), any()); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_wrongPassword_maxAttemptsDisabled() { |
|||
securitySettings.setMaxFailedLoginAttempts(null); |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(false); |
|||
when(userService.increaseFailedLoginAttempts(tenantId, userId)).thenReturn(100); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(BadCredentialsException.class); |
|||
|
|||
verify(userService).increaseFailedLoginAttempts(tenantId, userId); |
|||
verify(userService, never()).setUserCredentialsEnabled(any(), any(), eq(false)); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_wrongPassword_maxAttemptsSetToZero() { |
|||
securitySettings.setMaxFailedLoginAttempts(0); |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(false); |
|||
when(userService.increaseFailedLoginAttempts(tenantId, userId)).thenReturn(100); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(BadCredentialsException.class); |
|||
|
|||
verify(userService).increaseFailedLoginAttempts(tenantId, userId); |
|||
verify(userService, never()).setUserCredentialsEnabled(any(), any(), eq(false)); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidateUserCredentials_wrongPassword_withNotificationEmail() throws ThingsboardException { |
|||
String notificationEmail = "admin@example.com"; |
|||
securitySettings.setUserLockoutNotificationEmail(notificationEmail); |
|||
when(encoder.matches(password, encodedPassword)).thenReturn(false); |
|||
when(userService.increaseFailedLoginAttempts(tenantId, userId)).thenReturn(6); |
|||
when(securitySettingsService.getSecuritySettings()).thenReturn(securitySettings); |
|||
|
|||
assertThatThrownBy(() -> systemSecurityService.validateUserCredentials(tenantId, userCredentials, username, password)) |
|||
.isInstanceOf(LockedException.class); |
|||
|
|||
verify(userService).setUserCredentialsEnabled(TenantId.SYS_TENANT_ID, userId, false); |
|||
verify(mailService).sendAccountLockoutEmail(eq(username), eq(notificationEmail), eq(5)); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,42 @@ |
|||
/** |
|||
* 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.utils; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.io.IOException; |
|||
import java.net.DatagramSocket; |
|||
import java.net.SocketException; |
|||
|
|||
@Slf4j |
|||
public class PortFinder { |
|||
public static int findAvailableUdpPort() { |
|||
try (DatagramSocket socket = new DatagramSocket(0)) { |
|||
return socket.getLocalPort(); |
|||
} catch (SocketException e) { |
|||
throw new IllegalStateException("No available UDP ports found", e); |
|||
} |
|||
} |
|||
|
|||
public static boolean isUDPPortAvailable(int port) { |
|||
try (DatagramSocket socket = new DatagramSocket(port)) { |
|||
return true; |
|||
} catch (IOException e) { |
|||
log.debug("Failed to open UDP port {}", port, e); |
|||
return false; |
|||
} |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue