Init dcae.apod.analytics

Change-Id: Ie62128895a199633d0afa695749d40c940d804b2
Signed-off-by: lj1412 <lji@research.att.com>

13 files changed, 1414 insertions, 0 deletions

diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/BaseAnalyticsTCAIT.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/BaseAnalyticsTCAIT.java
new file mode 100644
index 0000000..4dedf98
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/BaseAnalyticsTCAIT.java
@@ -0,0 +1,141 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.google.common.base.Suppliers;
+import org.junit.BeforeClass;
+import org.openecomp.dcae.analytics.common.AnalyticsConstants;
+import org.openecomp.dcae.analytics.model.util.AnalyticsModelIOUtils;
+import org.openecomp.dcae.analytics.model.util.json.AnalyticsModelObjectMapperSupplier;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppConfig;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.analytics.test.BaseDCAEAnalyticsIT;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ *
+ * @author Rajiv Singla. Creation Date: 10/25/2016.
+ */
+public class BaseAnalyticsTCAIT extends BaseDCAEAnalyticsIT {
+
+    protected static ObjectMapper objectMapper;
+
+    @BeforeClass
+    public static void beforeClass() {
+        final AnalyticsModelObjectMapperSupplier analyticsModelObjectMapperSupplier =
+                new AnalyticsModelObjectMapperSupplier();
+        objectMapper = Suppliers.memoize(analyticsModelObjectMapperSupplier).get();
+        objectMapper.enable(SerializationFeature.INDENT_OUTPUT);
+    }
+
+    protected static final String TCA_CONTROLLER_POLICY_FILE_LOCATION =
+            "data/properties/tca_controller_policy.properties";
+
+    // App Settings
+    protected static final String DCAE_ANALYTICS_TCA_TEST_APP_NAME = "dcae-tca";
+    protected static final String DCAE_ANALYTICS_TCA_TEST_APP_DESC =
+            "DCAE Analytics Threshold Crossing Alert Application";
+    
+    // Subscriber App Preferences
+    protected static final String SUBSCRIBER_HOST_NAME = "mrlocal-mtnjftle01.homer.com";
+    protected static final Integer SUBSCRIBER_PORT_NUMBER = 3905;
+    protected static final String SUBSCRIBER_TOPIC_NAME = "com.dcae.dmaap.mtnje2.DcaeTestVESSub";
+    protected static final String SUBSCRIBER_USERNAME = "m00502@tca.af.dcae.com";
+    protected static final String SUBSCRIBER_PASSWORD = "Te5021abc";
+    protected static final String SUBSCRIBER_HTTP_PROTOCOL = "https";
+    protected static final String SUBSCRIBER_CONTENT_TYPE = "application/json";
+    protected static final Integer SUBSCRIBER_POLLING_INTERVAL = 20000;
+
+    protected static final String SUBSCRIBER_CONSUMER_ID = "c12";
+    protected static final String SUBSCRIBER_CONSUMER_GROUP_NAME = AnalyticsConstants.DMAAP_GROUP_PREFIX + 
+            SUBSCRIBER_CONSUMER_ID;
+    protected static final int SUBSCRIBER_TIMOUT_MS = -1;
+    protected static final int SUBSCRIBER_MESSAGE_LIMIT = -1;
+
+    // Publisher App Preferences
+    protected static final String PUBLISHER_HOST_NAME = "mrlocal-mtnjftle01.homer.com";
+    protected static final Integer PUBLISHER_PORT_NUMBER = 3905;
+    protected static final String PUBLISHER_TOPIC_NAME = "com.dcae.dmaap.mtnje2.DcaeTestVESPub";
+    protected static final String PUBLISHER_USERNAME = "m00502@tca.af.dcae.com";
+    protected static final String PUBLISHER_PASSWORD = "Te5021abc";
+    protected static final String PUBLISHER_HTTP_PROTOCOL = "https";
+    protected static final String PUBLISHER_CONTENT_TYPE = "application/json";
+    protected static final Integer PUBLISHER_BATCH_QUEUE_SIZE = 10;
+    protected static final Integer PUBLISHER_RECOVERY_QUEUE_SIZE = 100000;
+    protected static final Integer PUBLISHER_POLLING_INTERVAL = 20000;
+
+    protected static TCATestAppConfig getTCATestAppConfig() {
+        final TCATestAppConfig tcaTestAppConfig = new TCATestAppConfig();
+        tcaTestAppConfig.setAppName(DCAE_ANALYTICS_TCA_TEST_APP_NAME);
+        tcaTestAppConfig.setAppDescription(DCAE_ANALYTICS_TCA_TEST_APP_DESC);
+        return tcaTestAppConfig;
+    }
+
+    protected static TCATestAppPreferences getTCATestAppPreferences() {
+        final TCATestAppPreferences tcaTestAppPreferences = new TCATestAppPreferences(getTCAPolicyPreferences());
+        tcaTestAppPreferences.setSubscriberHostName(SUBSCRIBER_HOST_NAME);
+        tcaTestAppPreferences.setSubscriberHostPortNumber(SUBSCRIBER_PORT_NUMBER);
+        tcaTestAppPreferences.setSubscriberTopicName(SUBSCRIBER_TOPIC_NAME);
+        tcaTestAppPreferences.setSubscriberUserName(SUBSCRIBER_USERNAME);
+        tcaTestAppPreferences.setSubscriberUserPassword(SUBSCRIBER_PASSWORD);
+        tcaTestAppPreferences.setSubscriberProtocol(SUBSCRIBER_HTTP_PROTOCOL);
+        tcaTestAppPreferences.setSubscriberContentType(SUBSCRIBER_CONTENT_TYPE);
+        tcaTestAppPreferences.setSubscriberConsumerId(SUBSCRIBER_CONSUMER_ID);
+        tcaTestAppPreferences.setSubscriberConsumerGroup(SUBSCRIBER_CONSUMER_GROUP_NAME);
+        tcaTestAppPreferences.setSubscriberTimeoutMS(SUBSCRIBER_TIMOUT_MS);
+        tcaTestAppPreferences.setSubscriberMessageLimit(SUBSCRIBER_MESSAGE_LIMIT);
+        tcaTestAppPreferences.setSubscriberPollingInterval(SUBSCRIBER_POLLING_INTERVAL);
+
+        tcaTestAppPreferences.setPublisherHostName(PUBLISHER_HOST_NAME);
+        tcaTestAppPreferences.setPublisherHostPort(PUBLISHER_PORT_NUMBER);
+        tcaTestAppPreferences.setPublisherTopicName(PUBLISHER_TOPIC_NAME);
+        tcaTestAppPreferences.setPublisherUserName(PUBLISHER_USERNAME);
+        tcaTestAppPreferences.setPublisherUserPassword(PUBLISHER_PASSWORD);
+        tcaTestAppPreferences.setPublisherProtocol(PUBLISHER_HTTP_PROTOCOL);
+        tcaTestAppPreferences.setPublisherContentType(PUBLISHER_CONTENT_TYPE);
+        tcaTestAppPreferences.setPublisherMaxBatchSize(PUBLISHER_BATCH_QUEUE_SIZE);
+        tcaTestAppPreferences.setPublisherMaxRecoveryQueueSize(PUBLISHER_RECOVERY_QUEUE_SIZE);
+        tcaTestAppPreferences.setPublisherPollingInterval(PUBLISHER_POLLING_INTERVAL);
+        return tcaTestAppPreferences;
+    }
+
+
+    protected static Map<String, String> getTCAPolicyPreferences() {
+        final Map<String, String> policyPreferences = new LinkedHashMap<>();
+        final Properties policyPreferencesProps =
+                AnalyticsModelIOUtils.loadPropertiesFile(TCA_CONTROLLER_POLICY_FILE_LOCATION);
+        for (Map.Entry<Object, Object> propEntry : policyPreferencesProps.entrySet()) {
+            policyPreferences.put(propEntry.getKey().toString(), propEntry.getValue().toString());
+        }
+
+        return policyPreferences;
+    }
+
+    protected static String serializeModelToJson(Object model) throws JsonProcessingException {
+        return objectMapper.writeValueAsString(model);
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/BaseAnalyticsTCAUnitTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/BaseAnalyticsTCAUnitTest.java
new file mode 100644
index 0000000..38e4ce1
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/BaseAnalyticsTCAUnitTest.java
@@ -0,0 +1,220 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Suppliers;
+import org.junit.Assert;
+import org.openecomp.dcae.analytics.model.domain.cef.EventListener;
+import org.openecomp.dcae.analytics.model.domain.policy.tca.Direction;
+import org.openecomp.dcae.analytics.model.domain.policy.tca.TCAPolicy;
+import org.openecomp.dcae.analytics.model.domain.policy.tca.Threshold;
+import org.openecomp.dcae.analytics.model.util.json.AnalyticsModelObjectMapperSupplier;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.analytics.test.BaseDCAEAnalyticsUnitTest;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author Rajiv Singla. Creation Date: 10/25/2016.
+ */
+public class BaseAnalyticsTCAUnitTest extends BaseDCAEAnalyticsUnitTest {
+
+    /**
+     * Object mapper to be used for all TCA Json Parsing
+     */
+    protected static final ObjectMapper ANALYTICS_MODEL_OBJECT_MAPPER =
+            Suppliers.memoize(new AnalyticsModelObjectMapperSupplier()).get();
+
+    protected static final String TCA_POLICY_JSON_FILE_LOCATION = "data/json/policy/tca_policy.json";
+    protected static final String CEF_MESSAGES_JSON_FILE_LOCATION = "data/json/cef/cef_messages.json";
+    protected static final String CEF_MESSAGE_JSON_FILE_LOCATION = "data/json/cef/cef_message.json";
+    protected static final String CEF_MESSAGE_WITH_THRESHOLD_VIOLATION_JSON_FILE_LOCATION =
+            "data/json/cef/cef_message_with_threshold_violation.json";
+
+    protected static final String TCA_CONTROLLER_POLICY_FILE_LOCATION =
+            "data/properties/tca_controller_policy.properties";
+
+
+    /**
+     * Provides TCA Policy that can be used for testing
+     *
+     * @return test TCA Policy Object
+     */
+    protected TCAPolicy getSampleTCAPolicy() {
+        return deserializeJsonFileToModel(TCA_POLICY_JSON_FILE_LOCATION, TCAPolicy.class);
+    }
+
+    /**
+     * Provides list containing 350 CEF messages
+     *
+     * @return CEF Test Message
+     * @throws Exception Exception
+     */
+    protected List<EventListener> getCEFMessages() throws Exception {
+        final String cefMessageAsString = fromStream(CEF_MESSAGES_JSON_FILE_LOCATION);
+        final TypeReference<List<EventListener>> eventListenerListTypeReference =
+                new TypeReference<List<EventListener>>() {
+                };
+        return ANALYTICS_MODEL_OBJECT_MAPPER.readValue(cefMessageAsString, eventListenerListTypeReference);
+    }
+
+    /**
+     * Provides 1 valid CEF messages which does not violate Threshold as String
+     *
+     * @return CEF Test Message String
+     * @throws Exception Exception
+     */
+    protected String getValidCEFMessage() throws Exception {
+        return fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+    }
+
+
+    /**
+     * Provides single CEF Test Message
+     *
+     * @return CEF Test Message
+     * @throws Exception Exception
+     */
+    protected EventListener getCEFEventListener() throws Exception {
+        final String cefMessageAsString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+        return ANALYTICS_MODEL_OBJECT_MAPPER.readValue(cefMessageAsString, EventListener.class);
+    }
+
+    /**
+     * Deserialize given Json file location to given model class and returns it back without any validation check
+     *
+     * @param jsonFileLocation Classpath location of the json file
+     * @param modelClass       Model Class type
+     * @param <T>              Json Model Type
+     * @return  Json model object
+     */
+    public static <T> T deserializeJsonFileToModel(String jsonFileLocation, Class<T> modelClass) {
+        final InputStream jsonFileInputStream =
+                BaseDCAEAnalyticsUnitTest.class.getClassLoader().getResourceAsStream(jsonFileLocation);
+        Assert.assertNotNull("Json File Location must be valid", jsonFileInputStream);
+        try {
+            return ANALYTICS_MODEL_OBJECT_MAPPER.readValue(jsonFileInputStream, modelClass);
+        } catch (IOException ex) {
+            LOG.error("Error while doing assert Json for fileLocation: {}, modelClass: {}, Exception {}",
+                    jsonFileLocation, modelClass, ex);
+            throw new RuntimeException(ex);
+        } finally {
+            try {
+                jsonFileInputStream.close();
+            } catch (IOException e) {
+                LOG.error("Error while closing input stream at file location: {}", jsonFileLocation);
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+    /**
+     * Provides a test application preference for unit testing
+     *
+     * @return tca app preferences
+     */
+    protected static TCATestAppPreferences getTCATestAppPreferences() {
+        final TCATestAppPreferences tcaTestAppPreferences = new TCATestAppPreferences();
+        tcaTestAppPreferences.setSubscriberHostName("SUBSCRIBER_HOST_NAME");
+        tcaTestAppPreferences.setSubscriberHostPortNumber(10000);
+        tcaTestAppPreferences.setSubscriberTopicName("SUBSCRIBER_TOPIC_NAME");
+        tcaTestAppPreferences.setSubscriberUserName("SUBSCRIBER_USERNAME");
+        tcaTestAppPreferences.setSubscriberUserPassword("SUBSCRIBER_PASSWORD");
+        tcaTestAppPreferences.setSubscriberProtocol("https");
+        tcaTestAppPreferences.setSubscriberContentType("application/json");
+        tcaTestAppPreferences.setSubscriberConsumerId("SUBSCRIBER_CONSUMER_ID");
+        tcaTestAppPreferences.setSubscriberConsumerGroup("SUBSCRIBER_CONSUMER_GROUP_NAME");
+        tcaTestAppPreferences.setSubscriberTimeoutMS(10);
+        tcaTestAppPreferences.setSubscriberMessageLimit(100);
+        tcaTestAppPreferences.setSubscriberPollingInterval(1000);
+
+        tcaTestAppPreferences.setPublisherHostName("PUBLISHER_HOST_NAME");
+        tcaTestAppPreferences.setPublisherHostPort(1234);
+        tcaTestAppPreferences.setPublisherTopicName("PUBLISHER_TOPIC_NAME");
+        tcaTestAppPreferences.setPublisherUserName("PUBLISHER_USERNAME");
+        tcaTestAppPreferences.setPublisherUserPassword("PUBLISHER_PASSWORD");
+        tcaTestAppPreferences.setPublisherProtocol("https");
+        tcaTestAppPreferences.setPublisherContentType("application/json");
+        tcaTestAppPreferences.setPublisherMaxBatchSize(100);
+        tcaTestAppPreferences.setPublisherMaxRecoveryQueueSize(100);
+        tcaTestAppPreferences.setPublisherPollingInterval(6000);
+        return tcaTestAppPreferences;
+    }
+
+    protected static Map<String, String> getPreferenceMap() {
+        Map<String, String> preference = new HashMap<>();
+        preference.put("subscriberHostName", "mrlocal-mtnjftle01.homer.com");
+        preference.put("subscriberHostPort", "3905");
+        preference.put("subscriberTopicName", "com.dcae.dmaap.mtnje2.DcaeTestVESPub");
+        preference.put("subscriberProtocol", "https");
+        preference.put("subscriberUserName", "m00502@tca.af.dcae.com");
+        preference.put("subscriberUserPassword", "Te5021abc");
+        preference.put("subscriberContentType", "application/json");
+        preference.put("subscriberConsumerId", "123");
+        preference.put("subscriberConsumerGroup", "testTCAConsumerName-123");
+        preference.put("subscriberTimeoutMS", "-1");
+        preference.put("subscriberMessageLimit", "-1");
+        preference.put("subscriberPollingInterval", "30000");
+
+        preference.put("publisherHostName", "publisherHostName");
+        preference.put("publisherHostPort", "3905");
+        preference.put("publisherTopicName", "publisherTopicName");
+        preference.put("publisherProtocol", "https");
+        preference.put("publisherUserName", "publisherUserName");
+        preference.put("publisherContentType", "application/json");
+        preference.put("publisherMaxBatchSize", "1000");
+        preference.put("publisherMaxRecoveryQueueSize", "100");
+        preference.put("publisherPollingInterval", "6000");
+        return preference;
+    }
+
+    protected static Threshold getCriticalThreshold() {
+        Threshold criticalThreshold = new Threshold();
+        criticalThreshold.setClosedLoopControlName("CL-LBAL-LOW-TRAFFIC-SIG-FB480F95-A453-6F24-B767-FD703241AB1A");
+        criticalThreshold.setThresholdValue(5000L);
+        criticalThreshold.setFieldPath("$.event.measurementsForVfScalingFields.vNicUsageArray[*].packetsIn");
+        criticalThreshold.setDirection(Direction.GREATER_OR_EQUAL);
+        return criticalThreshold;
+    }
+
+    protected static List<Threshold> getThresholds() {
+        Threshold majorThreshold = new Threshold();
+        majorThreshold.setClosedLoopControlName("CL-LBAL-LOW-TRAFFIC-SIG-FB480F95-A453-6F24-B767-FD703241AB1A");
+        majorThreshold.setFieldPath("$.event.measurementsForVfScalingFields.vNicUsageArray[*].packetsIn");
+        majorThreshold.setVersion("Test Version");
+        majorThreshold.setThresholdValue(500L);
+        majorThreshold.setDirection(Direction.LESS_OR_EQUAL);
+
+        Threshold criticalThreshold = new Threshold();
+        criticalThreshold.setClosedLoopControlName("CL-LBAL-LOW-TRAFFIC-SIG-FB480F95-A453-6F24-B767-FD703241AB1A");
+        criticalThreshold.setThresholdValue(5000L);
+        criticalThreshold.setFieldPath("$.event.measurementsForVfScalingFields.vNicUsageArray[*].packetsIn");
+        criticalThreshold.setDirection(Direction.GREATER_OR_EQUAL);
+        return Arrays.asList(majorThreshold, criticalThreshold);
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/it/TCAnalyticsAppConfigIT.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/it/TCAnalyticsAppConfigIT.java
new file mode 100644
index 0000000..2ed5252
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/it/TCAnalyticsAppConfigIT.java
@@ -0,0 +1,53 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.it;
+
+import org.junit.Test;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAIT;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppConfig;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppConfigHolder;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppPreferences;
+
+/**
+ *
+ * @author Rajiv Singla. Creation Date: 10/25/2016.
+ */
+public class TCAnalyticsAppConfigIT extends BaseAnalyticsTCAIT {
+
+
+    @Test
+    public void createTestAppConfigJson() throws Exception {
+        final TCATestAppConfig tcaTestAppConfig = getTCATestAppConfig();
+        final TCATestAppConfigHolder appConfigHolder = new TCATestAppConfigHolder(tcaTestAppConfig);
+        final String appConfigJson = serializeModelToJson(appConfigHolder);
+        LOG.info("AppConfigJson: \n{}", appConfigJson);
+        writeToOutputTextFile("appSettings/tca_app_config.json", appConfigJson, TCAnalyticsAppConfigIT.class);
+    }
+
+    @Test
+    public void createTestAppPreferencesJson() throws Exception {
+        final TCATestAppPreferences tcaTestAppPreferences = getTCATestAppPreferences();
+        final String appPreferencesJson = serializeModelToJson(tcaTestAppPreferences);
+        LOG.info("AppPreferences: \n{}", appPreferencesJson);
+        writeToOutputTextFile("appSettings/tca_app_preferences.json",
+                appPreferencesJson, TCAnalyticsAppConfigIT.class);
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFJsonProcessorTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFJsonProcessorTest.java
new file mode 100644
index 0000000..05efcd6
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFJsonProcessorTest.java
@@ -0,0 +1,117 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.processor;
+
+import org.junit.Test;
+import org.openecomp.dcae.analytics.common.exception.MessageProcessingException;
+import org.openecomp.dcae.analytics.model.domain.cef.EventListener;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+
+/**
+ *
+ * @author Rajiv Singla. Creation Date: 11/9/2016.
+ */
+public class TCACEFJsonProcessorTest extends BaseAnalyticsTCAUnitTest {
+
+
+    // A valid CEF Message
+    @Test
+    public void testCEFJsonProcessorWithValidCEFMessage() throws Exception {
+
+        final String cefMessageString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+        final TCACEFProcessorContext tcacefProcessorContext =
+                new TCACEFProcessorContext(cefMessageString, getSampleTCAPolicy());
+
+        TCACEFJsonProcessor tcacefJsonProcessor = new TCACEFJsonProcessor();
+        final TCACEFProcessorContext finalProcessorContext = tcacefJsonProcessor.apply(tcacefProcessorContext);
+
+        final EventListener cefEventListener = finalProcessorContext.getCEFEventListener();
+
+        assertNotNull("CEF Event Listener must be present", cefEventListener);
+
+    }
+
+    // Even if message is not a valid CEF format but still a Json - Json Processor will parse it
+    @Test
+    public void testCEFJsonProcessorWithValidJson() throws Exception {
+
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(
+                " { \"key\" : \"value\" } ", getSampleTCAPolicy());
+
+        TCACEFJsonProcessor tcacefJsonProcessor = new TCACEFJsonProcessor();
+        final TCACEFProcessorContext finalProcessorContext = tcacefJsonProcessor.apply(tcacefProcessorContext);
+        final EventListener cefEventListener = finalProcessorContext.getCEFEventListener();
+
+        assertNotNull("Even if message is not a valid CEF format but a valid Json.Json Processor must be able to " +
+                        "parse it",
+                cefEventListener);
+    }
+
+    @Test(expected = MessageProcessingException.class)
+    public void testCEFJsonProcessorWithCEFMessageAsNull() throws Exception {
+
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(null, getSampleTCAPolicy());
+
+        TCACEFJsonProcessor tcacefJsonProcessor = new TCACEFJsonProcessor();
+        tcacefJsonProcessor.apply(tcacefProcessorContext);
+
+    }
+
+    @Test
+    public void testCEFJsonProcessorWithCEFMessageIsBlank() throws Exception {
+
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext("   ", getSampleTCAPolicy());
+
+        TCACEFJsonProcessor tcacefJsonProcessor = new TCACEFJsonProcessor();
+        final TCACEFProcessorContext finalProcessorContext = tcacefJsonProcessor.apply(tcacefProcessorContext);
+        assertFalse("Blank message must terminate processing of message chain", finalProcessorContext
+                .canProcessingContinue());
+    }
+
+
+    @Test
+    public void testCEFJsonProcessorWithCEFMessageWhichIsNotValidMessage() throws Exception {
+
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(" Invalid Message ",
+                getSampleTCAPolicy());
+
+        TCACEFJsonProcessor tcacefJsonProcessor = new TCACEFJsonProcessor();
+        final TCACEFProcessorContext finalProcessorContext = tcacefJsonProcessor.apply(tcacefProcessorContext);
+        assertFalse("Invalid message must terminate processing of message chain", finalProcessorContext
+                .canProcessingContinue());
+    }
+
+
+    @Test(expected = MessageProcessingException.class)
+    public void testCEFJsonProcessorWithCEFMessageWhichIsNotValidJson() throws Exception {
+
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(
+                " { \"Invalid Event Listener Json\" } ", getSampleTCAPolicy());
+
+        TCACEFJsonProcessor tcacefJsonProcessor = new TCACEFJsonProcessor();
+        tcacefJsonProcessor.apply(tcacefProcessorContext);
+    }
+
+
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFPolicyThresholdsProcessorTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFPolicyThresholdsProcessorTest.java
new file mode 100644
index 0000000..8d83307
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFPolicyThresholdsProcessorTest.java
@@ -0,0 +1,77 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.processor;
+
+import org.junit.Test;
+import org.openecomp.dcae.analytics.common.service.processor.ProcessingState;
+import org.openecomp.dcae.analytics.model.domain.cef.EventListener;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+/**
+ *
+ * @author Rajiv Singla. Creation Date: 11/9/2016.
+ */
+public class TCACEFPolicyThresholdsProcessorTest extends BaseAnalyticsTCAUnitTest {
+
+    @Test
+    public void testCEFPolicyThresholdProcessorWithNoThresholdViolation() throws Exception {
+
+        final String cefMessageString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(cefMessageString,
+                getSampleTCAPolicy());
+        tcacefProcessorContext.setCEFEventListener(getCEFEventListener());
+
+        AbstractTCAECEFPolicyProcessor policyThresholdsProcessor = new TCACEFPolicyThresholdsProcessor();
+        final TCACEFProcessorContext finalProcessorContext = policyThresholdsProcessor.apply(tcacefProcessorContext);
+
+        assertFalse("Process Context can Processing Continue flag should be false", finalProcessorContext
+                .canProcessingContinue());
+        assertThat("Policy Threshold Processor State must be terminated early",
+                policyThresholdsProcessor.getProcessingState(), is(ProcessingState.PROCESSING_TERMINATED_EARLY));
+
+    }
+
+    @Test
+    public void testCEFPolicyThresholdProcessorWithThresholdViolation() throws Exception {
+
+        final String cefMessageString = fromStream(CEF_MESSAGE_WITH_THRESHOLD_VIOLATION_JSON_FILE_LOCATION);
+        final TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(cefMessageString,
+                getSampleTCAPolicy());
+
+        final EventListener eventListener = getCEFEventListener();
+        tcacefProcessorContext.setCEFEventListener(eventListener);
+
+        AbstractTCAECEFPolicyProcessor policyThresholdsProcessor = new TCACEFPolicyThresholdsProcessor();
+        final TCACEFProcessorContext finalProcessorContext = policyThresholdsProcessor.apply(tcacefProcessorContext);
+
+        assertTrue("Process Context can Processing Continue flag should be true", finalProcessorContext
+                .canProcessingContinue());
+        assertThat("Policy Threshold Processor State must be successful",
+                policyThresholdsProcessor.getProcessingState(), is(ProcessingState.PROCESSING_FINISHED_SUCCESSFULLY));
+
+    }
+
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFProcessorContextTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFProcessorContextTest.java
new file mode 100644
index 0000000..d955d81
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/processor/TCACEFProcessorContextTest.java
@@ -0,0 +1,38 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.processor;
+
+import org.junit.Test;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+
+/**
+ *
+ * @author Rajiv Singla. Creation Date: 11/14/2016.
+ */
+public class TCACEFProcessorContextTest extends BaseAnalyticsTCAUnitTest {
+
+    @Test
+    public void testProcessorContextSerialization() throws Exception {
+        TCACEFProcessorContext tcacefProcessorContext = new TCACEFProcessorContext(getValidCEFMessage(),
+                getSampleTCAPolicy());
+        testSerialization(tcacefProcessorContext, TCACEFProcessorContextTest.class);
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppConfig.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppConfig.java
new file mode 100644
index 0000000..4e057cc
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppConfig.java
@@ -0,0 +1,38 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.settings;
+
+/**
+ * TCA Test App Config is used for testing purposes only
+ *
+ * @author Rajiv Singla. Creation Date: 11/3/2016.
+ */
+public class TCATestAppConfig extends TCAAppConfig {
+
+    public void setAppName(String appName) {
+        this.appName = appName;
+    }
+
+    public void setAppDescription(String appDescription) {
+        this.appDescription = appDescription;
+    }
+
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppConfigHolder.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppConfigHolder.java
new file mode 100644
index 0000000..e854cac
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppConfigHolder.java
@@ -0,0 +1,40 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.settings;
+
+/**
+ * Holder for TCA Test App Config
+ *
+ * @author Rajiv Singla. Creation Date: 11/3/2016.
+ */
+public class TCATestAppConfigHolder {
+
+    private final TCATestAppConfig config;
+
+    public TCATestAppConfigHolder(TCATestAppConfig config) {
+        this.config = config;
+    }
+
+    public TCATestAppConfig getConfig() {
+        return config;
+    }
+
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppPreferences.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppPreferences.java
new file mode 100644
index 0000000..f8a0474
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/settings/TCATestAppPreferences.java
@@ -0,0 +1,136 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.settings;
+
+
+import com.fasterxml.jackson.annotation.JsonAnyGetter;
+
+import java.util.Map;
+
+/**
+ * TCA Test App Preferences are used for testing purposes only
+ *
+ * @author Rajiv Singla. Creation Date: 11/3/2016.
+ */
+public class TCATestAppPreferences extends TCAAppPreferences {
+
+    private Map<String, String> policyPreferences;
+
+    public TCATestAppPreferences() {
+    }
+
+    public TCATestAppPreferences(final Map<String, String> policyPreferences) {
+        this.policyPreferences = policyPreferences;
+    }
+
+    @JsonAnyGetter
+    public Map<String, String> getPolicyPreferences() {
+        return policyPreferences;
+    }
+
+    public void setSubscriberHostName(String subscriberHostName) {
+        this.subscriberHostName = subscriberHostName;
+    }
+
+    public void setSubscriberHostPortNumber(Integer subscriberHostPort) {
+        this.subscriberHostPort = subscriberHostPort;
+    }
+
+    public void setSubscriberTopicName(String subscriberTopicName) {
+        this.subscriberTopicName = subscriberTopicName;
+    }
+
+    public void setSubscriberProtocol(String subscriberProtocol) {
+        this.subscriberProtocol = subscriberProtocol;
+    }
+
+    public void setSubscriberUserName(String subscriberUserName) {
+        this.subscriberUserName = subscriberUserName;
+    }
+
+    public void setSubscriberUserPassword(String subscriberUserPassword) {
+        this.subscriberUserPassword = subscriberUserPassword;
+    }
+
+    public void setSubscriberContentType(String subscriberContentType) {
+        this.subscriberContentType = subscriberContentType;
+    }
+
+    public void setSubscriberConsumerId(String subscriberConsumerId) {
+        this.subscriberConsumerId = subscriberConsumerId;
+    }
+
+    public void setSubscriberConsumerGroup(String subscriberConsumerGroup) {
+        this.subscriberConsumerGroup = subscriberConsumerGroup;
+    }
+
+    public void setSubscriberTimeoutMS(Integer subscriberTimeoutMS) {
+        this.subscriberTimeoutMS = subscriberTimeoutMS;
+    }
+
+    public void setSubscriberMessageLimit(Integer subscriberMessageLimit) {
+        this.subscriberMessageLimit = subscriberMessageLimit;
+    }
+
+    public void setSubscriberPollingInterval(Integer subscriberPollingInterval) {
+        this.subscriberPollingInterval = subscriberPollingInterval;
+    }
+
+    public void setPublisherHostName(String publisherHostName) {
+        this.publisherHostName = publisherHostName;
+    }
+
+    public void setPublisherHostPort(Integer publisherHostPort) {
+        this.publisherHostPort = publisherHostPort;
+    }
+
+    public void setPublisherTopicName(String publisherTopicName) {
+        this.publisherTopicName = publisherTopicName;
+    }
+
+    public void setPublisherProtocol(String publisherProtocol) {
+        this.publisherProtocol = publisherProtocol;
+    }
+
+    public void setPublisherUserName(String publisherUserName) {
+        this.publisherUserName = publisherUserName;
+    }
+
+    public void setPublisherUserPassword(String publisherUserPassword) {
+        this.publisherUserPassword = publisherUserPassword;
+    }
+
+    public void setPublisherContentType(String publisherContentType) {
+        this.publisherContentType = publisherContentType;
+    }
+
+    public void setPublisherMaxBatchSize(Integer publisherMaxBatchSize) {
+        this.publisherMaxBatchSize = publisherMaxBatchSize;
+    }
+
+    public void setPublisherMaxRecoveryQueueSize(Integer publisherMaxRecoveryQueueSize) {
+        this.publisherMaxRecoveryQueueSize = publisherMaxRecoveryQueueSize;
+    }
+
+    public void setPublisherPollingInterval(Integer publisherPollingInterval) {
+        this.publisherPollingInterval = publisherPollingInterval;
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/AppPreferencesToPublisherConfigMapperTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/AppPreferencesToPublisherConfigMapperTest.java
new file mode 100644
index 0000000..d66ef52
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/AppPreferencesToPublisherConfigMapperTest.java
@@ -0,0 +1,58 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.utils;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.openecomp.dcae.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppPreferences;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+/**
+ * @author Manjesh Gowda. Creation Date: 11/21/2016.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class AppPreferencesToPublisherConfigMapperTest extends BaseAnalyticsTCAUnitTest {
+    @Test
+    public void testMapTCAConfigToSubscriberConfigFunctionGood() {
+        DMaaPMRPublisherConfig dMaaPMRPublisherConfig =
+                (new AppPreferencesToPublisherConfigMapper()).apply(getTCATestAppPreferences());
+        assertEquals(dMaaPMRPublisherConfig.getHostName(), "PUBLISHER_HOST_NAME");
+    }
+
+    @Test
+    public void testMapTCAConfigToSubscriberConfigFunctionMap() {
+        DMaaPMRPublisherConfig dMaaPMRPublisherConfig = AppPreferencesToPublisherConfigMapper.map(
+                getTCATestAppPreferences());
+        assertEquals(dMaaPMRPublisherConfig.getHostName(), "PUBLISHER_HOST_NAME");
+    }
+
+    @Test
+    public void testMapTCAConfigToSubscriberConfigFunctionAllNull() {
+        DMaaPMRPublisherConfig dMaaPMRPublisherConfig =
+                (new AppPreferencesToPublisherConfigMapper()).apply(new TCATestAppPreferences());
+        assertNull(dMaaPMRPublisherConfig.getHostName());
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/AppPreferencesToSubscriberConfigMapperTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/AppPreferencesToSubscriberConfigMapperTest.java
new file mode 100644
index 0000000..6963201
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/AppPreferencesToSubscriberConfigMapperTest.java
@@ -0,0 +1,59 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.utils;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.openecomp.dcae.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+import org.openecomp.dcae.analytics.tca.settings.TCATestAppPreferences;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+/**
+ * @author Manjesh Gowda. Creation Date: 11/21/2016.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class AppPreferencesToSubscriberConfigMapperTest extends BaseAnalyticsTCAUnitTest {
+
+    @Test
+    public void testMapTCAConfigToSubscriberConfigFunctionGood() {
+        DMaaPMRSubscriberConfig dMaaPMRSubscriberConfig =
+                (new AppPreferencesToSubscriberConfigMapper()).apply(getTCATestAppPreferences());
+        assertEquals(dMaaPMRSubscriberConfig.getHostName(), "SUBSCRIBER_HOST_NAME");
+    }
+
+    @Test
+    public void testMapTCAConfigToSubscriberConfigFunctionMap() {
+        DMaaPMRSubscriberConfig dMaaPMRSubscriberConfig =
+                AppPreferencesToSubscriberConfigMapper.map(getTCATestAppPreferences());
+        assertEquals(dMaaPMRSubscriberConfig.getHostName(), "SUBSCRIBER_HOST_NAME");
+    }
+
+    @Test
+    public void testMapTCAConfigToSubscriberConfigFunctionAllNull() {
+        DMaaPMRSubscriberConfig dMaaPMRSubscriberConfig =
+                (new AppPreferencesToSubscriberConfigMapper()).apply(new TCATestAppPreferences());
+        assertNull(dMaaPMRSubscriberConfig.getHostName());
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/TCAUtilsTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/TCAUtilsTest.java
new file mode 100644
index 0000000..19c80d2
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/utils/TCAUtilsTest.java
@@ -0,0 +1,357 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.utils;
+
+import co.cask.cdap.api.RuntimeContext;
+import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Table;
+import org.apache.commons.lang3.tuple.Pair;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.openecomp.dcae.analytics.common.exception.MessageProcessingException;
+import org.openecomp.dcae.analytics.model.domain.cef.CommonEventHeader;
+import org.openecomp.dcae.analytics.model.domain.cef.Event;
+import org.openecomp.dcae.analytics.model.domain.cef.EventListener;
+import org.openecomp.dcae.analytics.model.domain.cef.EventSeverity;
+import org.openecomp.dcae.analytics.model.domain.policy.tca.MetricsPerFunctionalRole;
+import org.openecomp.dcae.analytics.model.domain.policy.tca.TCAPolicy;
+import org.openecomp.dcae.analytics.model.domain.policy.tca.Threshold;
+import org.openecomp.dcae.analytics.model.facade.tca.TCAVESResponse;
+import org.openecomp.dcae.analytics.model.util.AnalyticsModelIOUtils;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+import org.openecomp.dcae.analytics.tca.processor.TCACEFProcessorContext;
+import org.openecomp.dcae.analytics.tca.settings.TCAAppPreferences;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.isA;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla. Creation Date: 11/9/2016.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class TCAUtilsTest extends BaseAnalyticsTCAUnitTest {
+
+    @Test
+    public void testGetPolicyFunctionalRoles() throws Exception {
+
+        final TCAPolicy sampleTCAPolicy = getSampleTCAPolicy();
+        final List<String> policyFunctionalRoles = TCAUtils.getPolicyFunctionalRoles(sampleTCAPolicy);
+
+        assertThat("Policy Functional Roles must contain vFirewall and vLoadBalancer", policyFunctionalRoles,
+                containsInAnyOrder("vFirewall", "vLoadBalancer"));
+    }
+
+    @Test
+    public void testGetPolicyFunctionalRoleSupplier() throws Exception {
+        final TCAPolicy sampleTCAPolicy = getSampleTCAPolicy();
+        final Supplier<List<String>> policyFunctionalRoleSupplier = TCAUtils.getPolicyFunctionalRoleSupplier
+                (sampleTCAPolicy);
+        final List<String> policyFunctionalRoles = policyFunctionalRoleSupplier.get();
+        assertThat("Policy Functional Roles must contain vFirewall and vLoadBalancer", policyFunctionalRoles,
+                containsInAnyOrder("vFirewall", "vLoadBalancer"));
+    }
+
+    @Test
+    public void testProcessCEFMessage() throws Exception {
+        final String cefMessageString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+        final TCACEFProcessorContext tcacefProcessorContext = TCAUtils.filterCEFMessage(cefMessageString,
+                getSampleTCAPolicy());
+        assertThat("TCAECEFProcessor Processor Context can continue flag is true", tcacefProcessorContext
+                .canProcessingContinue(), is(true));
+    }
+
+    @Test
+    public void testGetPolicyFRThresholdsTableSupplier() throws Exception {
+        final Table<String, String, List<Threshold>> policyFRThresholdPathTable = TCAUtils
+                .getPolicyFRThresholdsTableSupplier(getSampleTCAPolicy()).get();
+
+        final Map<String, List<Threshold>> vFirewall = policyFRThresholdPathTable.row("vFirewall");
+        final Map<String, List<Threshold>> vLoadBalancer = policyFRThresholdPathTable.row("vLoadBalancer");
+
+        final Set<String> vFirewallThresholdPaths = vFirewall.keySet();
+        final Set<String> vLoadBalancerPaths = vLoadBalancer.keySet();
+
+        assertThat("vFirewall threshold field path size must be " +
+                        "\"$.event.measurementsForVfScalingFields.vNicUsageArray[*].bytesIn\"",
+                vFirewallThresholdPaths.iterator().next(),
+                is("$.event.measurementsForVfScalingFields.vNicUsageArray[*].bytesIn"));
+
+        assertThat("vLoadBalancer threshold field path size must be " +
+                        "\"\"$.event.measurementsForVfScalingFields.vNicUsageArray[*].packetsIn\"",
+                vLoadBalancerPaths.iterator().next(),
+                is("$.event.measurementsForVfScalingFields.vNicUsageArray[*].packetsIn"));
+
+        final List<Threshold> firewallThresholds = policyFRThresholdPathTable.get("vFirewall",
+                "$.event.measurementsForVfScalingFields.vNicUsageArray[*].bytesIn");
+        final List<Threshold> vLoadBalancerThresholds = policyFRThresholdPathTable.get("vLoadBalancer",
+                "$.event.measurementsForVfScalingFields.vNicUsageArray[*].packetsIn");
+
+        assertThat("vFirewall Threshold size must be 2", firewallThresholds.size(), is(2));
+        assertThat("vLoadBalancer Threshold size must be 2", vLoadBalancerThresholds.size(), is(2));
+    }
+
+    @Test
+    public void testGetJsonPathValueWithValidMessageAndPolicy() throws Exception {
+        final String cefMessageString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+        final String jsonPath = "$.event.measurementsForVfScalingFields.vNicUsageArray[*].bytesIn";
+        final ImmutableSet<String> fieldPaths = ImmutableSet.of(jsonPath);
+        final Map<String, List<Long>> jsonPathValueMap = TCAUtils.getJsonPathValue(cefMessageString, fieldPaths);
+        assertThat("Json Path value must match", jsonPathValueMap.get(jsonPath).get(0), is(6086L));
+
+    }
+
+    @Test
+    public void testGetJsonPathValueWithValidPath() throws Exception {
+        final String cefMessageString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
+        final String jsonPath = "$.event.measurementsForVfScalingFields.vNicUsageArray[*].invalid";
+        final ImmutableSet<String> fieldPaths = ImmutableSet.of(jsonPath);
+        final Map<String, List<Long>> jsonPathValueMap = TCAUtils.getJsonPathValue(cefMessageString, fieldPaths);
+        assertThat("Json path value must be empty", jsonPathValueMap.size(), is(0));
+
+    }
+
+    @Test
+    public void testGetValidatedTCAAppPreferences() throws Exception {
+        RuntimeContext runtimeContext = mock(RuntimeContext.class);
+        when(runtimeContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        TCAAppPreferences validatedTCAAppPreferences = TCAUtils.getValidatedTCAAppPreferences(runtimeContext);
+        assertEquals(validatedTCAAppPreferences.getSubscriberHostName(), "mrlocal-mtnjftle01.homer.com");
+    }
+
+    @Test
+    public void testGetValidatedTCAAppPreferencesWhenDMaaPUrlArePresent() throws Exception {
+        RuntimeContext runtimeContext = mock(RuntimeContext.class);
+        final Map<String, String> preferenceMap = getPreferenceMap();
+        preferenceMap.put("dmaap.in.event-input.dmaapUrl",
+                "http://zldcmtd1njcoll00.research.com/unauthenticated.SEC_MEASUREMENT_OUTPUT");
+        preferenceMap.put("dmaap.out.alert-output.dmaapUrl",
+                "http://zldcmtd1njcoll00.research.com/unauthenticated.TCA_EVENT_OUTPUT");
+
+        preferenceMap.put("dmaap.in.event-input.dmaapUserName", null);
+        preferenceMap.put("dmaap.in.event-input.dmaapPassword", null);
+        preferenceMap.put("dmaap.out.alert-output.dmaapUserName", null);
+        preferenceMap.put("dmaap.out.alert-output.dmaapPassword", null);
+
+        when(runtimeContext.getRuntimeArguments()).thenReturn(preferenceMap);
+        TCAAppPreferences validatedTCAAppPreferences = TCAUtils.getValidatedTCAAppPreferences(runtimeContext);
+
+        assertEquals(validatedTCAAppPreferences.getSubscriberProtocol(), "http");
+        assertEquals(validatedTCAAppPreferences.getPublisherProtocol(), "http");
+
+        assertEquals(validatedTCAAppPreferences.getSubscriberHostName(), "zldcmtd1njcoll00.research.com");
+        assertEquals(validatedTCAAppPreferences.getPublisherHostName(), "zldcmtd1njcoll00.research.com");
+
+        assertEquals(validatedTCAAppPreferences.getSubscriberHostPort(), new Integer(3904));
+        assertEquals(validatedTCAAppPreferences.getPublisherHostPort(), new Integer(3904));
+
+        assertEquals(validatedTCAAppPreferences.getSubscriberTopicName(), "unauthenticated.SEC_MEASUREMENT_OUTPUT");
+        assertEquals(validatedTCAAppPreferences.getPublisherTopicName(), "unauthenticated.TCA_EVENT_OUTPUT");
+
+        assertNull(validatedTCAAppPreferences.getSubscriberUserName());
+        assertNull(validatedTCAAppPreferences.getSubscriberUserPassword());
+        assertNull(validatedTCAAppPreferences.getPublisherUserName());
+        assertNull(validatedTCAAppPreferences.getPublisherUserPassword());
+
+    }
+
+    @Test
+    public void testCreateNewTCAVESResponse() throws Exception {
+        TCACEFProcessorContext tcacefProcessorContext = mock(TCACEFProcessorContext.class);
+
+        MetricsPerFunctionalRole metricsPerFunctionalRole = mock(MetricsPerFunctionalRole.class);
+        when(metricsPerFunctionalRole.getThresholds()).thenReturn(getThresholds());
+        when(metricsPerFunctionalRole.getPolicyScope()).thenReturn("Test Policy scope");
+        when(tcacefProcessorContext.getMetricsPerFunctionalRole()).thenReturn(metricsPerFunctionalRole);
+        when(metricsPerFunctionalRole.getFunctionalRole()).thenReturn("vLoadBalancer");
+
+        when(tcacefProcessorContext.getCEFEventListener()).thenReturn(getCEFEventListener());
+        TCAVESResponse tcaVESResponse = TCAUtils.createNewTCAVESResponse(tcacefProcessorContext, "TCA_APP_NAME");
+
+        //TODO :  Add proper assertions, as the usage is not clearly understood
+        assertThat(tcaVESResponse.getClosedLoopControlName(),
+                is("CL-LBAL-LOW-TRAFFIC-SIG-FB480F95-A453-6F24-B767-FD703241AB1A"));
+        assertThat(tcaVESResponse.getVersion(), is("Test Version"));
+        assertThat(tcaVESResponse.getPolicyScope(), is("Test Policy scope"));
+    }
+
+    @Rule
+    public ExpectedException expectedIllegalArgumentException = ExpectedException.none();
+
+    @Test
+    public void testCreateNewTCAVESResponseNullFunctionalRole() throws Exception {
+        expectedIllegalArgumentException.expect(MessageProcessingException.class);
+        expectedIllegalArgumentException.expectCause(isA(IllegalArgumentException.class));
+        expectedIllegalArgumentException.expectMessage("No violations metrics. Unable to create VES Response");
+
+        TCACEFProcessorContext tcacefProcessorContext = mock(TCACEFProcessorContext.class);
+        TCAVESResponse tcaVESResponse = TCAUtils.createNewTCAVESResponse(tcacefProcessorContext, "TCA_APP_NAME");
+        assertNotNull(tcaVESResponse.getClosedLoopControlName());
+    }
+
+    @Test
+    public void testPrioritizeThresholdViolations() throws Exception {
+
+        Map<String, Threshold> thresholdMap = new HashMap<>();
+        Threshold majorThreshold = mock(Threshold.class);
+        when(majorThreshold.getSeverity()).thenReturn(EventSeverity.MAJOR);
+        thresholdMap.put("MAJOR", majorThreshold);
+
+        Threshold result1 = TCAUtils.prioritizeThresholdViolations(thresholdMap);
+        assertEquals(result1.getSeverity(), EventSeverity.MAJOR);
+
+        Threshold criticalThreshold = mock(Threshold.class);
+        when(criticalThreshold.getSeverity()).thenReturn(EventSeverity.CRITICAL);
+        thresholdMap.put("CRITICAL", criticalThreshold);
+
+        Threshold result2 = TCAUtils.prioritizeThresholdViolations(thresholdMap);
+        assertEquals(result2.getSeverity(), EventSeverity.CRITICAL);
+    }
+
+    @Test
+    public void testCreateViolatedMetrics() throws Exception {
+        TCAPolicy tcaPolicy = getSampleTCAPolicy();
+        Threshold violatedThreshold = getCriticalThreshold();
+        String functionalRole = "vFirewall";
+        MetricsPerFunctionalRole result = TCAUtils.createViolatedMetrics(tcaPolicy, violatedThreshold, functionalRole);
+        assertThat(result.getPolicyScope(), is("resource=vFirewall;type=configuration"));
+        assertThat(result.getPolicyName(), is("configuration.dcae.microservice.tca.xml"));
+    }
+
+    @Test
+    public void testCreateViolatedMetricsWrongFunctionalRole() throws Exception {
+        expectedIllegalArgumentException.expect(MessageProcessingException.class);
+        expectedIllegalArgumentException.expectCause(isA(IllegalStateException.class));
+        expectedIllegalArgumentException.expectMessage("TCA Policy must contain functional Role: badFunctionRoleName");
+
+        TCAPolicy tcaPolicy = getSampleTCAPolicy();
+        Threshold violatedThreshold = getCriticalThreshold();
+        String functionalRole = "badFunctionRoleName";
+        MetricsPerFunctionalRole result = TCAUtils.createViolatedMetrics(tcaPolicy, violatedThreshold, functionalRole);
+    }
+
+    @Test
+    public void testGetDomainAndFunctionalRole() {
+        TCACEFProcessorContext tcacefProcessorContext = mock(TCACEFProcessorContext.class);
+        EventListener eventListener = mock(EventListener.class);
+        Event event = mock(Event.class);
+        CommonEventHeader commonEventHeader = mock(CommonEventHeader.class);
+
+        Pair<String, String> result = TCAUtils.getDomainAndFunctionalRole(tcacefProcessorContext);
+        assertNull(result.getLeft());
+        assertNull(result.getRight());
+
+        when(tcacefProcessorContext.getCEFEventListener()).thenReturn(eventListener);
+        result = TCAUtils.getDomainAndFunctionalRole(tcacefProcessorContext);
+        assertNull(result.getLeft());
+        assertNull(result.getRight());
+
+        when(eventListener.getEvent()).thenReturn(event);
+        result = TCAUtils.getDomainAndFunctionalRole(tcacefProcessorContext);
+        assertNull(result.getLeft());
+        assertNull(result.getRight());
+
+        when(event.getCommonEventHeader()).thenReturn(commonEventHeader);
+        result = TCAUtils.getDomainAndFunctionalRole(tcacefProcessorContext);
+        assertNull(result.getLeft());
+        assertNull(result.getRight());
+
+        when(commonEventHeader.getDomain()).thenReturn("testDomain");
+        when(commonEventHeader.getFunctionalRole()).thenReturn("functionalRole");
+
+        result = TCAUtils.getDomainAndFunctionalRole(tcacefProcessorContext);
+        assertEquals(result.getLeft(), "testDomain");
+        assertEquals(result.getRight(), "functionalRole");
+
+    }
+
+    @Test
+    public void testComputeThresholdViolationsNotPresent() throws Exception {
+        TCACEFProcessorContext tcacefProcessorContext = mock(TCACEFProcessorContext.class);
+        when(tcacefProcessorContext.canProcessingContinue()).thenReturn(true);
+        when(tcacefProcessorContext.getMessage()).thenReturn(getValidCEFMessage());
+
+        when(tcacefProcessorContext.getTCAPolicy()).thenReturn(getSampleTCAPolicy());
+        when(tcacefProcessorContext.getCEFEventListener()).thenReturn(getCEFEventListener());
+
+        TCACEFProcessorContext result = TCAUtils.computeThresholdViolations(tcacefProcessorContext);
+        assertNotNull(result);
+        verify(result, times(0)).setMetricsPerFunctionalRole(Mockito.any(MetricsPerFunctionalRole.class));
+    }
+
+    @Test
+    public void testComputeThresholdViolationsPresent() throws Exception {
+        TCACEFProcessorContext tcacefProcessorContext = mock(TCACEFProcessorContext.class);
+        when(tcacefProcessorContext.canProcessingContinue()).thenReturn(true);
+        final String cefMessageString = fromStream(CEF_MESSAGE_WITH_THRESHOLD_VIOLATION_JSON_FILE_LOCATION);
+        when(tcacefProcessorContext.getMessage()).thenReturn(cefMessageString);
+
+        when(tcacefProcessorContext.getTCAPolicy()).thenReturn(getSampleTCAPolicy());
+        when(tcacefProcessorContext.getCEFEventListener()).thenReturn(getCEFEventListener());
+
+        TCACEFProcessorContext result = TCAUtils.computeThresholdViolations(tcacefProcessorContext);
+        verify(result, times(1)).setMetricsPerFunctionalRole(Mockito.any(MetricsPerFunctionalRole.class));
+    }
+
+
+    @Test
+    public void testConvertRuntimeContextToTCAPolicy() throws Exception {
+
+        final Properties controllerProperties =
+                AnalyticsModelIOUtils.loadPropertiesFile(TCA_CONTROLLER_POLICY_FILE_LOCATION);
+
+        Map<String, String> runtimeArgs = new LinkedHashMap<>();
+        for (Map.Entry<Object, Object> property : controllerProperties.entrySet()) {
+            runtimeArgs.put(property.getKey().toString(), property.getValue().toString());
+        }
+
+        RuntimeContext runtimeContext = mock(RuntimeContext.class);
+        when(runtimeContext.getRuntimeArguments()).thenReturn(runtimeArgs);
+        final TCAPolicy tcaPolicy = TCAUtils.getValidatedTCAPolicyPreferences(runtimeContext);
+
+        assertThat("Policy Domain must be measurementsForVfScaling",
+                tcaPolicy.getDomain(), is("measurementsForVfScaling"));
+
+        assertThat("Policy must have 2 metrics per functional roles",
+                tcaPolicy.getMetricsPerFunctionalRole().size(), is(2));
+
+    }
+}
diff --git a/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/worker/TCADMaaPMRSubscriberJobTest.java b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/worker/TCADMaaPMRSubscriberJobTest.java
new file mode 100644
index 0000000..bc8a6f0
--- /dev/null
+++ b/dcae-analytics-tca/src/test/java/org/openecomp/dcae/analytics/tca/worker/TCADMaaPMRSubscriberJobTest.java
@@ -0,0 +1,80 @@
+/*
+ * ============LICENSE_START=========================================================
+ * dcae-analytics
+ * ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ *  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.openecomp.dcae.analytics.tca.worker;
+
+import co.cask.cdap.api.metrics.Metrics;
+import co.cask.cdap.api.worker.WorkerConfigurer;
+import co.cask.cdap.api.worker.WorkerContext;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.openecomp.dcae.analytics.common.AnalyticsConstants;
+import org.openecomp.dcae.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.openecomp.dcae.analytics.tca.BaseAnalyticsTCAUnitTest;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Manjesh Gowda. Creation Date: 11/18/2016.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class TCADMaaPMRSubscriberJobTest extends BaseAnalyticsTCAUnitTest {
+
+    @Test
+    public void testExecute() throws Exception {
+
+        DMaaPMRSubscriber mockDMaaPMRSubscriber = mock(DMaaPMRSubscriber.class);
+        Metrics mockMetrics = mock(Metrics.class);
+
+        WorkerContext workerContext = mock(WorkerContext.class);
+        WorkerConfigurer workerConfigurer = mock(WorkerConfigurer.class);
+        //when(workerContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+
+        JobExecutionContext mockJobExecutionContext = mock(JobExecutionContext.class);
+        JobDataMap mockJobDataMap = mock(JobDataMap.class);
+        when(mockJobExecutionContext.getMergedJobDataMap()).thenReturn(mockJobDataMap);
+
+        /*when(mockJobDataMap.getString(AnalyticsConstants.CDAP_STREAM_VARIABLE_NAME))
+                .thenReturn(CDAPComponentsConstants.TCA_FIXED_SUBSCRIBER_OUTPUT_NAME_STREAM);*/
+
+        when(mockJobDataMap.get(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME))
+                .thenReturn(workerContext);
+
+        when(mockJobDataMap.get(AnalyticsConstants.DMAAP_SUBSCRIBER_VARIABLE_NAME))
+                .thenReturn(mockDMaaPMRSubscriber);
+
+        /*when(mockJobDataMap.get(AnalyticsConstants.DMAAP_SUBSCRIBER_METRICS_VARIABLE_NAME))
+                .thenReturn(mockMetrics);*/
+
+        TCADMaaPMRSubscriberJob tcaDMaaPMRSubscriberJob = new TCADMaaPMRSubscriberJob();
+        tcaDMaaPMRSubscriberJob.execute(mockJobExecutionContext);
+
+        verify(mockJobDataMap, times(1)).getString(Mockito.anyString());
+        verify(mockJobDataMap, times(3)).get(any());
+    }
+}