TbRuleEngineQueueConsumerManagerTest fixed, bumped to JUnit5, parametrized test added

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<TbProtoQueueMsg<ToRuleEngineMsg>> 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<TbRuleEngineQueueFactory> 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<TopicPartitionInfo> 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