2 * ============LICENSE_START=======================================================
3 * PNF-REGISTRATION-HANDLER
4 * ================================================================================
5 * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.dcaegen2.services.prh.tasks;
23 import org.apache.kafka.clients.consumer.ConsumerRecord;
24 import org.apache.kafka.common.header.Headers;
25 import org.apache.kafka.common.header.internals.RecordHeaders;
26 import org.apache.kafka.common.record.TimestampType;
27 import org.junit.jupiter.api.Test;
28 import org.junit.jupiter.api.extension.ExtendWith;
29 import org.mockito.InjectMocks;
30 import org.mockito.Mock;
31 import org.mockito.junit.jupiter.MockitoExtension;
32 import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
33 import org.onap.dcaegen2.services.prh.adapter.aai.api.ImmutableConsumerDmaapModel;
34 import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
35 import org.onap.dcaegen2.services.prh.tasks.commit.EpochDateTimeConversion;
36 import org.onap.dcaegen2.services.prh.tasks.commit.KafkaConsumerTaskImpl;
37 import org.springframework.boot.configurationprocessor.json.JSONException;
38 import org.springframework.kafka.support.Acknowledgment;
39 import reactor.core.publisher.Flux;
41 import java.util.ArrayList;
42 import java.util.List;
44 import static org.mockito.Mockito.when;
46 @ExtendWith(MockitoExtension.class)
47 public class KafkaConsumerTaskImplTest {
50 private Acknowledgment acknowledgment;
53 private DmaapConsumerJsonParser dmaapConsumerJsonParser;
56 private EpochDateTimeConversion epochDateTimeConversion;
59 private KafkaConsumerTaskImpl kafkaConsumerTask;
62 public void onMessageTest(){
63 List<ConsumerRecord<String, String>> list = new ArrayList<>();
64 TimestampType timestampType = null;
65 Headers headers = new RecordHeaders();
66 epochDateTimeConversion.setDaysForRecords("3");
67 ConsumerRecord<String, String> records = new ConsumerRecord<>
68 ("test-topic", 1, 1l, 0l, timestampType, 1, 1, "test-key", "test-value", headers
71 kafkaConsumerTask.onMessage(list, acknowledgment);
75 public void commitOffsetTest(){
76 kafkaConsumerTask.commitOffset();
80 public void executeTest() throws JSONException {
81 List<String> jsonEvent = new ArrayList<>();
82 ConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder().correlationId("123").build();
83 when(dmaapConsumerJsonParser.getConsumerDmaapModelFromKafkaConsumerRecord(jsonEvent)).thenReturn(Flux.just(consumerDmaapModel));
84 kafkaConsumerTask.execute();