catalog-be/src/test/java/org/openecomp/sdc/be/components/kafka/KafkaHandlerTest.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2022-2023 Nordix Foundation. 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20 package org.openecomp.sdc.be.components.kafka;
  21
  22 import static org.junit.jupiter.api.Assertions.assertEquals;
  23 import static org.junit.jupiter.api.Assertions.assertFalse;
  24 import static org.junit.jupiter.api.Assertions.assertTrue;
  25 import static org.mockito.ArgumentMatchers.any;
  26 import static org.mockito.Mockito.doThrow;
  27 import static org.mockito.Mockito.when;
  28
  29 import com.google.gson.JsonSyntaxException;
  30 import fj.data.Either;
  31 import java.util.ArrayList;
  32 import java.util.List;
  33 import org.apache.kafka.common.KafkaException;
  34 import org.junit.jupiter.api.Test;
  35 import org.junit.jupiter.api.extension.ExtendWith;
  36 import org.mockito.Mock;
  37 import org.mockito.junit.jupiter.MockitoExtension;
  38 import org.openecomp.sdc.be.components.distribution.engine.CambriaErrorResponse;
  39 import org.openecomp.sdc.be.components.distribution.engine.INotificationData;
  40 import org.openecomp.sdc.be.components.distribution.engine.NotificationDataImpl;
  41 import org.openecomp.sdc.be.distribution.api.client.CambriaOperationStatus;
  42
  43 @ExtendWith(MockitoExtension.class)
  44 public class KafkaHandlerTest {
  45
  46     @Mock
  47     private SdcKafkaConsumer mockSdcKafkaConsumer;
  48
  49     @Mock
  50     private SdcKafkaProducer mockSdcKafkaProducer;
  51
  52     @Test
  53     public void testIsKafkaActiveTrue(){
  54         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
  55         assertTrue(kafkaHandler.isKafkaActive());
  56     }
  57
  58     @Test
  59     public void testIsKafkaActiveFalse(){
  60         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
  61         kafkaHandler.setKafkaActive(false);
  62         assertFalse(kafkaHandler.isKafkaActive());
  63     }
  64
  65     @Test
  66     public void testFetchFromTopicSuccess(){
  67         String testTopic = "testTopic";
  68         List<String> mockedReturnedMessages = new ArrayList<>();
  69         mockedReturnedMessages.add("message1");
  70         mockedReturnedMessages.add("message2");
  71         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
  72         when(mockSdcKafkaConsumer.poll(any())).thenReturn(mockedReturnedMessages);
  73         Either<Iterable<String>, CambriaErrorResponse> response = kafkaHandler.fetchFromTopic(testTopic);
  74         Iterable<String> actualReturnedMessages = response.left().value();
  75         assertTrue(response.isLeft());
  76         assertEquals(actualReturnedMessages, mockedReturnedMessages);
  77     }
  78
  79     @Test
  80     public void testFetchFromTopicFail(){
  81         String testTopic = "testTopic";
  82         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
  83         when(mockSdcKafkaConsumer.poll(any())).thenThrow(new KafkaException());
  84         Either<Iterable<String>, CambriaErrorResponse> response = kafkaHandler.fetchFromTopic(testTopic);
  85         CambriaErrorResponse responseValue = response.right().value();
  86         assertTrue(response.isRight());
  87         assertEquals(responseValue.getOperationStatus(), CambriaOperationStatus.INTERNAL_SERVER_ERROR);
  88     }
  89
  90     @Test
  91     public void testSendNotificationSuccess(){
  92         String testTopic = "testTopic";
  93         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
  94         INotificationData testData = new NotificationDataImpl();
  95         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
  96         assertEquals(response.getOperationStatus(), CambriaOperationStatus.OK);
  97         assertEquals(response.getHttpCode(), 200);
  98     }
  99
 100     @Test
 101     public void testSendNotificationKafkaException(){
 102         String testTopic = "testTopic";
 103         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
 104         INotificationData testData = new NotificationDataImpl();
 105         doThrow(KafkaException.class).when(mockSdcKafkaProducer).send(any(), any());
 106         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
 107         assertEquals(response.getOperationStatus(), CambriaOperationStatus.INTERNAL_SERVER_ERROR);
 108         assertEquals(response.getHttpCode(), 500);
 109     }
 110
 111     @Test
 112     public void testSendNotificationJsonSyntaxException(){
 113         String testTopic = "testTopic";
 114         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
 115         INotificationData testData = new NotificationDataImpl();
 116         doThrow(JsonSyntaxException.class).when(mockSdcKafkaProducer).send(any(), any());
 117         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
 118         assertEquals(response.getOperationStatus(), CambriaOperationStatus.INTERNAL_SERVER_ERROR);
 119         assertEquals(response.getHttpCode(), 500);
 120     }
 121
 122     @Test
 123     public void testSendNotificationFlushException(){
 124         String testTopic = "testTopic";
 125         KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
 126         INotificationData testData = new NotificationDataImpl();
 127         doThrow(KafkaException.class).when(mockSdcKafkaProducer).flush();
 128         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
 129         assertEquals(response.getOperationStatus(), CambriaOperationStatus.INTERNAL_SERVER_ERROR);
 130         assertEquals(response.getHttpCode(), 500);
 131     }
 132 }