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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP Policy Engine - Common Modules
   4  * ================================================================================
   5  * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
   6  * Modifications Copyright (C) 2023 Nordix Foundation.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.common.endpoints.event.comm.bus;
  23
  24 import static org.junit.Assert.assertEquals;
  25 import static org.junit.Assert.assertTrue;
  26 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_KAFKA_SOURCE_TOPICS;
  27
  28 import java.util.Arrays;
  29 import java.util.Deque;
  30 import java.util.LinkedList;
  31 import java.util.List;
  32 import java.util.Properties;
  33 import org.junit.After;
  34 import org.junit.Before;
  35 import org.junit.Test;
  36 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
  37
  38 public class KafkaTopicSourceFactoryTest extends KafkaTopicFactoryTestBase<KafkaTopicSource> {
  39
  40     private SourceFactory factory;
  41
  42     public static final String KAFKA_SERVER = "localhost:9092";
  43
  44     /**
  45      * Creates the object to be tested.
  46      */
  47     @Before
  48     @Override
  49     public void setUp() {
  50         super.setUp();
  51
  52         factory = new SourceFactory();
  53     }
  54
  55     @After
  56     public void tearDown() {
  57         factory.destroy();
  58     }
  59
  60     @Test
  61     @Override
  62     public void testBuildProperties() {
  63
  64         initFactory();
  65
  66         List<KafkaTopicSource> topics = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build());
  67         assertEquals(1, topics.size());
  68         assertEquals(MY_TOPIC, topics.get(0).getTopic());
  69         assertEquals(MY_EFFECTIVE_TOPIC, topics.get(0).getEffectiveTopic());
  70
  71         BusTopicParams params = getLastParams();
  72         assertEquals(true, params.isManaged());
  73         assertEquals(false, params.isUseHttps());
  74         assertEquals(Arrays.asList(KAFKA_SERVER), params.getServers());
  75         assertEquals(MY_TOPIC, params.getTopic());
  76         assertEquals(MY_EFFECTIVE_TOPIC, params.getEffectiveTopic());
  77     }
  78
  79     @Test
  80     @Override
  81     public void testDestroyString_testGet_testInventory() {
  82         super.testDestroyString_testGet_testInventory();
  83         super.testDestroyString_Ex();
  84     }
  85
  86     @Test
  87     @Override
  88     public void testDestroy() {
  89         super.testDestroy();
  90     }
  91
  92     @Test
  93     public void testGet() {
  94         super.testGet_Ex();
  95     }
  96
  97     @Test
  98     public void testToString() {
  99         assertTrue(factory.toString().startsWith("IndexedKafkaTopicSourceFactory ["));
 100     }
 101
 102     @Override
 103     protected void initFactory() {
 104         if (factory != null) {
 105             factory.destroy();
 106         }
 107
 108         factory = new SourceFactory();
 109     }
 110
 111     @Override
 112     protected List<KafkaTopicSource> buildTopics(Properties properties) {
 113         return factory.build(properties);
 114     }
 115
 116     @Override
 117     protected KafkaTopicSource buildTopic(BusTopicParams params) {
 118         return factory.build(params);
 119     }
 120
 121     @Override
 122     protected KafkaTopicSource buildTopic(List<String> servers, String topic) {
 123         return factory.build(servers, topic);
 124     }
 125
 126     @Override
 127     protected void destroyFactory() {
 128         factory.destroy();
 129     }
 130
 131     @Override
 132     protected void destroyTopic(String topic) {
 133         factory.destroy(topic);
 134     }
 135
 136     @Override
 137     protected List<KafkaTopicSource> getInventory() {
 138         return factory.inventory();
 139     }
 140
 141     @Override
 142     protected KafkaTopicSource getTopic(String topic) {
 143         return factory.get(topic);
 144     }
 145
 146     @Override
 147     protected BusTopicParams getLastParams() {
 148         return factory.params.getLast();
 149     }
 150
 151     @Override
 152     protected TopicPropertyBuilder makePropBuilder() {
 153         return new KafkaTopicPropertyBuilder(PROPERTY_KAFKA_SOURCE_TOPICS);
 154     }
 155
 156     /**
 157      * Factory that records the parameters of all of the sources it creates.
 158      */
 159     private static class SourceFactory extends IndexedKafkaTopicSourceFactory {
 160         private Deque<BusTopicParams> params = new LinkedList<>();
 161
 162         @Override
 163         protected KafkaTopicSource makeSource(BusTopicParams busTopicParams) {
 164             params.add(busTopicParams);
 165             return super.makeSource(busTopicParams);
 166         }
 167     }
 168 }