Browse Source

added tests for entry

pull/14141/head
IrynaMatveieva 1 year ago
parent
commit
6a307041b6
  1. 4
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 4
      application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
  3. 24
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java
  4. 3
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  5. 7
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java
  6. 30
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggSingleEntityArgumentEntry.java
  7. 112
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggArgumentEntryTest.java
  8. 101
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggSingleEntityArgumentEntryTest.java
  9. 2
      common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java
  10. 4
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java
  11. 13
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java
  12. 15
      dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java

4
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@ -67,7 +67,6 @@ import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@ -686,9 +685,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
Map.Entry::getKey,
argEntry -> new AggSingleEntityArgumentEntry(entityId, argEntry.getValue())
));
} else {
fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true));
}
fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true));
return fetchedArgs;
}

4
application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java

@ -332,7 +332,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
var attributeOptFuture = attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey());
return Futures.transform(attributeOptFuture, attrOpt -> {
log.debug("[{}][{}] Fetched attribute for key {}: {}", tenantId, entityId, argument.getRefEntityKey(), attrOpt);
AttributeKvEntry attributeKvEntry = attrOpt.orElseGet(() -> new BaseAttributeKvEntry(createDefaultKvEntry(argument), defaultLastUpdateTs, 0L));
AttributeKvEntry attributeKvEntry = attrOpt.orElseGet(() -> new BaseAttributeKvEntry(createDefaultKvEntry(argument), defaultLastUpdateTs, SingleValueArgumentEntry.DEFAULT_VERSION));
return transformAggSingleArgument(entityId, Optional.of(attributeKvEntry));
}, calculatedFieldCallbackExecutor);
}
@ -344,7 +344,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
timeseriesService.findLatest(tenantId, entityId, key),
result -> {
log.debug("[{}][{}] Fetched latest timeseries {}: {}", tenantId, entityId, key, result);
Optional<TsKvEntry> tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), 0L)));
Optional<TsKvEntry> tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), SingleValueArgumentEntry.DEFAULT_VERSION)));
return transformAggSingleArgument(entityId, tsKvEntry);
}, calculatedFieldCallbackExecutor);
}

24
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java

