gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019 Samsung. All rights reserved.
   4  *  Modifications Copyright (C) 2020 Nordix Foundation
   5  *  Modifications Copyright (C) 2022 Bell Canada. 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  *
  19  * SPDX-License-Identifier: Apache-2.0
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.policy.apex.plugins.event.carrier.kafka;
  24
  25 import static org.assertj.core.api.Assertions.assertThatCode;
  26 import static org.junit.Assert.assertEquals;
  27 import static org.junit.Assert.assertNotNull;
  28 import static org.junit.Assert.assertNull;
  29
  30 import org.junit.Before;
  31 import org.junit.Test;
  32 import org.onap.policy.apex.service.engine.event.ApexEventException;
  33 import org.onap.policy.apex.service.engine.event.ApexEventProducer;
  34 import org.onap.policy.apex.service.engine.event.ApexEventReceiver;
  35 import org.onap.policy.apex.service.engine.event.PeeredReference;
  36 import org.onap.policy.apex.service.parameters.carriertechnology.CarrierTechnologyParameters;
  37 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  38 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
  39
  40 public class ApexKafkaConsumerTest {
  41     ApexKafkaConsumer apexKafkaConsumer = null;
  42     ApexKafkaConsumer apexKafkaConsumer2 = null;
  43     EventHandlerParameters consumerParameters = null;
  44     EventHandlerParameters consumerParameters2 = null;
  45     ApexEventReceiver incomingEventReceiver = null;
  46     ApexEventProducer apexKafkaProducer = null;
  47     KafkaCarrierTechnologyParameters kafkaParameters = null;
  48
  49     /**
  50      * Set up testing.
  51      *
  52      * @throws ApexEventException on test set up errors.
  53      */
  54     @Before
  55     public void setUp() throws ApexEventException {
  56         apexKafkaConsumer = new ApexKafkaConsumer();
  57         consumerParameters = new EventHandlerParameters();
  58         apexKafkaProducer = new ApexKafkaProducer();
  59         consumerParameters
  60                 .setCarrierTechnologyParameters(new KafkaCarrierTechnologyParameters() {});
  61         apexKafkaConsumer.init("TestApexKafkaConsumer", consumerParameters, incomingEventReceiver);
  62
  63         apexKafkaConsumer2 = new ApexKafkaConsumer();
  64         consumerParameters2 = new EventHandlerParameters();
  65         kafkaParameters = new KafkaCarrierTechnologyParameters();
  66         String[][] kafkaProperties = {
  67             {"value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"},
  68             {"schema.registry.url", "[http://test-registory:8080]"}
  69         };
  70         kafkaParameters.setKafkaProperties(kafkaProperties);
  71
  72         consumerParameters2
  73                 .setCarrierTechnologyParameters(kafkaParameters);
  74         apexKafkaConsumer2.init("TestApexKafkaConsumer2", consumerParameters2, incomingEventReceiver);
  75     }
  76
  77     @Test
  78     public void testStart() {
  79         assertThatCode(apexKafkaConsumer::start).doesNotThrowAnyException();
  80         assertThatCode(apexKafkaConsumer2::start).doesNotThrowAnyException();
  81     }
  82
  83     @Test
  84     public void testGetName() {
  85         assertEquals("TestApexKafkaConsumer", apexKafkaConsumer.getName());
  86         assertEquals("TestApexKafkaConsumer2", apexKafkaConsumer2.getName());
  87     }
  88
  89     @Test
  90     public void testGetPeeredReference() {
  91         assertNull(apexKafkaConsumer.getPeeredReference(EventHandlerPeeredMode.REQUESTOR));
  92         assertNull(apexKafkaConsumer2.getPeeredReference(EventHandlerPeeredMode.REQUESTOR));
  93     }
  94
  95     @Test
  96     public void testSetPeeredReference() {
  97         PeeredReference peeredReference = new PeeredReference(EventHandlerPeeredMode.REQUESTOR,
  98                 apexKafkaConsumer, apexKafkaProducer);
  99         apexKafkaConsumer.setPeeredReference(EventHandlerPeeredMode.REQUESTOR, peeredReference);
 100         assertNotNull(apexKafkaConsumer.getPeeredReference(EventHandlerPeeredMode.REQUESTOR));
 101
 102         PeeredReference peeredReference2 = new PeeredReference(EventHandlerPeeredMode.REQUESTOR,
 103                         apexKafkaConsumer2, apexKafkaProducer);
 104         apexKafkaConsumer2.setPeeredReference(EventHandlerPeeredMode.REQUESTOR, peeredReference2);
 105         assertNotNull(apexKafkaConsumer2.getPeeredReference(EventHandlerPeeredMode.REQUESTOR));
 106     }
 107
 108     @Test(expected = java.lang.NullPointerException.class)
 109     public void testRun() {
 110         apexKafkaConsumer.run();
 111         apexKafkaConsumer2.run();
 112     }
 113
 114     @Test(expected = java.lang.NullPointerException.class)
 115     public void testStop() {
 116         apexKafkaConsumer.stop();
 117         apexKafkaConsumer2.stop();
 118     }
 119
 120     @Test(expected = ApexEventException.class)
 121     public void testInitWithNonKafkaCarrierTechnologyParameters() throws ApexEventException {
 122         consumerParameters.setCarrierTechnologyParameters(new CarrierTechnologyParameters() {});
 123         apexKafkaConsumer.init("TestApexKafkaConsumer", consumerParameters, incomingEventReceiver);
 124         apexKafkaConsumer2.init("TestApexKafkaConsumer2", consumerParameters, incomingEventReceiver);
 125     }
 126
 127 }