From e638e60340bc76305420e14b87599ad3278472ee Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Tue, 7 May 2024 12:20:35 +0200 Subject: [PATCH] TbRuleEngineQueueConsumerManagerTest fixed, bumped to JUnit5, parametrized test added --- .../TbRuleEngineQueueConsumerManagerTest.java | 50 ++++++++++++++----- 1 file changed, 37 insertions(+), 13 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java index 8e0b294241..33b79cc82c 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java @@ -18,13 +18,15 @@ package org.thingsboard.server.service.queue.ruleengine; import lombok.Getter; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; +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.ValueSource; import org.mockito.Mock; import org.mockito.Mockito; -import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; import org.testcontainers.shaded.org.apache.commons.lang3.RandomUtils; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.actors.ActorSystemContext; @@ -51,6 +53,8 @@ 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.provider.KafkaMonolithQueueFactory; +import org.thingsboard.server.queue.provider.KafkaTbRuleEngineQueueFactory; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; @@ -72,10 +76,12 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.willCallRealMethod; import static org.mockito.Mockito.after; import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.atLeastOnce; @@ -91,7 +97,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; @Slf4j -@RunWith(MockitoJUnitRunner.class) +@MockitoSettings(strictness = Strictness.LENIENT) public class TbRuleEngineQueueConsumerManagerTest { @Mock @@ -108,6 +114,7 @@ public class TbRuleEngineQueueConsumerManagerTest { private PartitionService partitionService; @Mock private TbQueueProducerProvider producerProvider; + @Mock private TbQueueProducer> ruleEngineMsgProducer; @Mock private TbQueueAdmin queueAdmin; @@ -121,7 +128,7 @@ public class TbRuleEngineQueueConsumerManagerTest { private AtomicInteger totalConsumedMsgs; private AtomicInteger totalProcessedMsgs; - @Before + @BeforeEach public void beforeEach() { ruleEngineConsumerContext = new TbRuleEngineConsumerContext( actorContext, statsFactory, spy(new TbRuleEngineSubmitStrategyFactory()), @@ -139,7 +146,7 @@ public class TbRuleEngineQueueConsumerManagerTest { log.trace("totalProcessedMsgs = {}", totalProcessedMsgs); return null; }).when(actorContext).tell(any()); - ruleEngineMsgProducer = mock(TbQueueProducer.class); + when(producerProvider.getRuleEngineMsgProducer()).thenReturn(ruleEngineMsgProducer); ruleEngineConsumerContext.setMgmtThreadPoolSize(2); ruleEngineConsumerContext.setTopicDeletionDelayInSec(5); @@ -171,13 +178,13 @@ public class TbRuleEngineQueueConsumerManagerTest { } consumers.add(consumer); return consumer; - }).when(queueFactory).createToRuleEngineMsgConsumer(any()); + }).when(queueFactory).createToRuleEngineMsgConsumer(any(), any()); QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queue); consumerManager = new TbRuleEngineQueueConsumerManager(ruleEngineConsumerContext, queueKey); } - @After + @AfterEach public void afterEach() { consumerManager.stop(); consumerManager.awaitStop(); @@ -192,6 +199,20 @@ public class TbRuleEngineQueueConsumerManagerTest { } } + @ParameterizedTest + @ValueSource(classes = {KafkaMonolithQueueFactory.class, KafkaTbRuleEngineQueueFactory.class}) + public void testUnsupported_createToRuleEngineMsgConsumer_KafkaTbRuleEngineQueueFactory(Class factoryClass) { + // obsolete, but need to pass the afterEach + queue.setConsumerPerPartition(false); + consumerManager.init(queue); + + var factory = mock(factoryClass); + willCallRealMethod().given(factory).createToRuleEngineMsgConsumer(any()); + assertThatThrownBy(() -> factory.createToRuleEngineMsgConsumer(mock(Queue.class))) + .isInstanceOf(UnsupportedOperationException.class); + + } + @Test public void testInit_consumerPerPartition() { queue.setConsumerPerPartition(true); @@ -244,7 +265,8 @@ public class TbRuleEngineQueueConsumerManagerTest { Set partitions = Collections.emptySet(); consumerManager.update(partitions); - verify(queueFactory, after(1000).never()).createToRuleEngineMsgConsumer(any()); + verify(queueFactory, after(1000).never()).createToRuleEngineMsgConsumer(any(), any()); + verify(queueFactory, never()).createToRuleEngineMsgConsumer(any()); partitions = createTpis(1); consumerManager.update(partitions); @@ -276,7 +298,8 @@ public class TbRuleEngineQueueConsumerManagerTest { ruleEngineConsumerContext.setReady(true); consumerManager.update(Collections.emptySet()); - verify(queueFactory, after(1000).never()).createToRuleEngineMsgConsumer(any()); + verify(queueFactory, after(1000).never()).createToRuleEngineMsgConsumer(any(), any()); + verify(queueFactory, never()).createToRuleEngineMsgConsumer(any()); consumerManager.update(createTpis(1)); TestConsumer consumer1 = getConsumer(1); @@ -423,7 +446,8 @@ public class TbRuleEngineQueueConsumerManagerTest { consumerManager.update(createTpis(1)); TestConsumer consumer = getConsumer(1); verifySubscribedAndLaunched(consumer, 1); - verify(queueFactory, times(1)).createToRuleEngineMsgConsumer(any()); + verify(queueFactory, times(1)).createToRuleEngineMsgConsumer(any(), any()); + verify(queueFactory, never()).createToRuleEngineMsgConsumer(any()); consumerManager.stop(); consumerManager.update(createTpis(1, 2, 3, 4)); // to check that no new tasks after stop are processed