// Path: src/main/java/com/ottogroup/bi/streaming/operator/json/JsonContentReference.java
// public class JsonContentReference implements Serializable {
// 
// 	private static final long serialVersionUID = -5075723737186875253L;
// 
// 	private String[] path = null;
// 	private JsonContentType contentType = null;
// 	private String conversionPattern = null;
// 	/** enforce strict evaluation and require the field to be present */
// 	private boolean required = false;
// 	
// 	public JsonContentReference() {		
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType) {
// 		this(path, contentType, false);
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType, final boolean required) {
// 		this.path = path;
// 		this.contentType = contentType;
// 		this.required = required;
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType, final String conversionPattern) {
// 		this.path = path;
// 		this.contentType = contentType;
// 		this.conversionPattern = conversionPattern;
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType, final String conversionPattern, final boolean required) {
// 		this.path = path;
// 		this.contentType = contentType;
// 		this.conversionPattern = conversionPattern;
// 		this.required = required;
// 	}
// 
// 	public String[] getPath() {
// 		return path;
// 	}
// 
// 	public void setPath(String[] path) {
// 		this.path = path;
// 	}
// 
// 	public JsonContentType getContentType() {
// 		return contentType;
// 	}
// 
// 	public void setContentType(JsonContentType contentType) {
// 		this.contentType = contentType;
// 	}
// 
// 	public String getConversionPattern() {
// 		return conversionPattern;
// 	}
// 
// 	public void setConversionPattern(String conversionPattern) {
// 		this.conversionPattern = conversionPattern;
// 	}
// 
// 	public boolean isRequired() {
// 		return required;
// 	}
// 
// 	public void setRequired(boolean required) {
// 		this.required = required;
// 	}
// 	
// 	
// }
// 
// Path: src/main/java/com/ottogroup/bi/streaming/operator/json/JsonContentType.java
// public enum JsonContentType implements Serializable {
// 	STRING, INTEGER, DOUBLE, TIMESTAMP, BOOLEAN 
// }


import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.common.functions.util.ListCollector;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.sling.commons.json.JSONException;
import org.apache.sling.commons.json.JSONObject;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import com.ottogroup.bi.streaming.operator.json.JsonContentReference;
import com.ottogroup.bi.streaming.operator.json.JsonContentType;
	 * Test case for {@link WindowedJsonContentAggregator#aggregateValue(java.io.Serializable, java.io.Serializable, com.ottogroup.bi.streaming.operator.json.JsonContentType, ContentAggregator)}
	 * being provided valid values to all parameters 
	 */
	@Test
	public void testAggregateValue_withValidValuesAndAVG() throws Exception {
		
		Pair<Integer, Integer> avgValues = new MutablePair<>(Integer.valueOf(10), Integer.valueOf(3));
		
		Assert.assertEquals(new MutablePair<>(Integer.valueOf(12), Integer.valueOf(4)), new WindowedJsonContentAggregator("test", new AggregatorConfiguration()).
			aggregateValue(Integer.valueOf(2), avgValues, JsonContentType.INTEGER, ContentAggregator.AVG));
	}
	
	/**
	 * Test case for {@link WindowedJsonContentAggregator#aggregateField(JSONObject, FieldAggregationConfiguration, String, Map)} being
	 * provided null as input to json document parameter
	 */
	@Test(expected=JSONException.class)
	public void testAggregateField_withNullJSONDocument() throws Exception {
		new WindowedJsonContentAggregator("test", new AggregatorConfiguration()).aggregateField(null, new FieldAggregationConfiguration(), "groupByKeyPrefix", new HashMap<String, Serializable>());
	}
	
	/**
	 * Test case for {@link WindowedJsonContentAggregator#aggregateField(JSONObject, FieldAggregationConfiguration, String, Map)} being
	 * provided valid values 
	 */
	@Test
	public void testAggregateField_withValidValuesBOOLEAN() throws Exception {
		JSONObject object = new JSONObject("{\"logical\":true, \"double\":\"2.34\", \"int\":10, \"string\":\"test\", \"ts\":\"20160131\"}");
		Assert.assertEquals(Integer.valueOf(1), (Integer)new WindowedJsonContentAggregator("test", new AggregatorConfiguration()).aggregateField(
				object, 
// Path: src/main/java/com/ottogroup/bi/streaming/operator/json/JsonContentReference.java
// public class JsonContentReference implements Serializable {
// 
// 	private static final long serialVersionUID = -5075723737186875253L;
// 
// 	private String[] path = null;
// 	private JsonContentType contentType = null;
// 	private String conversionPattern = null;
// 	/** enforce strict evaluation and require the field to be present */
// 	private boolean required = false;
// 	
// 	public JsonContentReference() {		
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType) {
// 		this(path, contentType, false);
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType, final boolean required) {
// 		this.path = path;
// 		this.contentType = contentType;
// 		this.required = required;
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType, final String conversionPattern) {
// 		this.path = path;
// 		this.contentType = contentType;
// 		this.conversionPattern = conversionPattern;
// 	}
// 	
// 	public JsonContentReference(final String[] path, final JsonContentType contentType, final String conversionPattern, final boolean required) {
// 		this.path = path;
// 		this.contentType = contentType;
// 		this.conversionPattern = conversionPattern;
// 		this.required = required;
// 	}
// 
// 	public String[] getPath() {
// 		return path;
// 	}
// 
// 	public void setPath(String[] path) {
// 		this.path = path;
// 	}
// 
// 	public JsonContentType getContentType() {
// 		return contentType;
// 	}
// 
// 	public void setContentType(JsonContentType contentType) {
// 		this.contentType = contentType;
// 	}
// 
// 	public String getConversionPattern() {
// 		return conversionPattern;
// 	}
// 
// 	public void setConversionPattern(String conversionPattern) {
// 		this.conversionPattern = conversionPattern;
// 	}
// 
// 	public boolean isRequired() {
// 		return required;
// 	}
// 
// 	public void setRequired(boolean required) {
// 		this.required = required;
// 	}
// 	
// 	
// }
// 
// Path: src/main/java/com/ottogroup/bi/streaming/operator/json/JsonContentType.java
// public enum JsonContentType implements Serializable {
// 	STRING, INTEGER, DOUBLE, TIMESTAMP, BOOLEAN 
// }


// Path: src/test/java/com/ottogroup/bi/streaming/operator/json/aggregate/WindowedJsonContentAggregatorTest.java
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.common.functions.util.ListCollector;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.sling.commons.json.JSONException;
import org.apache.sling.commons.json.JSONObject;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import com.ottogroup.bi.streaming.operator.json.JsonContentReference;
import com.ottogroup.bi.streaming.operator.json.JsonContentType;

	 * Test case for {@link WindowedJsonContentAggregator#aggregateValue(java.io.Serializable, java.io.Serializable, com.ottogroup.bi.streaming.operator.json.JsonContentType, ContentAggregator)}
	 * being provided valid values to all parameters 
	 */
	@Test
	public void testAggregateValue_withValidValuesAndAVG() throws Exception {
		
		Pair<Integer, Integer> avgValues = new MutablePair<>(Integer.valueOf(10), Integer.valueOf(3));
		
		Assert.assertEquals(new MutablePair<>(Integer.valueOf(12), Integer.valueOf(4)), new WindowedJsonContentAggregator("test", new AggregatorConfiguration()).
			aggregateValue(Integer.valueOf(2), avgValues, JsonContentType.INTEGER, ContentAggregator.AVG));
	}
	
	/**
	 * Test case for {@link WindowedJsonContentAggregator#aggregateField(JSONObject, FieldAggregationConfiguration, String, Map)} being
	 * provided null as input to json document parameter
	 */
	@Test(expected=JSONException.class)
	public void testAggregateField_withNullJSONDocument() throws Exception {
		new WindowedJsonContentAggregator("test", new AggregatorConfiguration()).aggregateField(null, new FieldAggregationConfiguration(), "groupByKeyPrefix", new HashMap<String, Serializable>());
	}
	
	/**
	 * Test case for {@link WindowedJsonContentAggregator#aggregateField(JSONObject, FieldAggregationConfiguration, String, Map)} being
	 * provided valid values 
	 */
	@Test
	public void testAggregateField_withValidValuesBOOLEAN() throws Exception {
		JSONObject object = new JSONObject("{\"logical\":true, \"double\":\"2.34\", \"int\":10, \"string\":\"test\", \"ts\":\"20160131\"}");
		Assert.assertEquals(Integer.valueOf(1), (Integer)new WindowedJsonContentAggregator("test", new AggregatorConfiguration()).aggregateField(
				object, 
				new FieldAggregationConfiguration("output", new JsonContentReference(new String[]{"logical"}, JsonContentType.BOOLEAN)), 