gerrit.onap Code Review - policy/common.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP Policy Engine - Common Modules
   4  * ================================================================================
   5  * Copyright (C) 2022-2024 Nordix Foundation.
   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
  21 package org.onap.policy.common.endpoints.event.comm.bus;
  22
  23 import static org.junit.jupiter.api.Assertions.assertEquals;
  24 import static org.junit.jupiter.api.Assertions.assertFalse;
  25 import static org.junit.jupiter.api.Assertions.assertTrue;
  26 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_KAFKA_SOURCE_TOPICS;
  27
  28 import java.util.Deque;
  29 import java.util.LinkedList;
  30 import java.util.List;
  31 import java.util.Properties;
  32 import org.junit.jupiter.api.AfterEach;
  33 import org.junit.jupiter.api.BeforeEach;
  34 import org.junit.jupiter.api.Test;
  35 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
  36
  37 class KafkaTopicSourceFactoryTest extends KafkaTopicFactoryTestBase<KafkaTopicSource> {
  38
  39     private SourceFactory factory;
  40
  41     public static final String KAFKA_SERVER = "localhost:9092";
  42
  43     /**
  44      * Creates the object to be tested.
  45      */
  46     @BeforeEach
  47     @Override
  48     public void setUp() {
  49         super.setUp();
  50
  51         factory = new SourceFactory();
  52     }
  53
  54     @AfterEach
  55     public void tearDown() {
  56         factory.destroy();
  57     }
  58
  59     @Test
  60     @Override
  61     void testBuildProperties() {
  62
  63         initFactory();
  64
  65         List<KafkaTopicSource> topics = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build());
  66         assertEquals(1, topics.size());
  67         assertEquals(MY_TOPIC, topics.get(0).getTopic());
  68         assertEquals(MY_EFFECTIVE_TOPIC, topics.get(0).getEffectiveTopic());
  69
  70         BusTopicParams params = getLastParams();
  71         assertTrue(params.isManaged());
  72         assertFalse(params.isUseHttps());
  73         assertEquals(List.of(KAFKA_SERVER), params.getServers());
  74         assertEquals(MY_TOPIC, params.getTopic());
  75         assertEquals(MY_EFFECTIVE_TOPIC, params.getEffectiveTopic());
  76     }
  77
  78     @Test
  79     @Override
  80     void testDestroyString_testGet_testInventory() {
  81         super.testDestroyString_testGet_testInventory();
  82         super.testDestroyString_Ex();
  83     }
  84
  85     @Test
  86     @Override
  87     void testDestroy() {
  88         super.testDestroy();
  89     }
  90
  91     @Test
  92     void testGet() {
  93         super.testGet_Ex();
  94     }
  95
  96     @Test
  97     void testToString() {
  98         assertTrue(factory.toString().startsWith("IndexedKafkaTopicSourceFactory ["));
  99     }
 100
 101     @Override
 102     protected void initFactory() {
 103         if (factory != null) {
 104             factory.destroy();
 105         }
 106
 107         factory = new SourceFactory();
 108     }
 109
 110     @Override
 111     protected List<KafkaTopicSource> buildTopics(Properties properties) {
 112         return factory.build(properties);
 113     }
 114
 115     @Override
 116     protected KafkaTopicSource buildTopic(BusTopicParams params) {
 117         return factory.build(params);
 118     }
 119
 120     @Override
 121     protected KafkaTopicSource buildTopic(List<String> servers, String topic) {
 122         return factory.build(servers, topic);
 123     }
 124
 125     @Override
 126     protected void destroyFactory() {
 127         factory.destroy();
 128     }
 129
 130     @Override
 131     protected void destroyTopic(String topic) {
 132         factory.destroy(topic);
 133     }
 134
 135     @Override
 136     protected List<KafkaTopicSource> getInventory() {
 137         return factory.inventory();
 138     }
 139
 140     @Override
 141     protected KafkaTopicSource getTopic(String topic) {
 142         return factory.get(topic);
 143     }
 144
 145     @Override
 146     protected BusTopicParams getLastParams() {
 147         return factory.params.getLast();
 148     }
 149
 150     @Override
 151     protected TopicPropertyBuilder makePropBuilder() {
 152         return new KafkaTopicPropertyBuilder(PROPERTY_KAFKA_SOURCE_TOPICS);
 153     }
 154
 155     /**
 156      * Factory that records the parameters of all the sources it creates.
 157      */
 158     private static class SourceFactory extends IndexedKafkaTopicSourceFactory {
 159         private Deque<BusTopicParams> params = new LinkedList<>();
 160
 161         @Override
 162         protected KafkaTopicSource makeSource(BusTopicParams busTopicParams) {
 163             params.add(busTopicParams);
 164             return super.makeSource(busTopicParams);
 165         }
 166     }
 167 }