if (this.getClass() != obj.getClass())
return false;
- return name.equals(((Db) obj).getName());
+ return id==((Db) obj).getId();
}
@Override
public int hashCode() {
- return name.hashCode();
+ return id;
}
}
import org.json.JSONObject;
import org.onap.datalake.feeder.dto.TopicConfig;
import org.onap.datalake.feeder.enumeration.DataFormat;
-import org.onap.datalake.feeder.enumeration.DbTypeEnum;
-import org.onap.datalake.feeder.service.db.CouchbaseService;
-import org.onap.datalake.feeder.service.db.DbStoreService;
-import org.onap.datalake.feeder.service.db.ElasticsearchService;
-import org.onap.datalake.feeder.service.db.HdfsService;
-import org.onap.datalake.feeder.service.db.MongodbService;
import com.fasterxml.jackson.annotation.JsonBackReference;
@Column(name = "`flatten_array_path`")
protected String flattenArrayPath;
- public Topic() {
- }
-/*
- public Topic(String name) {//TODO
- //this.name = name;
- }
-*/
public String getName() {
return topicName.getId();
}
return 3650;//default to 10 years for safe
}
}
-/*
- public boolean supportHdfs() {
- return supportDb(DbTypeEnum.HDFS);
- }
-
- public boolean supportElasticsearch() {
- return supportDb(DbTypeEnum.ES);
- }
-
- public boolean supportCouchbase() {
- return supportDb(DbTypeEnum.CB);
- }
- public boolean supportDruid() {
- return supportDb(DbTypeEnum.DRUID);
- }
-
- public boolean supportMongoDB() {
- return supportDb(DbTypeEnum.MONGO);
- }
-
- private boolean supportDb(DbTypeEnum dbTypeEnum) {
- for(Db db : dbs) {
-
- }
- }
-*/
public DataFormat getDataFormat2() {
if (dataFormat != null) {
return DataFormat.fromString(dataFormat);
//extract DB id from JSON attributes, support multiple attributes
public String getMessageId(JSONObject json) {
- String id = null;
+ String ret = null;
if (StringUtils.isNotBlank(messageIdPath)) {
String[] paths = messageIdPath.split(",");
}
sb.append(json.query(paths[i]).toString());
}
- id = sb.toString();
+ ret = sb.toString();
}
- return id;
+ return ret;
}
public TopicConfig getTopicConfig() {
if (this.getClass() != obj.getClass())
return false;
- return name.equals(((TopicConfig) obj).getName());
+ return id==((TopicConfig) obj).getId();
}
@Override
public int hashCode() {
- return name.hashCode();
+ return id;
}
}
}
@PostConstruct
- private void init() {
+ @Override
+ public void init() {
// Initialize Couchbase Connection
try {
//this tunes the SDK (to customize connection timeout)
public interface DbStoreService {
void saveJsons(EffectiveTopic topic, List<JSONObject> jsons);
+
+ void init();
}
//ES Encrypted communication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_encrypted_communication.html#_encrypted_communication
//Basic authentication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_basic_authentication.html
@PostConstruct
- private void init() {
+ @Override
+ public void init() {
String elasticsearchHost = elasticsearch.getHost();
// Initialize the Connection
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.onap.datalake.feeder.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;
import lombok.Getter;
-import lombok.Setter;
/**
* Service to write data to HDFS
public class HdfsService implements DbStoreService {
private final Logger log = LoggerFactory.getLogger(this.getClass());
-
+
private Db hdfs;
@Autowired
ApplicationConfiguration config;
FileSystem fileSystem;
- private boolean isReady = false;
private ThreadLocal<Map<String, Buffer>> bufferLocal = ThreadLocal.withInitial(HashMap::new);
private ThreadLocal<SimpleDateFormat> dayFormat = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd"));
private ThreadLocal<SimpleDateFormat> timeFormat = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-SSS"));
- @Setter
@Getter
private class Buffer {
long lastFlush;
}
}
- public void addData(List<Pair<Long, String>> messages) {
- if (data.isEmpty()) { //reset the last flush time stamp to current if no existing data in buffer
- lastFlush = System.currentTimeMillis();
- }
-
- messages.stream().forEach(message -> data.add(message.getRight()));//note that message left is not used
- }
-
+ /*
+ public void addData(List<Pair<Long, String>> messages) {
+ if (data.isEmpty()) { //reset the last flush time stamp to current if no existing data in buffer
+ lastFlush = System.currentTimeMillis();
+ }
+
+ messages.stream().forEach(message -> data.add(message.getRight()));//note that message left is not used
+ }
+ */
public void addData2(List<JSONObject> messages) {
if (data.isEmpty()) { //reset the last flush time stamp to current if no existing data in buffer
lastFlush = System.currentTimeMillis();
}
- messages.stream().forEach(message -> data.add(message.toString()));
+ messages.stream().forEach(message -> data.add(message.toString()));
}
private void saveMessages(String topic, List<String> bufferList) throws IOException {
public HdfsService(Db db) {
hdfs = db;
}
-
+
@PostConstruct
- private void init() {
+ @Override
+ public void init() {
// Initialize HDFS Connection
try {
//Get configuration of Hadoop system
ShutdownHookManager hadoopShutdownHookManager = ShutdownHookManager.get();
hadoopShutdownHookManager.clearShutdownHooks();
- isReady = true;
} catch (Exception ex) {
log.error("error connection to HDFS.", ex);
- isReady = false;
}
}
bufferLocal.get().forEach((topic, buffer) -> buffer.flushStall(topic));
}
- //used if raw data should be saved
- public void saveMessages(EffectiveTopic topic, List<Pair<Long, String>> messages) {
- String topicStr = topic.getName();
-
- Map<String, Buffer> bufferMap = bufferLocal.get();
- final Buffer buffer = bufferMap.computeIfAbsent(topicStr, k -> new Buffer());
-
- buffer.addData(messages);
-
- if (!config.isAsync() || buffer.getData().size() >= config.getHdfsBatchSize()) {
- buffer.flush(topicStr);
- } else {
- log.debug("buffer size too small to flush {}: bufferData.size() {} < config.getHdfsBatchSize() {}", topicStr, buffer.getData().size(), config.getHdfsBatchSize());
+ /*
+ //used if raw data should be saved
+ public void saveMessages(EffectiveTopic topic, List<Pair<Long, String>> messages) {
+ String topicStr = topic.getName();
+
+ Map<String, Buffer> bufferMap = bufferLocal.get();
+ final Buffer buffer = bufferMap.computeIfAbsent(topicStr, k -> new Buffer());
+
+ buffer.addData(messages);
+
+ if (!config.isAsync() || buffer.getData().size() >= config.getHdfsBatchSize()) {
+ buffer.flush(topicStr);
+ } else {
+ log.debug("buffer size too small to flush {}: bufferData.size() {} < config.getHdfsBatchSize() {}", topicStr, buffer.getData().size(), config.getHdfsBatchSize());
+ }
}
- }
-
+ */
@Override
public void saveJsons(EffectiveTopic topic, List<JSONObject> jsons) {
String topicStr = topic.getName();
} else {
log.debug("buffer size too small to flush {}: bufferData.size() {} < config.getHdfsBatchSize() {}", topicStr, buffer.getData().size(), config.getHdfsBatchSize());
}
-
+
}
-
+
}
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoClientOptions.Builder;
import com.mongodb.MongoCredential;
+import com.mongodb.MongoTimeoutException;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
}
@PostConstruct
- private void init() {
+ @Override
+ public void init() {
String host = mongodb.getHost();
Integer port = mongodb.getPort();
for (BulkWriteError bulkWriteError : bulkWriteErrors) {
log.error("Failed record: {}", bulkWriteError);
}
+ } catch (MongoTimeoutException e) {
+ log.error("saveJsons()", e);
}
log.debug("saved text to effectiveTopic = {}, batch count = {} ", effectiveTopic, jsons.size());
import org.onap.datalake.feeder.dto.TopicConfig;
import org.onap.datalake.feeder.domain.Db;
import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.repository.TopicNameRepository;
import org.onap.datalake.feeder.repository.TopicRepository;
import org.onap.datalake.feeder.service.DbService;
import org.onap.datalake.feeder.service.DmaapService;
import org.onap.datalake.feeder.service.TopicService;
+import org.onap.datalake.feeder.util.TestUtil;
import org.springframework.validation.BindingResult;
import javax.servlet.http.HttpServletResponse;
@RunWith(MockitoJUnitRunner.class)
public class TopicControllerTest {
- static String DEFAULT_TOPIC_NAME = "_DL_DEFAULT_";
-
- @Mock
- private HttpServletResponse httpServletResponse;
-
- @Mock
- private BindingResult mockBindingResult;
-
- @Mock
- private TopicRepository topicRepository;
-
- @Mock
-
- private TopicService topicServiceMock;
-
- @InjectMocks
- private TopicService topicService1;
-
- @Mock
- private ApplicationConfiguration config;
-
- @Mock
- private DbService dbService1;
-
- @Mock
- private DmaapService dmaapService1;
-
- @Before
- public void setupTest() {
- MockitoAnnotations.initMocks(this);
- // While the default boolean return value for a mock is 'false',
- // it's good to be explicit anyway:
- when(mockBindingResult.hasErrors()).thenReturn(false);
- }
-
- public void setAccessPrivateFields(TopicController topicController) throws NoSuchFieldException,
- IllegalAccessException {
- Field topicService = topicController.getClass().getDeclaredField("topicService");
- topicService.setAccessible(true);
- topicService.set(topicController, topicService1);
- Field topicRepository1 = topicController.getClass().getDeclaredField("topicRepository");
- topicRepository1.setAccessible(true);
- topicRepository1.set(topicController, topicRepository);
-// Field dbService = topicController.getClass().getDeclaredField("dbService");
- // dbService.setAccessible(true);
- // dbService.set(topicController, dbService1);
- }
-
- @Test
- public void testListTopic() throws IOException, NoSuchFieldException, IllegalAccessException{
- TopicController topicController = new TopicController();
- setAccessPrivateFields(topicController);
- }
-
- //@Test
- public void testCreateTopic() throws IOException, NoSuchFieldException, IllegalAccessException {
- TopicController topicController = new TopicController();
- setAccessPrivateFields(topicController);
- //when(topicRepository.findById("ab")).thenReturn(Optional.of(new Topic("ab")));
- // when(config.getDefaultTopicName()).thenReturn(DEFAULT_TOPIC_NAME);
- PostReturnBody<TopicConfig> postTopic = topicController.createTopic(new TopicConfig(), mockBindingResult, httpServletResponse);
- assertEquals(postTopic.getStatusCode(), 200);
- when(mockBindingResult.hasErrors()).thenReturn(true);
- PostReturnBody<TopicConfig> topicConfig= topicController.createTopic(new TopicConfig(), mockBindingResult, httpServletResponse);
- assertEquals(null, topicConfig);
- when(mockBindingResult.hasErrors()).thenReturn(false);
- TopicConfig a = new TopicConfig();
- a.setName(DEFAULT_TOPIC_NAME);
- //when(topicRepository.findById(DEFAULT_TOPIC_NAME)).thenReturn(Optional.of(new Topic(DEFAULT_TOPIC_NAME)));
- PostReturnBody<TopicConfig> postTopic2= topicController.createTopic(a, mockBindingResult, httpServletResponse);
- //assertEquals(null, postTopic2);
- }
-
- @Test
- public void testUpdateTopic() throws IOException, NoSuchFieldException, IllegalAccessException {
- TopicController topicController = new TopicController();
- setAccessPrivateFields(topicController);
- PostReturnBody<TopicConfig> postTopic = topicController.updateTopic(1, new TopicConfig(), mockBindingResult, httpServletResponse);
- assertEquals(null, postTopic);
- Topic a = new Topic();
- a.setId(1);
- //when(topicRepository.findById(1)).thenReturn(Optional.of(a));
- TopicConfig ac = new TopicConfig();
- ac.setName("a");
- ac.setEnabled(true);
- PostReturnBody<TopicConfig> postConfig1 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
- //assertEquals(200, postConfig1.getStatusCode());
- assertNull(postConfig1);
- //TopicConfig ret = postConfig1.getReturnBody();
- //assertEquals("a", ret.getName());
- //assertEquals(true, ret.isEnabled());
- when(mockBindingResult.hasErrors()).thenReturn(true);
- PostReturnBody<TopicConfig> postConfig2 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
- assertEquals(null, postConfig2);
-
- }
-
- //@Test
- public void testListDmaapTopics() throws NoSuchFieldException, IllegalAccessException, IOException {
- TopicController topicController = new TopicController();
- Field dmaapService = topicController.getClass().getDeclaredField("dmaapService");
- dmaapService.setAccessible(true);
- dmaapService.set(topicController, dmaapService1);
- ArrayList<String> topics = new ArrayList<>();
- topics.add("a");
- when(dmaapService1.getTopics()).thenReturn(topics);
- List<String> strings = topicController.listDmaapTopics("KAFKA");
- for (String topic : strings) {
- assertEquals("a", topic);
- }
- }
+ static String DEFAULT_TOPIC_NAME = "_DL_DEFAULT_";
+
+ @Mock
+ private HttpServletResponse httpServletResponse;
+
+ @Mock
+ private BindingResult mockBindingResult;
+
+ @Mock
+ private TopicRepository topicRepository;
+
+ @Mock
+ private TopicService topicService;
+
+ @Mock
+ private TopicNameRepository topicNameRepository;
+
+ @InjectMocks
+ TopicController topicController;
+
+ @Mock
+ private ApplicationConfiguration config;
+
+ @Mock
+ private DbService dbService;
+
+ @Mock
+ private DmaapService dmaapService;
+
+ @Before
+ public void setupTest() throws NoSuchFieldException, IllegalAccessException {
+ // While the default boolean return value for a mock is 'false',
+ // it's good to be explicit anyway:
+ when(mockBindingResult.hasErrors()).thenReturn(false);
+ }
+
+ @Test
+ public void testListTopic() throws IOException, NoSuchFieldException, IllegalAccessException {
+ }
+
+ @Test
+ public void testCreateTopic() throws IOException {
+ Topic a = TestUtil.newTopic("a");
+ a.setId(1);
+ a.setEnabled(true);
+
+ TopicConfig ac = a.getTopicConfig();
+
+ when(topicService.fillTopicConfiguration(ac)).thenReturn(a);
+ PostReturnBody<TopicConfig> postTopic = topicController.createTopic(ac, mockBindingResult, httpServletResponse);
+ assertEquals(postTopic.getStatusCode(), 200);
+
+ when(mockBindingResult.hasErrors()).thenReturn(true);
+ PostReturnBody<TopicConfig> topicConfig = topicController.createTopic(ac, mockBindingResult, httpServletResponse);
+ assertEquals(null, topicConfig);
+ }
+
+ @Test
+ public void testUpdateTopic() throws IOException {
+ Topic a = TestUtil.newTopic("a");
+ a.setId(1);
+ a.setEnabled(true);
+
+ TopicConfig ac = a.getTopicConfig();
+
+ when(topicService.getTopic(1)).thenReturn(a);
+ PostReturnBody<TopicConfig> postConfig1 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
+ assertEquals(200, postConfig1.getStatusCode());
+ TopicConfig ret = postConfig1.getReturnBody();
+ assertEquals("a", ret.getName());
+ assertEquals(true, ret.isEnabled());
+
+ topicController.updateTopic(0, ac, mockBindingResult, httpServletResponse);
+
+ when(topicService.getTopic(1)).thenReturn(null);
+ topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
+
+ when(mockBindingResult.hasErrors()).thenReturn(true);
+ PostReturnBody<TopicConfig> postConfig2 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
+ assertNull(postConfig2);
+
+ }
+
+ @Test
+ public void testGetTopic() throws IOException {
+ Topic a = TestUtil.newTopic("a");
+ a.setId(1);
+ a.setEnabled(true);
+
+ when(topicService.getTopic(1)).thenReturn(a);
+ TopicConfig ac = topicController.getTopic(1, httpServletResponse);
+ when(topicService.getTopic(1)).thenReturn(null);
+ ac = topicController.getTopic(1, httpServletResponse);
+ }
+
+ @Test
+ public void testDeleteTopic() throws IOException {
+ Topic a = TestUtil.newTopic("a");
+ a.setId(1);
+ a.setEnabled(true);
+
+ when(topicService.getTopic(1)).thenReturn(a);
+ topicController.deleteTopic(1, httpServletResponse);
+ when(topicService.getTopic(1)).thenReturn(null);
+ topicController.deleteTopic(1, httpServletResponse);
+ }
+
+ @Test
+ public void testList() {
+ ArrayList<Topic> topics = new ArrayList<>();
+ topics.add(TestUtil.newTopic("a"));
+ topics.add(TestUtil.newTopic(DEFAULT_TOPIC_NAME));
+ when(topicRepository.findAll()).thenReturn(topics);
+
+ List<String> strings = topicController.list();
+ for (String topic : strings) {
+ System.out.println(topic);
+ }
+ }
}
Db mongoDB2 = TestUtil.newDb("MongoDB");
assertNotEquals(couchbase.hashCode(), mongoDB.hashCode());
assertNotEquals(couchbase, mongoDB);
- assertEquals(mongoDB, mongoDB2);
- assertFalse(mongoDB2.equals(null));
- assertFalse(mongoDB2.equals(new Topic()));
+ assertNotEquals(mongoDB, mongoDB2);
+ assertEquals(mongoDB, mongoDB);
+ assertFalse(mongoDB2.equals(null));
+
+ DbType dbType = new DbType("MONGO", "MongoDB");
+ dbType.setTool(false);
+ mongoDB.setDbType(dbType);
+ assertNotEquals(mongoDB2, dbType);
+ assertFalse(mongoDB.isTool());
+ assertFalse(mongoDB.isHdfs());
+ assertFalse(mongoDB.isElasticsearch());
+ assertFalse(mongoDB.isCouchbase());
+ assertFalse(mongoDB.isDruid());
+ assertTrue(mongoDB.isMongoDB());
+ assertFalse(mongoDB.getDbType().isTool());
+ System.out.println(mongoDB);
+
new Db();
mongoDB2.setHost("localhost");
mongoDB2.setPort(1234);
@Test
public void test(){
- DbType dbType = new DbType("123","Elasticsearch");
+ DbType dbType = new DbType("ES","Elasticsearch");
+
+
+ DbType dbType2 = new DbType("MONGO", "MongoDB");
+ dbType.setTool(false);
+
+
assertNotNull(dbType.toString());
assertEquals(dbType, dbType);
+ assertNotEquals(dbType, null);
+ assertNotEquals(dbType, "ES");
+ assertNotEquals(dbType, dbType2);
assertNotNull(dbType.hashCode());
+
+ assertEquals("MongoDB", dbType2.getName());
+ dbType2.setName(null);
+ dbType2.setDefaultPort(1);
+ assertTrue(1==dbType2.getDefaultPort());
+
+ dbType2.setDbs(null);
+ assertNull(dbType2.getDbs());
}
}
\ No newline at end of file
public class DesignTypeTest {
@Test
- public void testIs(){
-
+ public void test(){
DesignType designType = new DesignType();
designType.setName("Kibana Dashboard");
designType.setNote("test");
Portal portal = new Portal();
portal.setName("Kibana");
designType.setPortal(portal);
- assertTrue("Kibana Dashboard".equals(designType.getName()));
- assertTrue("test".equals(designType.getNote()));
- assertFalse("Kibana".equals(designType.getPortal()));
- }
+ assertEquals("Kibana Dashboard", designType.getName());
+ assertEquals("test", designType.getNote());
+ assertNotEquals("Kibana", designType.getPortal());
+ designType.setDbType(null);
+ designType.getDbType();
+
+ designType.setDesigns(null);
+ designType.getDesigns();
+
+ designType.getDesignTypeConfig();
+ }
}
\ No newline at end of file
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP : DataLake
+ * ================================================================================
+ * Copyright 2019 China Mobile
+ *=================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.datalake.feeder.domain;
+
+import org.junit.Test;
+import org.onap.datalake.feeder.util.TestUtil;
+
+/**
+ * Test TopicName
+ *
+ */
+
+public class EffectiveTopicTest {
+
+
+ @Test
+ public void test() {
+
+ Topic topic = TestUtil.newTopic("test Topic");
+
+ EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+ effectiveTopic = new EffectiveTopic(topic);
+ effectiveTopic.getName();
+ effectiveTopic.setName("");
+ effectiveTopic.getName();
+ effectiveTopic.setTopic(topic);
+ effectiveTopic.getTopic();
+ System.out.println(effectiveTopic);
+ }
+
+}
designType.setName("Kibana");
portalDesign.setDesignType(designType);
portalDesign.setNote("test");
+ portalDesign.setDbs(null);
assertFalse("1".equals(portalDesign.getId()));
assertTrue("templateTest".equals(portalDesign.getName()));
assertTrue("jsonString".equals(portalDesign.getBody()));
assertTrue("test".equals(portalDesign.getNote()));
assertFalse("Kibana".equals(portalDesign.getDesignType()));
assertFalse("false".equals(portalDesign.getSubmitted()));
+ assertNull(portalDesign.getDbs());
}
}
\ No newline at end of file
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP : DataLake
+ * ================================================================================
+ * Copyright 2019 China Mobile
+ *=================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.datalake.feeder.domain;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import org.junit.Test;
+
+/**
+ * Test TopicName
+ *
+ * @author Guobiao Mo
+ */
+
+public class TopicNameTest {
+
+
+ @Test
+ public void test() {
+ TopicName topicName = new TopicName("tes");
+ topicName.setDesigns(null);
+ topicName.getDesigns();
+ topicName.setTopics(null);
+ topicName.getTopics();
+ topicName.hashCode();
+
+ assertEquals(topicName, topicName);
+ assertNotEquals(topicName, null);
+ assertNotEquals(topicName, "test");
+
+ }
+
+}
*/
package org.onap.datalake.feeder.domain;
+import org.json.JSONObject;
import org.junit.Test;
+import org.onap.datalake.feeder.dto.TopicConfig;
import org.onap.datalake.feeder.enumeration.DataFormat;
import org.onap.datalake.feeder.util.TestUtil;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
public class TopicTest {
+
+ @Test
+ public void getMessageId() {
+ String text = "{ data: { data2 : { value : 'hello'}}}";
+
+ JSONObject json = new JSONObject(text);
+
+ Topic topic = TestUtil.newTopic("test getMessageId");
+ topic.setMessageIdPath("/data/data2/value");
+ }
+
+ @Test
+ public void getMessageIdFromMultipleAttributes() {
+ String text = "{ data: { data2 : { value : 'hello'}, data3 : 'world'}}";
+
+ JSONObject json = new JSONObject(text);
+
+ Topic topic = TestUtil.newTopic("test getMessageId");
+ topic.setMessageIdPath("/data/data2/value,/data/data3");
+
+ assertEquals("hello^world", topic.getMessageId(json));
+
+ topic.setMessageIdPath("");
+ assertNull(topic.getMessageId(json));
+ }
+/*
+ @Test
+ public void testArrayPath() {
+ Topic topic = TestUtil.newTopic("testArrayPath");
+ topic.setAggregateArrayPath("/data/data2/value,/data/data3");
+ topic.setFlattenArrayPath("/data/data2/value,/data/data3");
+
+ TopicConfig topicConfig = topic.getTopicConfig();
+ }
+
@Test
public void getMessageIdFromMultipleAttributes() {
Topic topic = TestUtil.newTopic("test getMessageId");
assertFalse(topic.equals(null));
assertFalse(topic.equals(new Db()));
}
+*/
+ @Test
+ public void testAggregate() {
+ Topic defaultTopic = TestUtil.newTopic("_DL_DEFAULT_");
+ Topic testTopic = TestUtil.newTopic("test");
+ testTopic.setId(1);
+ Topic testTopic2 = TestUtil.newTopic("test2");
+ testTopic2.setId(2);
+
+ //test null cases
+ testTopic.getAggregateArrayPath2() ;
+ testTopic.getFlattenArrayPath2() ;
+ //test not null cases
+ testTopic.setAggregateArrayPath("/data/data2/value,/data/data3");
+ testTopic.setFlattenArrayPath("/data/data2/value,/data/data3");
+
+ testTopic.getAggregateArrayPath2() ;
+ testTopic.getFlattenArrayPath2() ;
+
+ }
+
+
@Test
public void testIs() {
Topic defaultTopic = TestUtil.newTopic("_DL_DEFAULT_");
Topic testTopic2 = TestUtil.newTopic("test2");
testTopic2.setId(1);
- assertTrue(testTopic.equals(testTopic2));
+ assertEquals(testTopic, testTopic2);
+ assertNotEquals(testTopic, null);
+ assertNotEquals(testTopic, "test");
assertEquals(testTopic.hashCode(), testTopic2.hashCode());
assertNotEquals(testTopic.toString(), "test");
assertTrue(defaultTopic.isEnabled());
assertTrue(defaultTopic.isSaveRaw());
- //assertEquals(defaultTopic.getTopicConfig().getDataFormat2(), DataFormat.XML);
+ assertEquals(defaultTopic.getDataFormat2(), DataFormat.XML);
+ defaultTopic.setDataFormat(null);
+ assertNull(defaultTopic.getDataFormat2());
defaultTopic.setDataFormat(null);
assertEquals(testTopic.getDataFormat(), null);
Topic testTopic1 = TestUtil.newTopic("test");
assertFalse(testTopic1.isCorrelateClearedMessage());
+
+
+ testTopic.setPass("root123");
+ assertTrue("root123".equals(testTopic.getPass()));
+
+ assertEquals(3650, testTopic.getTtl());
+ defaultTopic.setTtl(20);
+ assertEquals(20, defaultTopic.getTtl());
}
}
import org.onap.datalake.feeder.dto.DbConfig;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
public class DbConfigTest {
@Test
public void testDbConfig() {
DbConfig dbConfig = new DbConfig();
+ dbConfig.setId(1);
+ assertEquals(1, dbConfig.getId());
dbConfig.setName("elasticsearch");
assertTrue("elasticsearch".equals(dbConfig.getName()));
dbConfig.setHost("localhost");
assertEquals(testPortalConfig.getEnabled(), null);
assertEquals(testPortalConfig.getLogin(), null);
assertEquals(testPortalConfig.getPass(), null);
+ assertEquals(testPortalConfig.getDb(), "Elasticsearch");
}
}
\ No newline at end of file
assertEquals(testPortalDesignConfig.getNote(), null);
assertEquals(testPortalDesignConfig.getName(), null);
assertEquals(testPortalDesignConfig.getSubmitted(), null);
+ assertEquals(testPortalDesignConfig.getDesignType(), null);
+ assertEquals(testPortalDesignConfig.getDisplay(), "test");
}
}
\ No newline at end of file
*/
package org.onap.datalake.feeder.dto;
-import org.json.JSONObject;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import java.util.HashSet;
+
import org.junit.Test;
import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.domain.Kafka;
import org.onap.datalake.feeder.domain.Topic;
import org.onap.datalake.feeder.util.TestUtil;
-import java.util.HashSet;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
/**
* Test Topic
*
*/
public class TopicConfigTest {
-
- @Test
- public void getMessageId() {
- String text = "{ data: { data2 : { value : 'hello'}}}";
-
- JSONObject json = new JSONObject(text);
-
- Topic topic = TestUtil.newTopic("test getMessageId");
- topic.setMessageIdPath("/data/data2/value");
-
- TopicConfig topicConfig = topic.getTopicConfig();
-
-// String value = topicConfig.getMessageId(json);
-
- // assertEquals(value, "hello");
- }
-
- @Test
- public void getMessageIdFromMultipleAttributes() {
- String text = "{ data: { data2 : { value : 'hello'}, data3 : 'world'}}";
-
- JSONObject json = new JSONObject(text);
-
- Topic topic = TestUtil.newTopic("test getMessageId");
- topic.setMessageIdPath("/data/data2/value,/data/data3");
-
- TopicConfig topicConfig = topic.getTopicConfig();
-
-// String value = topicConfig.getMessageId(json);
- // assertEquals(value, "hello^world");
-
- topic.setMessageIdPath("");
- topicConfig = topic.getTopicConfig();
- // assertNull(topicConfig.getMessageId(json));
-
- }
-
- @Test
- public void testArrayPath() {
- Topic topic = TestUtil.newTopic("testArrayPath");
- topic.setAggregateArrayPath("/data/data2/value,/data/data3");
- topic.setFlattenArrayPath("/data/data2/value,/data/data3");
-
- TopicConfig topicConfig = topic.getTopicConfig();
-/*
- String[] value = topicConfig.getAggregateArrayPath2();
- assertEquals(value[0], "/data/data2/value");
- assertEquals(value[1], "/data/data3");
-
- value = topicConfig.getFlattenArrayPath2();
- assertEquals(value[0], "/data/data2/value");
- assertEquals(value[1], "/data/data3");*/
- }
-
- @Test
- public void testIs() {
- Topic testTopic = TestUtil.newTopic("test");
-
- TopicConfig testTopicConfig = testTopic.getTopicConfig();
- testTopicConfig.setSinkdbs(null);
- testTopicConfig.setEnabledSinkdbs(null);
- //assertFalse(testTopicConfig.supportElasticsearch());
- //assertNull(testTopicConfig.getDataFormat2());
-
- testTopic.setDbs(new HashSet<>());
- Db esDb = TestUtil.newDb("Elasticsearch");
- esDb.setEnabled(true);
- testTopic.getDbs().add(esDb);
-
- testTopicConfig = testTopic.getTopicConfig();
-
- //assertEquals(testTopicConfig, new Topic("test").getTopicConfig());
- assertNotEquals(testTopicConfig, testTopic);
- assertNotEquals(testTopicConfig, null);
- //assertEquals(testTopicConfig.hashCode(), (new Topic("test").getTopicConfig()).hashCode());
- /*
- assertTrue(testTopicConfig.supportElasticsearch());
- assertFalse(testTopicConfig.supportCouchbase());
- assertFalse(testTopicConfig.supportDruid());
- assertFalse(testTopicConfig.supportMongoDB());
- assertFalse(testTopicConfig.supportHdfs());
-
- testTopic.getDbs().remove(new Db("Elasticsearch"));
- testTopicConfig = testTopic.getTopicConfig();
- assertFalse(testTopicConfig.supportElasticsearch());
- */
- }
+ @Test
+ public void testIs() {
+ Topic testTopic = TestUtil.newTopic("test");
+
+ TopicConfig testTopicConfig = testTopic.getTopicConfig();
+ testTopicConfig.setSinkdbs(null);
+ testTopicConfig.setEnabledSinkdbs(null);
+
+ testTopic.setDbs(null);
+ testTopic.setKafkas(null);
+ testTopicConfig = testTopic.getTopicConfig();
+
+ testTopic.setDbs(new HashSet<>());
+ Db esDb = TestUtil.newDb("Elasticsearch");
+ esDb.setEnabled(true);
+ testTopic.getDbs().add(esDb);
+
+ esDb = TestUtil.newDb("MongoDB");
+ esDb.setEnabled(false);
+ testTopic.getDbs().add(esDb);
+
+
+ testTopic.setKafkas(new HashSet<>());
+ Kafka kafka = TestUtil.newKafka("k1");
+ kafka.setEnabled(true);
+ testTopic.getKafkas().add(kafka);
+ testTopicConfig = testTopic.getTopicConfig();
+
+
+
+ TopicConfig testTopicConfig2 = TestUtil.newTopic("test").getTopicConfig();
+ assertNotEquals(testTopicConfig, testTopicConfig2);
+ assertEquals(testTopicConfig, testTopicConfig);
+ assertNotEquals(testTopicConfig.hashCode(), testTopicConfig2.hashCode());
+ assertNotEquals(testTopicConfig, testTopic);
+ assertNotEquals(testTopicConfig, null);
+ }
}
import static org.mockito.Mockito.when;
import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.util.TestUtil;
@RunWith(MockitoJUnitRunner.class)
public class DmaapServiceTest {
static String DMAPP_ZOOKEEPER_HOST_PORT = "test:2181";
- @InjectMocks
private DmaapService dmaapService;
@Mock
private ApplicationConfiguration config;
@Mock
private TopicService topicService;
-
+
+ @Before
+ public void init() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+ Kafka kafka = TestUtil.newKafka("kafka");
+ dmaapService = new DmaapService(kafka);
+
+ Field configField = DmaapService.class.getDeclaredField("config");
+ configField.setAccessible(true);
+ configField.set(dmaapService, config);
+
+ /*
+ Method initMethod = DmaapService.class.getDeclaredMethod("init");
+ initMethod.setAccessible(true);
+ initMethod.invoke(dmaapService); */
+ }
+
@Test
public void testGetTopics() throws InterruptedException {
List<String> list = new ArrayList<>();
//when(config.getDmaapZookeeperHostPort()).thenReturn(DMAPP_ZOOKEEPER_HOST_PORT);
assertNotEquals(list, dmaapService.getTopics());
- //when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
- //dmaapService.cleanUp();
+ when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+ dmaapService.cleanUp();
}
@Test
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.springframework.context.ApplicationContext;
+import java.lang.reflect.Field;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import static org.junit.Assert.*;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class PullServiceTest {
- @InjectMocks
- private PullService pullService;
+ @InjectMocks
+ private PullService pullService;
- @Mock
- private ApplicationContext context;
+ @Mock
+ private ApplicationContext context;
- @Mock
- private ApplicationConfiguration config;
+ @Mock
+ private ApplicationConfiguration config;
- @Mock
- private ExecutorService executorService;
+ @Mock
+ private ExecutorService executorService;
- @Mock
- private List<Puller> consumers;
+ @Mock
+
private List<Puller> consumers;
- @Test
- public void isRunning() {
- assertEquals(pullService.isRunning(), false);
- }
+ @Test
+ public void isRunning() {
+ assertFalse(pullService.isRunning());
+ }
- @Test(expected = NullPointerException.class)
- public void start() {
+ @Test(expected = NullPointerException.class)
+ public void start() {
+ setRunning(false);
+ pullService.start();
+ setRunning(true);
+ pullService.start();
+ }
- //when(config.getKafkaConsumerCount()).thenReturn(1);
+ @Test
+ public void shutdown() {
+ when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+ setRunning(false);
+ pullService.shutdown();
+ setRunning(true);
+ pullService.shutdown();
+ }
- pullService.start();
- }
+ private void setRunning(boolean running) {
+ Field configField;
+ try {
+ configField = PullService.class.getDeclaredField("isRunning");
+ configField.setAccessible(true);
+ configField.set(pullService, running);
+ } catch (IllegalArgumentException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ } catch (IllegalAccessException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
- @Test
- public void shutdown() {
- pullService.shutdown();
- }
+ } catch (NoSuchFieldException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ } catch (SecurityException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
}
\ No newline at end of file
import static org.mockito.Mockito.when;
+import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.util.TestUtil;
import org.springframework.context.ApplicationContext;
/**
@RunWith(MockitoJUnitRunner.class)
public class PullerTest {
- @InjectMocks
- private Puller puller = new Puller(null);
+ private Puller puller;
@Mock
private ApplicationContext context;
@Mock
private TopicConfigPollingService topicConfigPollingService;
- public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException {
- when(config.isAsync()).thenReturn(true);
+ @Before
+ public void init() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+ Kafka kafka = TestUtil.newKafka("kafka");
+ kafka.setBrokerList("brokerList:1,brokerList2:1");
+ kafka.setGroup("group");
+ kafka.setLogin("login");
+ kafka.setSecure(true);
+ kafka.setSecurityProtocol("securityProtocol");
+ puller = new Puller(kafka);
+
+ Field configField = Puller.class.getDeclaredField("config");
+ configField.setAccessible(true);
+ configField.set(puller, config);
- Method init = puller.getClass().getDeclaredMethod("init");
- init.setAccessible(true);
- init.invoke(puller);
+ when(config.isAsync()).thenReturn(true);
+ Method initMethod = Puller.class.getDeclaredMethod("init");
+ initMethod.setAccessible(true);
+ initMethod.invoke(puller);
}
@Test
- public void testRun() throws InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
- testInit();
-
-
+ public void testRun() throws InterruptedException {
Thread thread = new Thread(puller);
thread.start();
Thread.sleep(50);
puller.shutdown();
thread.join();
-
}
}
\ No newline at end of file
package org.onap.datalake.feeder.service;
-import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.when;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import org.apache.commons.lang3.tuple.Pair;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.EffectiveTopic;
import org.onap.datalake.feeder.domain.Kafka;
-import org.onap.datalake.feeder.dto.TopicConfig;
-import org.onap.datalake.feeder.service.db.*;
+import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.service.db.CouchbaseService;
+import org.onap.datalake.feeder.service.db.ElasticsearchService;
+import org.onap.datalake.feeder.service.db.HdfsService;
+import org.onap.datalake.feeder.service.db.MongodbService;
+import org.onap.datalake.feeder.util.TestUtil;
import org.springframework.context.ApplicationContext;
/**
@Mock
private HdfsService hdfsService;
-
+
@Mock
private Kafka kafka;
- public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+ @Before
+ public void init() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
Method init = storeService.getClass().getDeclaredMethod("init");
init.setAccessible(true);
init.invoke(storeService);
}
- private TopicConfig createTopicConfig(String topicStr, String type) {
+ private EffectiveTopic createTopicConfig(String topicStr, String type) {
+ Topic topic = new Topic();
+ topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
+ topic.setDataFormat(type);
+ topic.setSaveRaw(true);
+ topic.setEnabled(true);
+
- TopicConfig topicConfig = new TopicConfig();
- topicConfig.setName(topicStr);
- topicConfig.setDataFormat(type);
- topicConfig.setSaveRaw(true);
+ EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+ List<EffectiveTopic> effectiveTopics = new ArrayList<>();
+ effectiveTopics.add(effectiveTopic);
-// when(configPollingService.getEffectiveTopicConfig(topicStr)).thenReturn(topicConfig);
+ when(configPollingService.getEffectiveTopic(kafka, topicStr)).thenReturn(effectiveTopics);
- return topicConfig;
+ return effectiveTopic;
}
@Test
public void saveMessages() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
- testInit();
-
- TopicConfig topicConfig = createTopicConfig("test1", "JSON");
- topicConfig.setAggregateArrayPath("/test");
- topicConfig.setFlattenArrayPath("/test");
-
- topicConfig = createTopicConfig("test2", "XML");
- topicConfig.setSaveRaw(false);
-
- topicConfig = createTopicConfig("test3", "YAML");
-
- topicConfig.setSinkdbs(new ArrayList<>());
- topicConfig.getSinkdbs().add("Elasticsearch");
- topicConfig.getSinkdbs().add("Couchbase");
- topicConfig.getSinkdbs().add("Druid");
- topicConfig.getSinkdbs().add("MongoDB");
- topicConfig.getSinkdbs().add("HDFS");
-
-
- topicConfig.setEnabledSinkdbs(new ArrayList<>());
- topicConfig.getEnabledSinkdbs().add("Elasticsearch");
+ EffectiveTopic effectiveTopic = createTopicConfig("test1", "JSON");
+ effectiveTopic.getTopic().setAggregateArrayPath("/test");
+ effectiveTopic.getTopic().setFlattenArrayPath("/test");
+
+ effectiveTopic = createTopicConfig("test2", "XML");
+ effectiveTopic.getTopic().setSaveRaw(false);
+
+ effectiveTopic = createTopicConfig("test3", "YAML");
+ effectiveTopic.getTopic().setDbs(new HashSet<>());
+ effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("ES"));
+ effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("CB"));
+ effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("DRUID"));
+ effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("MONGO"));
+ effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("HDFS"));
+ // effectiveTopic.getTopic().setEnabledSinkdbs(new ArrayList<>());
+ // effectiveTopic.getTopic().getEnabledSinkdbs().add("Elasticsearch");
//assertTrue(topicConfig.supportElasticsearch());
-
-
+
createTopicConfig("test4", "TEXT");
+
+ effectiveTopic = createTopicConfig("test5", "TEXT");
+ effectiveTopic.getTopic().setEnabled(false);
-// when(config.getTimestampLabel()).thenReturn("ts");
-// when(config.getRawDataLabel()).thenReturn("raw");
+ when(config.getTimestampLabel()).thenReturn("ts");
+ when(config.getRawDataLabel()).thenReturn("raw");
//JSON
List<Pair<Long, String>> messages = new ArrayList<>();
//XML
List<Pair<Long, String>> messagesXml = new ArrayList<>();
- messagesXml.add(Pair.of(100L, "<test></test>"));
+ messagesXml.add(Pair.of(100L, "<test></test>"));
messagesXml.add(Pair.of(100L, "<test></test"));//bad xml to trigger exception
storeService.saveMessages(kafka, "test2", messagesXml);
messagesText.add(Pair.of(100L, "test message"));
storeService.saveMessages(kafka, "test4", messagesText);
+
+ storeService.saveMessages(kafka, "test5", messagesText);
//Null mesg
storeService.saveMessages(kafka, "test", null);
package org.onap.datalake.feeder.service;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.when;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
-import java.util.List;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.util.TestUtil;
/**
* Test TopicConfigPollingService
@InjectMocks
private TopicConfigPollingService topicConfigPollingService = new TopicConfigPollingService();
- @Test
- public void testRun() {
-
- }
-
- /*
- public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+ static String KAFKA_NAME = "kafka1";
+
+ @Before
+ public void init() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
Method init = topicConfigPollingService.getClass().getDeclaredMethod("init");
init.setAccessible(true);
init.invoke(topicConfigPollingService);
- List<String> activeTopics = Arrays.asList("test");
- Field activeTopicsField = topicConfigPollingService.getClass().getDeclaredField("activeTopics");
+ Set<String> activeTopics = new HashSet<>(Arrays.asList("test"));
+ Map<String, Set<String>> activeTopicMap = new HashMap<>();
+ activeTopicMap.put(KAFKA_NAME, activeTopics);
+
+ Field activeTopicsField = TopicConfigPollingService.class.getDeclaredField("activeTopicMap");
activeTopicsField.setAccessible(true);
- activeTopicsField.set(topicConfigPollingService, activeTopics);
+ activeTopicsField.set(topicConfigPollingService, activeTopicMap);
+
+ Method initMethod = TopicConfigPollingService.class.getDeclaredMethod("init");
+ initMethod.setAccessible(true);
+ initMethod.invoke(topicConfigPollingService);
}
@Test
- public void testRun() throws InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
- testInit();
+ public void testRun() throws InterruptedException {
- //when(config.getDmaapCheckNewTopicInterval()).thenReturn(1);
+ when(config.getCheckTopicInterval()).thenReturn(1L);
Thread thread = new Thread(topicConfigPollingService);
thread.start();
@Test
public void testRunNoChange() throws InterruptedException {
-
-// when(config.getDmaapCheckNewTopicInterval()).thenReturn(1);
+
+ when(config.getCheckTopicInterval()).thenReturn(1L);
Thread thread = new Thread(topicConfigPollingService);
thread.start();
topicConfigPollingService.shutdown();
thread.join();
- assertFalse(topicConfigPollingService.isActiveTopicsChanged(new Kafka()));
+ assertTrue(topicConfigPollingService.isActiveTopicsChanged(new Kafka()));
}
@Test
public void testGet() {
- Kafka kafka=null;
- assertNull(topicConfigPollingService.getEffectiveTopic (new Kafka(), "test"));
- assertNull(topicConfigPollingService.getActiveTopics(kafka));
+ Kafka kafka = TestUtil.newKafka(KAFKA_NAME);
+ //assertNull(topicConfigPollingService.getEffectiveTopic (kafka, "test"));
+ assertNotNull(topicConfigPollingService.getActiveTopics(kafka));
}
- */
+
}
\ No newline at end of file
package org.onap.datalake.feeder.service.db;
-import com.couchbase.client.java.Cluster;
-import com.couchbase.client.java.CouchbaseCluster;
-import com.couchbase.client.java.env.DefaultCouchbaseEnvironment;
-import com.couchbase.mock.Bucket;
-import com.couchbase.mock.BucketConfiguration;
-import com.couchbase.mock.CouchbaseMock;
-import com.couchbase.mock.client.MockClient;
+import java.util.ArrayList;
+import java.util.List;
+
import org.jetbrains.annotations.NotNull;
import org.json.JSONObject;
import org.junit.After;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Db;
-import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.domain.EffectiveTopic;
import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.service.db.CouchbaseService;
import org.onap.datalake.feeder.util.TestUtil;
-import static org.mockito.Mockito.when;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import com.couchbase.client.java.Cluster;
+import com.couchbase.client.java.CouchbaseCluster;
+import com.couchbase.client.java.env.DefaultCouchbaseEnvironment;
+import com.couchbase.mock.Bucket;
+import com.couchbase.mock.BucketConfiguration;
+import com.couchbase.mock.CouchbaseMock;
+import com.couchbase.mock.client.MockClient;
@RunWith(MockitoJUnitRunner.class)
public class CouchbaseServiceTest {
- protected final BucketConfiguration bucketConfiguration = new BucketConfiguration();
- protected MockClient mockClient;
- protected CouchbaseMock couchbaseMock;
- protected Cluster cluster;
- protected com.couchbase.client.java.Bucket bucket;
- protected int carrierPort;
- protected int httpPort;
-
- protected void getPortInfo(String bucket) throws Exception {
- httpPort = couchbaseMock.getHttpPort();
- carrierPort = couchbaseMock.getCarrierPort(bucket);
- }
-
- protected void createMock(@NotNull String name, @NotNull String password) throws Exception {
- bucketConfiguration.numNodes = 1;
- bucketConfiguration.numReplicas = 1;
- bucketConfiguration.numVBuckets = 1024;
- bucketConfiguration.name = name;
- bucketConfiguration.type = Bucket.BucketType.COUCHBASE;
- bucketConfiguration.password = password;
- ArrayList<BucketConfiguration> configList = new ArrayList<BucketConfiguration>();
- configList.add(bucketConfiguration);
- couchbaseMock = new CouchbaseMock(0, configList);
- couchbaseMock.start();
- couchbaseMock.waitForStartup();
- }
-
- protected void createClient() {
- cluster = CouchbaseCluster.create(DefaultCouchbaseEnvironment.builder()
- .bootstrapCarrierDirectPort(carrierPort)
- .bootstrapHttpDirectPort(httpPort)
- .build(), "couchbase://127.0.0.1");
- bucket = cluster.openBucket("default");
- }
-
- @Before
- public void setUp() throws Exception {
- createMock("default", "");
- getPortInfo("default");
- createClient();
- }
-
- @After
- public void tearDown() {
- if (cluster != null) {
- cluster.disconnect();
- }
- if (couchbaseMock != null) {
- couchbaseMock.stop();
- }
- if (mockClient != null) {
- mockClient.shutdown();
- }
- }
-
- @Test
- public void testSaveJsonsWithTopicId() {
- ApplicationConfiguration appConfig = new ApplicationConfiguration();
- appConfig.setTimestampLabel("datalake_ts_");
-
- String text = "{ data: { data2 : { value : 'hello'}}}";
-
- JSONObject json = new JSONObject(text);
-
- Topic topic = TestUtil.newTopic("test getMessageId");
- topic.setMessageIdPath("/data/data2/value");
- List<JSONObject> jsons = new ArrayList<>();
- json.put(appConfig.getTimestampLabel(), 1234);
- jsons.add(json);
- CouchbaseService couchbaseService = new CouchbaseService(new Db());
- couchbaseService.bucket = bucket;
- couchbaseService.config = appConfig;
- // couchbaseService.saveJsons(topic.getTopicConfig(), jsons);
-
- }
-
- @Test
- public void testSaveJsonsWithOutTopicId() {
- ApplicationConfiguration appConfig = new ApplicationConfiguration();
- appConfig.setTimestampLabel("datalake_ts_");
-
- String text = "{ data: { data2 : { value : 'hello'}}}";
-
- JSONObject json = new JSONObject(text);
-
- Topic topic = TestUtil.newTopic("test getMessageId");
- List<JSONObject> jsons = new ArrayList<>();
- json.put(appConfig.getTimestampLabel(), 1234);
- jsons.add(json);
- CouchbaseService couchbaseService = new CouchbaseService(new Db());
- couchbaseService.bucket = bucket;
- couchbaseService.config = appConfig;
-// couchbaseService.saveJsons(topic.getTopicConfig(), jsons);
- }
-
- @Test
- public void testCleanupBucket() {
- CouchbaseService couchbaseService = new CouchbaseService(new Db());
- couchbaseService.bucket = bucket;
- ApplicationConfiguration appConfig = new ApplicationConfiguration();
- couchbaseService.config = appConfig;
- couchbaseService.cleanUp();
- }
+ protected final BucketConfiguration bucketConfiguration = new BucketConfiguration();
+ protected MockClient mockClient;
+ protected CouchbaseMock couchbaseMock;
+ protected Cluster cluster;
+ protected com.couchbase.client.java.Bucket bucket;
+ protected int carrierPort;
+ protected int httpPort;
+
+ protected void getPortInfo(String bucket) throws Exception {
+ httpPort = couchbaseMock.getHttpPort();
+ carrierPort = couchbaseMock.getCarrierPort(bucket);
+ }
+
+ protected void createMock(@NotNull String name, @NotNull String password) throws Exception {
+ bucketConfiguration.numNodes = 1;
+ bucketConfiguration.numReplicas = 1;
+ bucketConfiguration.numVBuckets = 1024;
+ bucketConfiguration.name = name;
+ bucketConfiguration.type = Bucket.BucketType.COUCHBASE;
+ bucketConfiguration.password = password;
+ ArrayList<BucketConfiguration> configList = new ArrayList<BucketConfiguration>();
+ configList.add(bucketConfiguration);
+ couchbaseMock = new CouchbaseMock(0, configList);
+ couchbaseMock.start();
+ couchbaseMock.waitForStartup();
+ }
+
+ protected void createClient() {
+ cluster = CouchbaseCluster.create(DefaultCouchbaseEnvironment.builder().bootstrapCarrierDirectPort(carrierPort).bootstrapHttpDirectPort(httpPort).build(), "couchbase://127.0.0.1");
+ bucket = cluster.openBucket("default");
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ createMock("default", "");
+ getPortInfo("default");
+ createClient();
+ }
+
+ @After
+ public void tearDown() {
+ if (cluster != null) {
+ cluster.disconnect();
+ }
+ if (couchbaseMock != null) {
+ couchbaseMock.stop();
+ }
+ if (mockClient != null) {
+ mockClient.shutdown();
+ }
+ }
+
+ @Test
+ public void testSaveJsonsWithTopicId() {
+ ApplicationConfiguration appConfig = new ApplicationConfiguration();
+ appConfig.setTimestampLabel("datalake_ts_");
+
+ String text = "{ data: { data2 : { value : 'hello'}}}";
+
+ JSONObject json = new JSONObject(text);
+
+ Topic topic = TestUtil.newTopic("test getMessageId");
+ topic.setMessageIdPath("/data/data2/value");
+ List<JSONObject> jsons = new ArrayList<>();
+ json.put(appConfig.getTimestampLabel(), 1234);
+ jsons.add(json);
+ CouchbaseService couchbaseService = new CouchbaseService(new Db());
+ couchbaseService.bucket = bucket;
+ couchbaseService.config = appConfig;
+
+ couchbaseService.init();
+ EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+ couchbaseService.saveJsons(effectiveTopic, jsons);
+
+ }
+
+ @Test
+ public void testSaveJsonsWithOutTopicId() {
+ ApplicationConfiguration appConfig = new ApplicationConfiguration();
+ appConfig.setTimestampLabel("datalake_ts_");
+
+ String text = "{ data: { data2 : { value : 'hello'}}}";
+
+ JSONObject json = new JSONObject(text);
+
+ Topic topic = TestUtil.newTopic("test getMessageId");
+ List<JSONObject> jsons = new ArrayList<>();
+ json.put(appConfig.getTimestampLabel(), 1234);
+ jsons.add(json);
+ CouchbaseService couchbaseService = new CouchbaseService(new Db());
+ couchbaseService.bucket = bucket;
+ couchbaseService.config = appConfig;
+
+ couchbaseService.init();
+ EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+ couchbaseService.saveJsons(effectiveTopic, jsons);
+ }
+
+ @Test
+ public void testCleanupBucket() {
+ CouchbaseService couchbaseService = new CouchbaseService(new Db());
+ couchbaseService.bucket = bucket;
+ ApplicationConfiguration appConfig = new ApplicationConfiguration();
+ couchbaseService.config = appConfig;
+
+ couchbaseService.init();
+ couchbaseService.cleanUp();
+ }
}
\ No newline at end of file
package org.onap.datalake.feeder.service.db;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.RestHighLevelClient;
-import org.json.JSONObject;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.domain.Db;
import org.onap.datalake.feeder.service.DbService;
-import org.onap.datalake.feeder.service.db.ElasticsearchService;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import static org.mockito.Mockito.when;
+import org.onap.datalake.feeder.util.TestUtil;
@RunWith(MockitoJUnitRunner.class)
public class ElasticsearchServiceTest {
+ @Mock
+ private ApplicationConfiguration config;
- static String DEFAULT_TOPIC_NAME = "_DL_DEFAULT_";
-
- @InjectMocks
- private ElasticsearchService elasticsearchService;
-
- @Mock
- private ApplicationConfiguration config;
-
- @Mock
- private RestHighLevelClient client;
-
- @Mock
- ActionListener<BulkResponse> listener;
-
- @Mock
- private DbService dbService;
+ @Mock
+ private RestHighLevelClient client;
- @Test(expected = NullPointerException.class)
- public void testCleanUp() throws IOException {
+ @Mock
+ ActionListener<BulkResponse> listener;
- elasticsearchService.cleanUp();
+ @Mock
+ private DbService dbService;
- }
+ private ElasticsearchService elasticsearchService;
- @Test(expected = NullPointerException.class)
- public void testEnsureTableExist() throws IOException {
+ @Before
+ public void init() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+ //MockitoAnnotations.initMocks(this);
- elasticsearchService.ensureTableExist(DEFAULT_TOPIC_NAME);
- }
+ Db db = TestUtil.newDb("Elasticsearch");
+ db.setHost("host");
+ elasticsearchService = new ElasticsearchService(db);
- @Test
- public void testSaveJsons() {
+ Field configField = ElasticsearchService.class.getDeclaredField("config");
+ configField.setAccessible(true);
+ configField.set(elasticsearchService, config);
+
+ elasticsearchService.init();
+ }
- Topic topic = new Topic();
- topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
- topic.setCorrelateClearedMessage(true);
- topic.setMessageIdPath("/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem");
- String jsonString = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailure\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
- String jsonString2 = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailureCleared\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+ @Test
+ public void testCleanUp() throws IOException {
+ when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+ elasticsearchService.cleanUp();
+ }
- JSONObject jsonObject = new JSONObject(jsonString);
- JSONObject jsonObject2 = new JSONObject(jsonString2);
+ @Test(expected = IOException.class)
+ public void testEnsureTableExist() throws IOException {
+ elasticsearchService.ensureTableExist("test");
+ }
- List<JSONObject> jsons = new ArrayList<>();
- jsons.add(jsonObject);
- jsons.add(jsonObject2);
-// when(config.getElasticsearchType()).thenReturn("doc");
- // when(config.isAsync()).thenReturn(true);
+ @Test
+ public void testSaveJsons() {
+ when(config.getElasticsearchType()).thenReturn("doc");
- //elasticsearchService.saveJsons(topic.getTopicConfig(), jsons);
+ when(config.isAsync()).thenReturn(true);
+ TestUtil.testSaveJsons(config, elasticsearchService);
- }
+ when(config.isAsync()).thenReturn(false);
+ TestUtil.testSaveJsons(config, elasticsearchService);
+ }
}
\ No newline at end of file
import static org.mockito.Mockito.when;
-import java.util.ArrayList;
-import java.util.List;
+import java.lang.reflect.Field;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.commons.lang3.tuple.Pair;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.dto.TopicConfig;
-import org.onap.datalake.feeder.service.db.HdfsService;
+import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.util.TestUtil;
import org.springframework.context.ApplicationContext;
/**
*/
@RunWith(MockitoJUnitRunner.class)
public class HdfsServiceTest {
-
- @InjectMocks
private HdfsService hdfsService;
@Mock
@Mock
private ExecutorService executorService;
- @Test
- public void saveMessages() {
- TopicConfig topicConfig = new TopicConfig();
- topicConfig.setName("test");
+ @Before
+ public void init() throws NoSuchFieldException, IllegalAccessException {
+ Db db = TestUtil.newDb("HDFS");
+ db.setHost("host");
+ db.setLogin("login");
+ hdfsService = new HdfsService(db);
+
+ Field configField = HdfsService.class.getDeclaredField("config");
+ configField.setAccessible(true);
+ configField.set(hdfsService, config);
+
+ hdfsService.init();
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void saveJsons() {
+ when(config.getHdfsBufferSize()).thenReturn(1000);
- List<Pair<Long, String>> messages = new ArrayList<>();
- messages.add(Pair.of(100L, "test message"));
+ when(config.isAsync()).thenReturn(true);
+ TestUtil.testSaveJsons(config , hdfsService);
- //when(config.getHdfsBufferSize()).thenReturn(1000);
- //hdfsService.saveMessages(topicConfig, messages);
+ when(config.isAsync()).thenReturn(false);
+ TestUtil.testSaveJsons(config , hdfsService);
}
@Test(expected = NullPointerException.class)
public void cleanUp() {
+ when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
hdfsService.flush();
hdfsService.flushStall();
hdfsService.cleanUp();
package org.onap.datalake.feeder.service.db;
-import com.mongodb.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
import org.bson.Document;
-import org.json.JSONObject;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.domain.Db;
import org.onap.datalake.feeder.service.DbService;
-import org.onap.datalake.feeder.service.db.MongodbService;
-
-import static org.mockito.Mockito.when;
+import org.onap.datalake.feeder.util.TestUtil;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import com.mongodb.MongoClient;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
@RunWith(MockitoJUnitRunner.class)
public class MongodbServiceTest {
- @InjectMocks
- private MongodbService mongodbService;
-
- @Mock
- private ApplicationConfiguration config;
-
- @Mock
- private DbService dbService;
-
- @Mock
- private MongoDatabase database;
+ private MongodbService mongodbService;
- @Mock
- private MongoClient mongoClient;
+ @Mock
+ private ApplicationConfiguration config;
- @Mock
- private Map<String, MongoCollection<Document>> mongoCollectionMap = new HashMap<>();
+ @Mock
+ private DbService dbService;
+ @Mock
+ private MongoDatabase database;
- @Test
- public void cleanUp() {
- // when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
-// mongodbService.cleanUp();
- }
+ @Mock
+ private MongoClient mongoClient;
- @Test
- public void saveJsons() {
+ @Mock
+ private Map<String, MongoCollection<Document>> mongoCollectionMap = new HashMap<>();
- Topic topic = new Topic();
- topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
- topic.setCorrelateClearedMessage(true);
- topic.setMessageIdPath("/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem");
- String jsonString = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailure\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
- String jsonString2 = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailureCleared\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+ @Before
+ public void init() throws NoSuchFieldException, IllegalAccessException {
+ Db db = TestUtil.newDb("Mongodb");
+ db.setDatabase("database");
+ db.setLogin("login");
+ mongodbService = new MongodbService(db);
- JSONObject jsonObject = new JSONObject(jsonString);
- JSONObject jsonObject2 = new JSONObject(jsonString2);
+ Field configField = MongodbService.class.getDeclaredField("config");
+ configField.setAccessible(true);
+ configField.set(mongodbService, config);
+
+ mongodbService.init();
+ }
- List<JSONObject> jsons = new ArrayList<>();
- jsons.add(jsonObject);
- jsons.add(jsonObject2);
+ @Test
+ public void cleanUp() {
+ when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+ mongodbService.cleanUp();
+ }
- //mongodbService.saveJsons(topic.getTopicConfig(), jsons);
- }
+ @Test
+ public void saveJsons() {
+ TestUtil.testSaveJsons(config, mongodbService);
+ }
}
\ No newline at end of file
*/
package org.onap.datalake.feeder.util;
+
+import java.util.ArrayList;
+import java.util.List;
-import org.junit.Test;
+import org.json.JSONObject;
+import org.onap.datalake.feeder.config.ApplicationConfiguration;
import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.domain.DbType;
+import org.onap.datalake.feeder.domain.EffectiveTopic;
+import org.onap.datalake.feeder.domain.Kafka;
import org.onap.datalake.feeder.domain.Topic;
import org.onap.datalake.feeder.domain.TopicName;
-
-import java.io.IOException;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import org.onap.datalake.feeder.service.db.DbStoreService;
+
/**
* test utils
static int i=0;
+ public static Kafka newKafka(String name) {
+ Kafka kafka = new Kafka();
+ kafka.setId(name);
+ return kafka ;
+ }
+
public static Db newDb(String name) {
Db db = new Db();
db.setId(i++);
- db.setName(name);
+ db.setName(name);
+ db.setDbType(new DbType(name, name));
return db;
}
return topic;
}
+ public static void testSaveJsons(ApplicationConfiguration config, DbStoreService dbStoreService) {
+ Topic topic = new Topic();
+ topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
+ topic.setCorrelateClearedMessage(true);
+ topic.setMessageIdPath("/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem");
+ String jsonString = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailure\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+ String jsonString2 = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailureCleared\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+
+ JSONObject jsonObject = new JSONObject(jsonString);
+ JSONObject jsonObject2 = new JSONObject(jsonString2);
+
+ List<JSONObject> jsons = new ArrayList<>();
+ jsons.add(jsonObject);
+ jsons.add(jsonObject2);
+
+ EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+
+ dbStoreService.saveJsons(effectiveTopic, jsons);
+ }
}
Code Review / dcaegen2 / services.git / commitdiff