components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java

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