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