@ -37,8 +37,9 @@ import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntityRelationPathQuery;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationPathLevel;
import org.thingsboard.server.common.data.relation.RelationTypeGroup;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.dao.relation.RelationService;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
@ -191,20 +192,15 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
for (CalculatedFieldCtx cfCtx : cfCtxs) {
if (cfCtx.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration aggConfig) {
RelationPathLevel relation = aggConfig.getRelation();
switch (relation.direction()) {
case FROM -> {
List<EntityRelation> byToAndType = relationService.findByToAndType(tenantId, entityId, relation.relationType(), RelationTypeGroup.COMMON);
if (!byToAndType.isEmpty()) {
return true;
}
}
case TO -> {
List<EntityRelation> byFromAndType = relationService.findByFromAndType(tenantId, entityId, relation.relationType(), RelationTypeGroup.COMMON);
if (!byFromAndType.isEmpty()) {
return true;
}
}
EntitySearchDirection inverseDirection = switch (relation.direction()) {
case FROM -> EntitySearchDirection.TO;
case TO -> EntitySearchDirection.FROM;
};
RelationPathLevel inverseRelation = new RelationPathLevel(inverseDirection, relation.relationType());
List<EntityRelation> byRelationPathQuery = relationService.findByRelationPathQuery(tenantId, new EntityRelationPathQuery(entityId, List.of(inverseRelation)));
if (!byRelationPathQuery.isEmpty()) {
return true;
}
}
}

3
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java

@ -22,6 +22,7 @@ import org.thingsboard.server.actors.TbActorRef;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry;
import org.thingsboard.server.utils.CalculatedFieldUtils;
import java.io.Closeable;
@ -73,7 +74,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
ArgumentEntry existingEntry = arguments.get(key);
boolean entryUpdated;
if (existingEntry == null || newEntry.isForceResetPrevious()) {
if (existingEntry == null || !(newEntry instanceof AggSingleEntityArgumentEntry) && newEntry.isForceResetPrevious()) {
validateNewEntry(key, newEntry);
arguments.put(key, newEntry);
entryUpdated = true;

7
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java

@ -52,7 +52,12 @@ public class AggArgumentEntry implements ArgumentEntry {
if (aggSingleEntityArgumentEntry.isDeleted()) {
aggInputs.remove(aggSingleEntityArgumentEntry.getEntityId());
} else {
aggInputs.put(aggSingleEntityArgumentEntry.getEntityId(), aggSingleEntityArgumentEntry);
ArgumentEntry argumentEntry = aggInputs.get(aggSingleEntityArgumentEntry.getEntityId());
if (argumentEntry != null) {
argumentEntry.updateEntry(aggSingleEntityArgumentEntry);
} else {
aggInputs.put(aggSingleEntityArgumentEntry.getEntityId(), aggSingleEntityArgumentEntry);
}
}
return true;
} else {

30
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggSingleEntityArgumentEntry.java

@ -62,25 +62,35 @@ public class AggSingleEntityArgumentEntry extends SingleValueArgumentEntry {
@Override
public boolean updateEntry(ArgumentEntry entry) {
if (entry instanceof AggSingleEntityArgumentEntry singleValueEntry) {
if (singleValueEntry.getTs() <= ts) {
return false;
if (entry instanceof AggSingleEntityArgumentEntry aggSingleEntityEntry) {
if (aggSingleEntityEntry.isForceResetPrevious()) {
return applyNewEntry(aggSingleEntityEntry);
}
Long newVersion = singleValueEntry.getVersion();
if (aggSingleEntityEntry.getTs() < this.ts) {
if (!isDefaultValue()) {
return false;
}
}
Long newVersion = aggSingleEntityEntry.getVersion();
if (newVersion == null || this.version == null || newVersion > this.version) {
this.ts = singleValueEntry.getTs();
this.version = newVersion;
this.kvEntryValue = singleValueEntry.getKvEntryValue();
this.entityId = singleValueEntry.getEntityId();
return true;
return applyNewEntry(aggSingleEntityEntry);
}
} else {
throw new IllegalArgumentException("Unsupported argument entry type for single value argument entry: " + entry.getType());
throw new IllegalArgumentException("Unsupported argument entry type for aggregation single entity argument entry: " + entry.getType());
}
return false;
}
private boolean applyNewEntry(AggSingleEntityArgumentEntry entry) {
this.ts = entry.getTs();
this.version = entry.getVersion();
this.kvEntryValue = entry.getKvEntryValue();
this.entityId = entry.getEntityId();
return true;
}
@Override
public ArgumentEntryType getType() {
return ArgumentEntryType.AGGREGATE_LATEST_SINGLE;

112
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggArgumentEntryTest.java

@ -0,0 +1,112 @@
/**
* 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.cf.ctx.state;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.AggArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class AggArgumentEntryTest {
private AggArgumentEntry entry;
private final DeviceId device1 = new DeviceId(UUID.fromString("1984e5f4-9ff0-4187-84ae-e4438bba4c8a"));
private final DeviceId device2 = new DeviceId(UUID.fromString("937fc062-1a9d-438f-aa22-55a93fc908b7"));
private final long ts = System.currentTimeMillis();
@BeforeEach
void setUp() {
Map<EntityId, ArgumentEntry> aggInputs = new HashMap<>();
aggInputs.put(device1, new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 12L), 1L)));
aggInputs.put(device2, new AggSingleEntityArgumentEntry(device2, new BasicTsKvEntry(ts - 150, new LongDataEntry("key", 16L), 6L)));
entry = new AggArgumentEntry(aggInputs, false);
}
@Test
void testUpdateEntryWhenNotAggEntryPassed() {
assertThatThrownBy(() -> entry.updateEntry(new TsRollingArgumentEntry(5, 30000L)))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Unsupported argument entry type for aggregation argument entry: " + ArgumentEntryType.TS_ROLLING);
}
@Test
void testUpdateEntryWhenAggArgumentEntryPasser() {
DeviceId device3 = new DeviceId(UUID.randomUUID());
DeviceId device4 = new DeviceId(UUID.randomUUID());
AggArgumentEntry aggArgumentEntry = new AggArgumentEntry(Map.of(
device3, new AggSingleEntityArgumentEntry(device3, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 16L), 13L)),
device4, new AggSingleEntityArgumentEntry(device4, new BasicTsKvEntry(ts - 60, new LongDataEntry("key", 23L), 7L))
), false);
assertThat(entry.updateEntry(aggArgumentEntry)).isTrue();
Map<EntityId, ArgumentEntry> aggInputs = entry.getAggInputs();
assertThat(aggInputs.size()).isEqualTo(4);
assertThat(aggInputs.get(device3)).isEqualTo(aggArgumentEntry.getAggInputs().get(device3));
assertThat(aggInputs.get(device4)).isEqualTo(aggArgumentEntry.getAggInputs().get(device4));
}
@Test
void testUpdateEntryWhenAggSingleEntityArgumentEntryPassedAndNoEntriesById() {
DeviceId device3 = new DeviceId(UUID.randomUUID());
AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device3, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 18L), 10L));
assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue();
Map<EntityId, ArgumentEntry> aggInputs = entry.getAggInputs();
assertThat(aggInputs.size()).isEqualTo(3);
assertThat(aggInputs.get(device3)).isEqualTo(singleEntityArgumentEntry);
}
@Test
void testUpdateEntryWhenAggSingleEntityArgumentEntryPassedAndEntryByIdExist() {
AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device2, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 18L), 10L));
assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue();
Map<EntityId, ArgumentEntry> aggInputs = entry.getAggInputs();
assertThat(aggInputs.size()).isEqualTo(2);
assertThat(aggInputs.get(device2)).isEqualTo(singleEntityArgumentEntry);
}
@Test
void testUpdateEntryWhenDeletedAggSingleEntityArgumentEntryPassed() {
AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device2, true);
assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue();
Map<EntityId, ArgumentEntry> aggInputs = entry.getAggInputs();
assertThat(aggInputs.size()).isEqualTo(1);
assertThat(aggInputs.get(device2)).isNull();
}
}

101
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggSingleEntityArgumentEntryTest.java

@ -0,0 +1,101 @@
/**
* 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.cf.ctx.state;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class AggSingleEntityArgumentEntryTest {
private AggSingleEntityArgumentEntry entry;
private final DeviceId device1 = new DeviceId(UUID.fromString("1984e5f4-9ff0-4187-84ae-e4438bba4c8a"));
private final long ts = System.currentTimeMillis();
@BeforeEach
void setUp() {
entry = new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 12L), 22L));
}
@Test
void testUpdateEntryWhenNotAggEntryPassed() {
assertThatThrownBy(() -> entry.updateEntry(new TsRollingArgumentEntry(5, 30000L)))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Unsupported argument entry type for aggregation single entity argument entry: " + ArgumentEntryType.TS_ROLLING);
}
@Test
void testUpdateEntryWhenResetPrevious() {
AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 18L), 100L));
singleEntityArgumentEntry.setForceResetPrevious(true);
assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue();
assertThat(entry.getTs()).isEqualTo(singleEntityArgumentEntry.getTs());
assertThat(entry.getKvEntryValue()).isEqualTo(singleEntityArgumentEntry.getKvEntryValue());
assertThat(entry.getVersion()).isEqualTo(singleEntityArgumentEntry.getVersion());
}
@Test
void testUpdateEntryWithTheSameTsAndVersion() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 19L), 22L)))).isFalse();
}
@Test
void testUpdateEntryWithTheSameTsAndDifferentVersion() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 134L), 23L)))).isTrue();
}
@Test
void testUpdateEntryWhenNewVersionIsNull() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 56L), null)))).isTrue();
assertThat(entry.getValue()).isEqualTo(56L);
assertThat(entry.getVersion()).isNull();
}
@Test
void testUpdateEntryWhenNewVersionIsGreaterThanCurrent() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 76L), 23L)))).isTrue();
assertThat(entry.getValue()).isEqualTo(76L);
assertThat(entry.getVersion()).isEqualTo(23);
}
@Test
void testUpdateEntryWhenNewVersionIsLessThanCurrent() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 11L), 20L)))).isFalse();
}
@Test
void testUpdateEntryWhenValueWasNotChanged() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 18L), 45L)))).isTrue();
}
@Test
void testUpdateEntryWithOldTs() {
assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 150, new LongDataEntry("key", 155L), 45L)))).isFalse();
}
}

2
common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java

@ -86,6 +86,8 @@ public interface RelationService {
ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
// TODO: This method may be useful for some validations in the future
// ListenableFuture<Boolean> checkRecursiveRelation(EntityId from, EntityId to);

4
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java

@ -45,4 +45,8 @@ public class Argument {
return hasDynamicSource() && refDynamicSourceConfiguration.getType() == CFArgumentDynamicSourceType.CURRENT_OWNER;
}
public boolean hasTsRollingArgument() {
return ArgumentType.TS_ROLLING.equals(refEntityKey.getType());
}
}

13
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java

@ -41,6 +41,19 @@ public class LatestValuesAggregationCalculatedFieldConfiguration implements Argu
@Override
public void validate() {
if (relation == null) {
throw new IllegalArgumentException("Relation must be specified!");
}
relation.validate();
if (arguments.containsKey("ctx")) {
throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used.");
}
if (arguments.values().stream().anyMatch(Argument::hasTsRollingArgument)) {
throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support TS_ROLLING arguments.");
}
if (metrics.isEmpty()) {
throw new IllegalArgumentException("Latest value aggregation calculated field must have at least one metric.");
}
}
}

15
dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java

@ -514,6 +514,21 @@ public class BaseRelationService implements RelationService {
return executor.submit(() -> relationDao.findByRelationPathQuery(tenantId, relationPathQuery));
}
@Override
public List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery) {
log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery);
validateId(tenantId, id -> "Invalid tenant id: " + id);
validate(relationPathQuery);
if (relationPathQuery.levels().size() == 1) {
RelationPathLevel relationPathLevel = relationPathQuery.levels().get(0);
return switch (relationPathLevel.direction()) {
case FROM -> findByFromAndType(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
case TO -> findByToAndType(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
};
}
return relationDao.findByRelationPathQuery(tenantId, relationPathQuery);
}
private void validate(EntityRelationPathQuery relationPathQuery) {
validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id);
List<RelationPathLevel> levels = relationPathQuery.levels();

Loading…
Cancel
Save