gerrit.onap Code Review - dcaegen2/services.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP : DATALAKE
   4  * ================================================================================
   5  * Copyright 2019 China Mobile
   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.datalake.feeder.service;
  22
  23 import static org.junit.Assert.assertTrue;
  24 import static org.mockito.Mockito.when;
  25
  26 import java.lang.reflect.InvocationTargetException;
  27 import java.lang.reflect.Method;
  28 import java.util.ArrayList;
  29 import java.util.List;
  30
  31 import org.apache.commons.lang3.tuple.Pair;
  32 import org.junit.Test;
  33 import org.junit.runner.RunWith;
  34 import org.mockito.InjectMocks;
  35 import org.mockito.Mock;
  36 import org.mockito.junit.MockitoJUnitRunner;
  37 import org.onap.datalake.feeder.config.ApplicationConfiguration;
  38 import org.onap.datalake.feeder.domain.Kafka;
  39 import org.onap.datalake.feeder.dto.TopicConfig;
  40 import org.springframework.context.ApplicationContext;
  41
  42 /**
  43  * Test StoreService
  44  *
  45  * @author Guobiao Mo
  46  *
  47  */
  48 @RunWith(MockitoJUnitRunner.class)
  49 public class StoreServiceTest {
  50
  51         @InjectMocks
  52         private StoreService storeService = new StoreService();
  53
  54         @Mock
  55         private ApplicationContext context;
  56
  57         @Mock
  58         private ApplicationConfiguration config;
  59
  60         @Mock
  61         private TopicConfigPollingService configPollingService;
  62
  63         @Mock
  64         private MongodbService mongodbService;
  65
  66         @Mock
  67         private CouchbaseService couchbaseService;
  68
  69         @Mock
  70         private ElasticsearchService elasticsearchService;
  71
  72         @Mock
  73         private HdfsService hdfsService;
  74
  75         @Mock
  76         private Kafka kafka;
  77
  78         public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
  79                 Method init = storeService.getClass().getDeclaredMethod("init");
  80                 init.setAccessible(true);
  81                 init.invoke(storeService);
  82         }
  83
  84         private TopicConfig createTopicConfig(String topicStr, String type) {
  85
  86                 TopicConfig topicConfig = new TopicConfig();
  87                 topicConfig.setName(topicStr);
  88                 topicConfig.setDataFormat(type);
  89                 topicConfig.setSaveRaw(true);
  90
  91                 when(configPollingService.getEffectiveTopicConfig(topicStr)).thenReturn(topicConfig);
  92
  93                 return topicConfig;
  94         }
  95
  96         @Test
  97         public void saveMessages() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
  98                 testInit();
  99
 100                 TopicConfig topicConfig = createTopicConfig("test1", "JSON");
 101                 topicConfig.setAggregateArrayPath("/test");
 102                 topicConfig.setFlattenArrayPath("/test");
 103
 104                 topicConfig = createTopicConfig("test2", "XML");
 105                 topicConfig.setSaveRaw(false);
 106
 107                 topicConfig = createTopicConfig("test3", "YAML");
 108
 109                 topicConfig.setSinkdbs(new ArrayList<>());
 110                 topicConfig.getSinkdbs().add("Elasticsearch");
 111                 topicConfig.getSinkdbs().add("Couchbase");
 112                 topicConfig.getSinkdbs().add("Druid");
 113                 topicConfig.getSinkdbs().add("MongoDB");
 114                 topicConfig.getSinkdbs().add("HDFS");
 115
 116
 117                 topicConfig.setEnabledSinkdbs(new ArrayList<>());
 118                 topicConfig.getEnabledSinkdbs().add("Elasticsearch");
 119                 assertTrue(topicConfig.supportElasticsearch());
 120
 121
 122                 createTopicConfig("test4", "TEXT");
 123
 124                 when(config.getTimestampLabel()).thenReturn("ts");
 125                 when(config.getRawDataLabel()).thenReturn("raw");
 126
 127                 //JSON
 128                 List<Pair<Long, String>> messages = new ArrayList<>();
 129                 messages.add(Pair.of(100L, "{test: 1}"));
 130
 131                 storeService.saveMessages(kafka, "test1", messages);
 132
 133                 //XML
 134                 List<Pair<Long, String>> messagesXml = new ArrayList<>();
 135                 messagesXml.add(Pair.of(100L, "<test></test>"));
 136                 messagesXml.add(Pair.of(100L, "<test></test"));//bad xml to trigger exception
 137
 138                 storeService.saveMessages(kafka, "test2", messagesXml);
 139
 140                 //YAML
 141                 List<Pair<Long, String>> messagesYaml = new ArrayList<>();
 142                 messagesYaml.add(Pair.of(100L, "test: yes"));
 143
 144                 storeService.saveMessages(kafka, "test3", messagesYaml);
 145
 146                 //TEXT
 147                 List<Pair<Long, String>> messagesText = new ArrayList<>();
 148                 messagesText.add(Pair.of(100L, "test message"));
 149
 150                 storeService.saveMessages(kafka, "test4", messagesText);
 151
 152                 //Null mesg
 153                 storeService.saveMessages(kafka, "test", null);
 154         }
 155
 156         @Test
 157         public void testFlush() {
 158                 storeService.flush();
 159                 storeService.flushStall();
 160         }
 161 }