Unit test for seed code

This project depends heavily on outside
infrastructure, thus there are not much
unit test cases.

Issue-ID: DCAEGEN2-1309
Change-Id: I8893f4029014673139e67bf373fd77e9e5c42f83
Signed-off-by: Guobiao Mo <guobiaomo@chinamobile.com>

5 files changed, 128 insertions, 21 deletions

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java
index 62ac37fb..108eb4e0 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java
@@ -20,10 +20,13 @@
 
 package org.onap.datalake.feeder.config;
 
-import java.util.Set;
-
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.couchbase.CouchbaseConfiguration;
 import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.FilterType;
+import org.springframework.context.annotation.ComponentScan;
 
 import lombok.Getter;
 import lombok.Setter;
@@ -37,8 +40,12 @@ import lombok.Setter;
  */
 @Getter
 @Setter
-@Configuration
+@SpringBootConfiguration
 @ConfigurationProperties
+//@ComponentScan(excludeFilters = @ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, value = CouchbaseConfiguration.class))
+//https://stackoverflow.com/questions/29344313/prevent-application-commandlinerunner-classes-from-executing-during-junit-test
+@EnableAutoConfiguration
+//@Profile("test")
 public class ApplicationConfiguration {
 
 	private String couchbaseHost;
@@ -49,10 +56,6 @@ public class ApplicationConfiguration {
 	//    private int mongodbPort;    
 	//  private String mongodbDatabase;    
 
-	private boolean storeJson;
-	private boolean storeYaml;
-	private boolean storeXml;
-
 	private String dmaapZookeeperHostPort;
 	private String dmaapKafkaHostPort;
 	private String dmaapKafkaGroup;
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
index 99216ad3..ace33dcc 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
@@ -23,10 +23,14 @@ import java.util.function.Predicate;
 
 import javax.validation.constraints.NotNull;
 
+import org.apache.commons.lang3.StringUtils;
+import org.json.JSONObject;
 import org.onap.datalake.feeder.enumeration.DataFormat;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Transient;
 import org.springframework.data.couchbase.core.mapping.Document;
+
+import lombok.Setter;
  
 /**
  * Domain class representing topic table in Couchbase
@@ -35,6 +39,7 @@ import org.springframework.data.couchbase.core.mapping.Document;
  *
  */
 @Document
+@Setter
 public class Topic {
 	@NotNull
 	@Id
@@ -77,6 +82,9 @@ public class Topic {
 	
 	//if this flag is true, need to correlate alarm cleared message to previous alarm 
 	private Boolean correlateClearedMessage;
+	
+	//the value in the JSON with this path will be used as DB id
+	private String messageIdPath;
 
 	public Topic() {
 	}
@@ -123,12 +131,16 @@ public class Topic {
 
 	//if 'this' Topic does not have the setting, use default Topic's
 	private boolean is(Boolean b, Predicate<Topic> pre) {
+		return is(b, pre, false);
+	}
+	
+	private boolean is(Boolean b, Predicate<Topic> pre, boolean defaultValue) {
 		if (b != null) {
 			return b;
 		} else if (defaultTopic != null) {
 			return pre.test(defaultTopic);
 		} else {
-			return false;
+			return defaultValue;
 		}
 	}
 
@@ -148,18 +160,33 @@ public class Topic {
 		return is(supportDruid, Topic::isSupportDruid);
 	}
 
+	//extract DB id from a JSON attribute, TODO support multiple attributes
+	public String getMessageId(JSONObject json) {
+		String id = null;
+		
+		if(StringUtils.isNotBlank(messageIdPath)) {
+			id = json.query(messageIdPath).toString();
+		}
+		
+		return id;
+	}
+	
 	@Override
 	public String toString() {
 		return id;
 	}
 
-	// for testing
-	public static void main(String[] args) {
-		Topic defaultTopic=new Topic("def");
-		Topic test = new Topic("test");
-		test.setDefaultTopic(defaultTopic);
-		defaultTopic.supportElasticsearch=true;
-		boolean b = test.isSupportElasticsearch();
-		System.out.println(b);
+	/**
+	 * @return the messageIdPath
+	 */
+	public String getMessageIdPath() {
+		return messageIdPath;
+	}
+
+	/**
+	 * @param messageIdPath the messageIdPath to set
+	 */
+	public void setMessageIdPath(String messageIdPath) {
+		this.messageIdPath = messageIdPath;
 	}
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DataFormat.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DataFormat.java
index 83ffac18..fdcbdfc1 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DataFormat.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DataFormat.java
@@ -27,4 +27,20 @@ package org.onap.datalake.feeder.enumeration;
  */
 public enum DataFormat {
 	JSON, XML, YAML, TEXT; 	
+
+    public static DataFormat fromString(String s) {
+        if ("JSON".equalsIgnoreCase(s)) {
+            return JSON;
+        }
+        if ("XML".equalsIgnoreCase(s)) {
+            return XML;
+        }
+        if ("YAML".equalsIgnoreCase(s)) {
+            return YAML;
+        }
+        if ("TEXT".equalsIgnoreCase(s)) {
+            return TEXT;
+        }
+        throw new IllegalArgumentException("Invalid value for format: " + s);
+    }
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java
index 819590b7..31f46362 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java
@@ -33,11 +33,15 @@ import org.apache.velocity.VelocityContext;
 import org.apache.velocity.app.Velocity;

 import org.apache.velocity.runtime.RuntimeConstants;

 import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;

+import org.onap.datalake.feeder.enumeration.DataFormat;

 

 import com.fasterxml.jackson.databind.JsonNode;

 import com.fasterxml.jackson.databind.ObjectMapper;

 import com.fasterxml.jackson.databind.node.JsonNodeType;

 

+import lombok.Getter;

+import lombok.Setter;

+

 

 /*

  * read sample json and output supervisor to  resources\druid\generated

@@ -56,6 +60,7 @@ import com.fasterxml.jackson.databind.node.JsonNodeType;
  * dimension type default is string, in msgrtr.apinode.metrics.dmaap , many are long/double, so need to generate dimensionsSpec, this is done at the end of printFlattenSpec()

  */

 

+@Getter

 public class DruidSupervisorGenerator {

 

 	Template template = null;

@@ -73,12 +78,12 @@ public class DruidSupervisorGenerator {
 

 		context = new VelocityContext();

 

-		context.put("host", "dl_dmaap_kf");

+		context.put("host", "message-router-kafka:9092");//TODO get from config

 

 		template = Velocity.getTemplate("druid/kafka-supervisor-template.vm");

 	}

 

-	public void printNode(String prefix, JsonNode node) {

+	private void printNode(String prefix, JsonNode node) {

 

 		// lets see what type the node is

 		//		System.out.println("NodeType=" + node.getNodeType() + ", isContainerNode=" + node.isContainerNode() + ", " + node); // prints OBJECT

@@ -109,7 +114,7 @@ public class DruidSupervisorGenerator {
 

 	}

 

-	public void printFlattenSpec(JsonNodeType type, String path) {

+	private void printFlattenSpec(JsonNodeType type, String path) {

 		String name = path.substring(2).replace('.', ':');

 		// lets see what type the node is

 		System.out.println("{");

@@ -151,7 +156,6 @@ public class DruidSupervisorGenerator {
 		context.put("topic", topic);

 		context.put("timestamp", "event-header:timestamp");//FIXME hard coded, should be topic based

 		context.put("timestampFormat", "yyyyMMdd-HH:mm:ss:SSS");//FIXME hard coded, should be topic based

-

 		context.put("dimensions", dimensions);

 

 		BufferedWriter out = new BufferedWriter(new FileWriter(outputFileName));

diff --git a/components/datalake-handler/feeder/src/main/resources/druid/kafka-supervisor-template.vm b/components/datalake-handler/feeder/src/main/resources/druid/kafka-supervisor-template.vm
new file mode 100644
index 00000000..01ebaf2e
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/resources/druid/kafka-supervisor-template.vm
@@ -0,0 +1,57 @@
+{
+	"type": "kafka",
+	"dataSchema": {
+		"dataSource": "$topic",
+		"parser": {
+			"type": "string",
+			"parseSpec": {
+				"format": "json",
+				"flattenSpec": {
+					"useFieldDiscovery": false,
+					"fields": [
+					#foreach($flatten in $dimensions)
+					
+					{
+"type": "path",
+"name": "$flatten[0]",
+"expr": "$flatten[1]"
+},
+					#end
+					]
+				},
+				"timestampSpec": {
+					"column": "$timestamp",
+					"format": "$timestampFormat"
+				},
+				"dimensionsSpec": {
+					"dimensions": [
+					],
+					"dimensionsExclusions": [
+					]
+				}
+			}
+		},
+		"metricsSpec": [],
+		"granularitySpec": {
+			"type": "uniform",
+			"segmentGranularity": "HOUR",
+			"queryGranularity": "MINUTE",
+			"rollup": false
+		}
+	},
+	"tuningConfig": {
+		"type": "kafka",
+		"reportParseExceptions": true
+	},
+	"ioConfig": {
+		"topic": "$topic",
+		"replicas": 1,
+		"startDelay": "PT1S",
+		"taskDuration": "PT1H",
+		"completionTimeout": "PT30M",
+		"consumerProperties": {
+			"bootstrap.servers": "$host"
+		},
+		"useEarliestOffset": true
+	}
+}