From 61b3ff91485571c24834b31c6ee7efc7ab1d0243 Mon Sep 17 00:00:00 2001 From: Rupali Shirode Date: Thu, 7 Dec 2023 16:39:19 +0530 Subject: [SO] Remove DMaap Dependency in SO-bpmn-infra Remove DMaap Dependency in SO-bpmn-infra Issue-ID: SO-4122 Change-Id: I8fbe5761430c21b3f49b31a45ede095fdb72628f Signed-off-by: Rupali Shirode --- .../so/client/kafka/KafkaConsumerImplTest.java | 51 ++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 common/src/test/java/org/onap/so/client/kafka/KafkaConsumerImplTest.java (limited to 'common/src/test/java/org/onap') diff --git a/common/src/test/java/org/onap/so/client/kafka/KafkaConsumerImplTest.java b/common/src/test/java/org/onap/so/client/kafka/KafkaConsumerImplTest.java new file mode 100644 index 0000000000..d71e562b64 --- /dev/null +++ b/common/src/test/java/org/onap/so/client/kafka/KafkaConsumerImplTest.java @@ -0,0 +1,51 @@ +package org.onap.so.client.kafka; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.MockConsumer; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; +import org.apache.kafka.common.TopicPartition; +import org.junit.Before; +import org.junit.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import uk.org.webcompere.systemstubs.environment.EnvironmentVariables; +import uk.org.webcompere.systemstubs.jupiter.SystemStub; +import uk.org.webcompere.systemstubs.jupiter.SystemStubsExtension; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import static org.assertj.core.api.Assertions.assertThat; + +@ExtendWith(SystemStubsExtension.class) +public class KafkaConsumerImplTest { + private KafkaConsumerImpl consumer; + private static MockConsumer mockConsumer; + @SystemStub + EnvironmentVariables environmentVariables = new EnvironmentVariables(); + + @Before + public void setup() { + environmentVariables.set("JAAS_CONFIG", "jaas.config"); + mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); + configureMockConsumer(); + } + + @Test + public void consumerShouldConsumeMessages() throws Exception { + consumer = new KafkaConsumerImpl("localhost:9092"); + consumer.setConsumer(mockConsumer); + List response = consumer.get("TOPIC", "CG1", "C1"); + assertThat(response).contains("I", "like", "pizza"); + } + + private void configureMockConsumer() { + mockConsumer.assign(Arrays.asList(new TopicPartition("TOPIC", 0))); + + HashMap beginningOffsets = new HashMap<>(); + beginningOffsets.put(new TopicPartition("TOPIC", 0), 0L); + mockConsumer.updateBeginningOffsets(beginningOffsets); + mockConsumer.addRecord(new ConsumerRecord("TOPIC", 0, 0L, "key", "I")); + mockConsumer.addRecord(new ConsumerRecord("TOPIC", 0, 1L, "key", "like")); + mockConsumer.addRecord(new ConsumerRecord("TOPIC", 0, 2L, "key", "pizza")); + + } +} -- cgit 1.2.3-korg