Initial TCA commit into DCAEGEN2

Change-Id: I5f7f8af2a00419854cafc34b79277df60d1af095
Issue-ID: DCAEGEN2-53
Signed-off-by: an4828 <nekrassov@att.com>

23 files changed, 2143 insertions, 0 deletions

diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/BaseAnalyticsCDAPTCAIT.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/BaseAnalyticsCDAPTCAIT.java
new file mode 100644
index 0000000..df5bb06
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/BaseAnalyticsCDAPTCAIT.java
@@ -0,0 +1,95 @@
+/*
+ * ===============================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.apod.analytics.cdap.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.apod.analytics.cdap.tca.settings.TCATestAppConfig;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelIOUtils;
+import org.openecomp.dcae.apod.analytics.model.util.json.AnalyticsModelObjectMapperSupplier;
+import org.openecomp.dcae.apod.analytics.test.BaseDCAEAnalyticsIT;
+
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeMap;
+
+/**
+ *
+ * @author Rajiv Singla . Creation Date: 10/25/2016.
+ */
+public abstract class BaseAnalyticsCDAPTCAIT 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";
+    
+
+    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.setSubscriberPollingInterval(null);
+        tcaTestAppPreferences.setPublisherMaxBatchSize(null);
+        tcaTestAppPreferences.setPublisherMaxRecoveryQueueSize(null);
+        tcaTestAppPreferences.setEnableAlertCEFFormat(null);
+        tcaTestAppPreferences.setPublisherPollingInterval(null);
+        return tcaTestAppPreferences;
+    }
+
+
+    protected static Map<String, String> getTCAPolicyPreferences() {
+        final Map<String, String> policyPreferences = new TreeMap<>();
+        final Properties policyPreferencesProps =
+                AnalyticsModelIOUtils.loadPropertiesFile(TCA_CONTROLLER_POLICY_FILE_LOCATION, new Properties());
+        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-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/BaseAnalyticsCDAPTCAUnitTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/BaseAnalyticsCDAPTCAUnitTest.java
new file mode 100644
index 0000000..6d84561
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/BaseAnalyticsCDAPTCAUnitTest.java
@@ -0,0 +1,288 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca;
+
+import co.cask.cdap.api.flow.flowlet.AbstractFlowlet;
+import co.cask.cdap.api.flow.flowlet.FlowletContext;
+import co.cask.cdap.internal.flow.DefaultFlowletConfigurer;
+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.apod.analytics.cdap.tca.settings.TCAPolicyPreferences;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppConfig;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
+import org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelIOUtils;
+import org.openecomp.dcae.apod.analytics.model.util.json.AnalyticsModelObjectMapperSupplier;
+import org.openecomp.dcae.apod.analytics.test.BaseDCAEAnalyticsUnitTest;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 10/25/2016.
+ */
+public abstract class BaseAnalyticsCDAPTCAUnitTest 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";
+
+    protected static final String TCA_CONTROLLER_POLICY_FROM_JSON_FILE_LOCATION =
+                        "data/properties/tca_controller_policy_from_json.properties";
+
+
+    protected static final String TCA_TEST_APP_CONFIG_NAME = "testTCAAppName";
+    protected static final String TCA_TEST_APP_CONFIG_DESCRIPTION = "testTCAAppDescription";
+    protected static final String TCA_TEST_APP_CONFIG_SUBSCRIBER_OUTPUT_STREAM_NAME =
+            "testTcaSubscriberOutputStreamName";
+    protected static final String TCA_TEST_APP_CONFIG_VES_ALERT_TABLE_NAME = "testTcaVESAlertsTableName";
+    protected static final String TCA_TEST_APP_CONFIG_VES_MESSAGE_STATUS_TABLE_NAME =
+            "testTcaVESMessageStatusTableName";
+
+
+    /**
+     * 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 TCA Policy that can be used for testing
+     *
+     * @return test {@link TCAPolicyPreferences}
+     */
+    protected TCAPolicyPreferences getSampleTCAPolicyPreferences() {
+        return deserializeJsonFileToModel(TCA_POLICY_JSON_FILE_LOCATION, TCAPolicyPreferences.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);
+            }
+        }
+    }
+
+    protected static TCATestAppConfig getTCATestAppConfig() {
+        final TCATestAppConfig tcaAppConfig = new TCATestAppConfig();
+        tcaAppConfig.setAppName(TCA_TEST_APP_CONFIG_NAME);
+        tcaAppConfig.setAppDescription(TCA_TEST_APP_CONFIG_DESCRIPTION);
+        tcaAppConfig.setTcaSubscriberOutputStreamName(TCA_TEST_APP_CONFIG_SUBSCRIBER_OUTPUT_STREAM_NAME);
+        tcaAppConfig.setTcaVESAlertsTableName(TCA_TEST_APP_CONFIG_VES_ALERT_TABLE_NAME);
+        tcaAppConfig.setTcaVESMessageStatusTableName(TCA_TEST_APP_CONFIG_VES_MESSAGE_STATUS_TABLE_NAME);
+        return tcaAppConfig;
+    }
+
+    /**
+     * 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", "USER");
+        preference.put("subscriberUserPassword", "PASSWORD");
+        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 <T extends AbstractFlowlet> void assertFlowletNameAndDescription(
+            final String expectedName, final String expectedDescription, final T flowlet) {
+        final DefaultFlowletConfigurer defaultFlowletConfigurer =
+                new DefaultFlowletConfigurer(flowlet);
+        flowlet.configure(defaultFlowletConfigurer);
+
+        final String flowletName = getPrivateFiledValue(defaultFlowletConfigurer, "name", String.class);
+        final String flowletDescription =
+                getPrivateFiledValue(defaultFlowletConfigurer, "description", String.class);
+
+        assertThat("Flowlet name must match with CDAPComponentsConstants",
+                flowletName, is(expectedName));
+
+        assertThat("Flowlet description must match with CDAPComponentsConstants",
+                flowletDescription, is(expectedDescription));
+
+    }
+
+    protected static FlowletContext getTestFlowletContextWithValidPolicy() {
+        final Properties controllerProperties =
+                AnalyticsModelIOUtils.loadPropertiesFile(TCA_CONTROLLER_POLICY_FILE_LOCATION, new Properties());
+
+        Map<String, String> runtimeArgs = new LinkedHashMap<>();
+        for (Map.Entry<Object, Object> property : controllerProperties.entrySet()) {
+            runtimeArgs.put(property.getKey().toString(), property.getValue().toString());
+        }
+
+        final FlowletContext flowletContext = mock(FlowletContext.class);
+        when(flowletContext.getRuntimeArguments()).thenReturn(runtimeArgs);
+        return flowletContext;
+    }
+
+    protected static FlowletContext getTestFlowletContextWithValidPolicyFromJSON() {
+        final Properties controllerProperties =
+                AnalyticsModelIOUtils.loadPropertiesFile(TCA_CONTROLLER_POLICY_FROM_JSON_FILE_LOCATION,
+                        new Properties());
+
+        Map<String, String> runtimeArgs = new LinkedHashMap<>();
+        for (Map.Entry<Object, Object> property : controllerProperties.entrySet()) {
+            runtimeArgs.put(property.getKey().toString(), property.getValue().toString());
+        }
+
+        final FlowletContext flowletContext = mock(FlowletContext.class);
+        when(flowletContext.getRuntimeArguments()).thenReturn(runtimeArgs);
+        return flowletContext;
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/TCAAnalyticsApplicationTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/TCAAnalyticsApplicationTest.java
new file mode 100644
index 0000000..564ce91
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/TCAAnalyticsApplicationTest.java
@@ -0,0 +1,47 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca;
+
+import co.cask.cdap.app.DefaultApplicationContext;
+import co.cask.cdap.app.MockAppConfigurer;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppConfig;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+/**
+ * @author Rajiv Singla . Creation Date: 1/12/2017.
+ */
+public class TCAAnalyticsApplicationTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    @Test
+    public void testConfigure() throws Exception {
+        final TCAAnalyticsApplication tcaAnalyticsApplication = new TCAAnalyticsApplication();
+        MockAppConfigurer mockAppConfigurer = new MockAppConfigurer(tcaAnalyticsApplication);
+        final DefaultApplicationContext<TCAAppConfig> applicationContext =
+                new DefaultApplicationContext<TCAAppConfig>(getTCATestAppConfig());
+        tcaAnalyticsApplication.configure(mockAppConfigurer, applicationContext);
+        assertThat(TCA_TEST_APP_CONFIG_NAME, is(mockAppConfigurer.getName()));
+    }
+
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flow/TCAVESCollectorFlowTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flow/TCAVESCollectorFlowTest.java
new file mode 100644
index 0000000..05444d7
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flow/TCAVESCollectorFlowTest.java
@@ -0,0 +1,76 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.flow;
+
+import co.cask.cdap.AllProgramsApp;
+import co.cask.cdap.api.flow.FlowletConnection;
+import co.cask.cdap.api.flow.FlowletDefinition;
+import co.cask.cdap.internal.app.runtime.flow.DefaultFlowConfigurer;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertThat;
+
+/**
+ * @author Rajiv Singla . Creation Date: 1/12/2017.
+ */
+public class TCAVESCollectorFlowTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    @Test
+    public void testConfigure() throws Exception {
+
+        final TCAVESCollectorFlow tcavesCollectorFlow = new TCAVESCollectorFlow(getTCATestAppConfig());
+        final DefaultFlowConfigurer configurer = new DefaultFlowConfigurer(new AllProgramsApp.NoOpFlow());
+        tcavesCollectorFlow.configure(configurer);
+        final String flowName = getPrivateFiledValue(configurer, "name", String.class);
+        final String flowDescription = getPrivateFiledValue(configurer, "description", String.class);
+
+        assertThat("TCAVESCollectorFlow Name must match with what is defined in CDAPComponents Constants",
+                flowName, is(CDAPComponentsConstants.TCA_FIXED_VES_COLLECTOR_NAME_FLOW));
+
+        assertThat("TCAVESCollectorFlow Description must match with what is defined in CDAPComponents Constants",
+                flowDescription, is(CDAPComponentsConstants.TCA_FIXED_VES_COLLECTOR_DESCRIPTION_FLOW));
+
+        final Map<String, FlowletDefinition> flowlets =
+                (Map<String, FlowletDefinition>) getPrivateFiledValue(configurer, "flowlets", HashMap.class);
+
+        assertThat("TCAVESCollector must contain all TCA VES flowlets", flowlets.keySet(),
+                containsInAnyOrder(CDAPComponentsConstants.TCA_FIXED_VES_MESSAGE_ROUTER_NAME_FLOWLET,
+                        CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_NAME_FLOWLET,
+                        CDAPComponentsConstants.TCA_FIXED_VES_ALERTS_SINK_NAME_FLOWLET));
+
+        final List<FlowletConnection> connections =
+                (List<FlowletConnection>) getPrivateFiledValue(configurer, "connections", ArrayList.class);
+
+        assertThat("There must be three connections in VES Collector Flow", connections.size(), is(3));
+
+    }
+
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESAlertsSinkFlowletTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESAlertsSinkFlowletTest.java
new file mode 100644
index 0000000..57e0c2a
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESAlertsSinkFlowletTest.java
@@ -0,0 +1,78 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.flowlet;
+
+import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
+import co.cask.cdap.api.flow.flowlet.FlowletContext;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertEntity;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/16/2016.
+ */
+public class TCAVESAlertsSinkFlowletTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+
+    @Test
+    public void testConfigure() throws Exception {
+        final TCAVESAlertsSinkFlowlet tcavesAlertsSinkFlowlet =
+                new TCAVESAlertsSinkFlowlet("testTCAVESAlertTableName");
+        assertFlowletNameAndDescription(CDAPComponentsConstants.TCA_FIXED_VES_ALERTS_SINK_NAME_FLOWLET,
+                CDAPComponentsConstants.TCA_FIXED_VES_ALERTS_SINK_DESCRIPTION_FLOWLET, tcavesAlertsSinkFlowlet);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void saveAlerts() throws Exception {
+
+        final String testAlertTableName = "testTCAVESAlertTableName";
+
+        final TCAVESAlertsSinkFlowlet tcavesAlertsSinkFlowlet = new TCAVESAlertsSinkFlowlet(testAlertTableName);
+
+        final FlowletContext mockFlowletContext = Mockito.mock(FlowletContext.class);
+        final ObjectMappedTable mockObjectMappedTable = Mockito.mock(ObjectMappedTable.class);
+        when(mockFlowletContext.getDataset(eq(testAlertTableName))).thenReturn(mockObjectMappedTable);
+        tcavesAlertsSinkFlowlet.initialize(mockFlowletContext);
+        final ObjectMappedTable tcaVESAlertsTableName =
+                getPrivateFiledValue(tcavesAlertsSinkFlowlet, "tcaVESAlertsTable", ObjectMappedTable.class);
+        assertTrue(tcaVESAlertsTableName == mockObjectMappedTable);
+
+        doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAVESAlertEntity.class));
+        final String testAlertMessage = "testMessage";
+        tcavesAlertsSinkFlowlet.saveAlerts(testAlertMessage);
+
+        verify(mockObjectMappedTable,
+                times(1)).write(any(String.class), any(TCAVESAlertEntity.class));
+
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESMessageRouterFlowletTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESMessageRouterFlowletTest.java
new file mode 100644
index 0000000..6961405
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESMessageRouterFlowletTest.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.apod.analytics.cdap.tca.flowlet;
+
+import co.cask.cdap.api.flow.flowlet.OutputEmitter;
+import co.cask.cdap.api.flow.flowlet.StreamEvent;
+import com.google.common.base.Charsets;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
+
+import java.nio.ByteBuffer;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/19/2016.
+ */
+public class TCAVESMessageRouterFlowletTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private static final String TEST_MESSAGE = "test message";
+    private final OutputEmitter mockOutputEmitter = Mockito.mock(OutputEmitter.class);
+
+    private class TCATestVESMessageRouterFlowlet extends TCAVESMessageRouterFlowlet {
+
+        @SuppressWarnings("unchecked")
+        public TCATestVESMessageRouterFlowlet() {
+            this.vesMessageEmitter = mockOutputEmitter;
+            doNothing().when(mockOutputEmitter).emit(eq(TEST_MESSAGE),
+                    eq(AnalyticsConstants.TCA_VES_MESSAGE_ROUTER_PARTITION_KEY),
+                    eq(TEST_MESSAGE.hashCode()));
+        }
+    }
+
+    @Test
+    public void testConfigure() throws Exception {
+        final TCAVESMessageRouterFlowlet tcavesMessageRouterFlowlet = new TCAVESMessageRouterFlowlet();
+        assertFlowletNameAndDescription(CDAPComponentsConstants.TCA_FIXED_VES_MESSAGE_ROUTER_NAME_FLOWLET,
+                CDAPComponentsConstants.TCA_FIXED_VES_MESSAGE_ROUTER_DESCRIPTION_FLOWLET, tcavesMessageRouterFlowlet);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void routeVESMessage() throws Exception {
+        final TCATestVESMessageRouterFlowlet tcavesMessageRouterFlowlet = new TCATestVESMessageRouterFlowlet();
+        final StreamEvent mockStreamEvent = Mockito.mock(StreamEvent.class);
+        final ByteBuffer testMessage = Charsets.UTF_8.encode(TEST_MESSAGE);
+        when(mockStreamEvent.getBody()).thenReturn(testMessage);
+        tcavesMessageRouterFlowlet.routeVESMessage(mockStreamEvent);
+        verify(mockOutputEmitter,
+                times(1)).emit(eq(TEST_MESSAGE),
+                eq(AnalyticsConstants.TCA_VES_MESSAGE_ROUTER_PARTITION_KEY),
+                eq(TEST_MESSAGE.hashCode()));
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESThresholdViolationCalculatorFlowletTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESThresholdViolationCalculatorFlowletTest.java
new file mode 100644
index 0000000..b70234a
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESThresholdViolationCalculatorFlowletTest.java
@@ -0,0 +1,161 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.flowlet;
+
+import co.cask.cdap.api.app.ApplicationSpecification;
+import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
+import co.cask.cdap.api.flow.flowlet.FlowletContext;
+import co.cask.cdap.api.flow.flowlet.OutputEmitter;
+import co.cask.cdap.api.metrics.Metrics;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAMessageStatusEntity;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;
+import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/19/2016.
+ */
+@SuppressWarnings("unchecked")
+public class TCAVESThresholdViolationCalculatorFlowletTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private static final String messageStatusTableName = "TEST_MESSAGE_STATUS_TABLE";
+
+    private TCAVESThresholdViolationCalculatorFlowlet violationCalculatorFlowlet;
+    private Metrics metrics;
+    private OutputEmitter outputEmitter;
+    private ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable;
+
+    private static class TCATestVESThresholdViolationCalculatorFlowlet extends
+            TCAVESThresholdViolationCalculatorFlowlet {
+        public TCATestVESThresholdViolationCalculatorFlowlet(
+                final String messageStatusTableName,
+                final OutputEmitter tcaAlertOutputEmitter,
+                ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable,
+                Metrics metrics) {
+            super(messageStatusTableName);
+            this.tcaAlertOutputEmitter = tcaAlertOutputEmitter;
+            this.metrics = metrics;
+        }
+    }
+
+    @Before
+    public void before() {
+        violationCalculatorFlowlet = new TCAVESThresholdViolationCalculatorFlowlet(messageStatusTableName);
+        vesMessageStatusTable = Mockito.mock(ObjectMappedTable.class);
+        outputEmitter = Mockito.mock(OutputEmitter.class);
+        metrics = Mockito.mock(Metrics.class);
+    }
+
+    @Test
+    public void testConfigure() throws Exception {
+        assertFlowletNameAndDescription(
+                CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_NAME_FLOWLET,
+                CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_DESCRIPTION_FLOWLET,
+                violationCalculatorFlowlet);
+    }
+
+    @Test
+    public void testInitialize() throws Exception {
+        final FlowletContext mockFlowletContext = initializeFlowlet(violationCalculatorFlowlet, vesMessageStatusTable);
+        verify(mockFlowletContext, times(1)).getDataset(anyString());
+    }
+
+    @Test
+    public void testFilterVESMessagesWhenVESMessageIsInApplicable() throws Exception {
+        final TCATestVESThresholdViolationCalculatorFlowlet thresholdViolationCalculatorFlowlet =
+                createTestViolationCalculator(vesMessageStatusTable, outputEmitter, metrics);
+        initializeFlowlet(thresholdViolationCalculatorFlowlet, vesMessageStatusTable);
+        thresholdViolationCalculatorFlowlet.filterVESMessages("inapplicable");
+        verify(vesMessageStatusTable, times(1)).write(anyString(),
+                any(TCAMessageStatusEntity.class));
+    }
+
+    @Test
+    public void testFilterVESMessagesWhenVESMessageIsCompliant() throws Exception {
+        final TCATestVESThresholdViolationCalculatorFlowlet thresholdViolationCalculatorFlowlet =
+                createTestViolationCalculator(vesMessageStatusTable, outputEmitter, metrics);
+        initializeFlowlet(thresholdViolationCalculatorFlowlet, vesMessageStatusTable);
+        thresholdViolationCalculatorFlowlet.filterVESMessages(getValidCEFMessage());
+        verify(vesMessageStatusTable, times(1)).write(anyString(),
+                any(TCAMessageStatusEntity.class));
+    }
+
+    @Test
+    public void testFilterVESMessagesWhenVESMessageNonCompliant() throws Exception {
+        final TCATestVESThresholdViolationCalculatorFlowlet thresholdViolationCalculatorFlowlet =
+                createTestViolationCalculator(vesMessageStatusTable, outputEmitter, metrics);
+        final FlowletContext flowletContext =
+                initializeFlowlet(thresholdViolationCalculatorFlowlet, vesMessageStatusTable);
+        final TCAPolicy policy = CDAPTCAUtils.getValidatedTCAPolicyPreferences(flowletContext);
+        final Threshold threshold = policy.getMetricsPerFunctionalRole().get(0).getThresholds().get(0);
+        final Long thresholdValue = threshold.getThresholdValue();
+        final EventListener thresholdViolatingMessage = getCEFEventListener();
+        thresholdViolatingMessage.getEvent().getMeasurementsForVfScalingFields().getVNicUsageArray().get(0).setBytesIn
+                (thresholdValue - 1);
+        thresholdViolationCalculatorFlowlet.filterVESMessages(
+                ANALYTICS_MODEL_OBJECT_MAPPER.writeValueAsString(thresholdViolatingMessage));
+        verify(vesMessageStatusTable, times(1)).write(anyString(),
+                any(TCAMessageStatusEntity.class));
+        verify(outputEmitter, times(1)).emit(anyString());
+    }
+
+    private static TCATestVESThresholdViolationCalculatorFlowlet createTestViolationCalculator(
+            final ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable,
+            final OutputEmitter outputEmitter, final Metrics metrics) {
+        doNothing().when(outputEmitter).emit(anyString());
+        doNothing().when(metrics).count(anyString(), anyInt());
+        doNothing().when(vesMessageStatusTable).write(anyString(), any(TCAMessageStatusEntity.class));
+        return new TCATestVESThresholdViolationCalculatorFlowlet(messageStatusTableName, outputEmitter,
+                vesMessageStatusTable, metrics);
+    }
+
+    private static <T extends TCAVESThresholdViolationCalculatorFlowlet> FlowletContext initializeFlowlet(
+            T calculatorFlowlet, ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable) {
+        final FlowletContext mockFlowletContext = getTestFlowletContextWithValidPolicy();
+        when(mockFlowletContext.getDataset(anyString())).thenReturn(vesMessageStatusTable);
+        when(mockFlowletContext.getInstanceId()).thenReturn(1);
+        ApplicationSpecification mockApplicationSpecification = Mockito.mock(ApplicationSpecification.class);
+        when(mockFlowletContext.getApplicationSpecification()).thenReturn(mockApplicationSpecification);
+        when(mockApplicationSpecification.getName()).thenReturn("TestTCAAppName");
+        try {
+            calculatorFlowlet.initialize(mockFlowletContext);
+            return mockFlowletContext;
+        } catch (Exception e) {
+            LOG.error("error while flowlet initialization");
+            throw new RuntimeException(e);
+        }
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/it/TCAnalyticsAppConfigIT.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/it/TCAnalyticsAppConfigIT.java
new file mode 100644
index 0000000..aa1d413
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/it/TCAnalyticsAppConfigIT.java
@@ -0,0 +1,52 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.it;
+
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAIT;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppConfig;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppConfigHolder;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppPreferences;
+
+/**
+ * @author Rajiv Singla . Creation Date: 10/25/2016.
+ */
+public class TCAnalyticsAppConfigIT extends BaseAnalyticsCDAPTCAIT {
+
+
+    @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-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppConfig.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppConfig.java
new file mode 100644
index 0000000..4376df0
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppConfig.java
@@ -0,0 +1,62 @@
+/*
+ * ===============================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.apod.analytics.cdap.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;
+    }
+
+    public void setTcaSubscriberOutputStreamName(String tcaSubscriberOutputStreamName) {
+        this.tcaSubscriberOutputStreamName = tcaSubscriberOutputStreamName;
+    }
+
+    public void setThresholdCalculatorFlowletInstances(Integer thresholdCalculatorFlowletInstances) {
+        this.thresholdCalculatorFlowletInstances = thresholdCalculatorFlowletInstances;
+    }
+
+    public void setTcaVESMessageStatusTableName(String tcaVESMessageStatusTableName) {
+        this.tcaVESMessageStatusTableName = tcaVESMessageStatusTableName;
+    }
+
+    public void setTcaVESMessageStatusTableTTLSeconds(Integer tcaVESMessageStatusTableTTLSeconds) {
+        this.tcaVESMessageStatusTableTTLSeconds = tcaVESMessageStatusTableTTLSeconds;
+    }
+
+    public void setTcaVESAlertsTableName(String tcaVESAlertsTableName) {
+        this.tcaVESAlertsTableName = tcaVESAlertsTableName;
+    }
+
+    public void setTcaVESAlertsTableTTLSeconds(Integer tcaVESAlertsTableTTLSeconds) {
+        this.tcaVESAlertsTableTTLSeconds = tcaVESAlertsTableTTLSeconds;
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppConfigHolder.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppConfigHolder.java
new file mode 100644
index 0000000..88d0207
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/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.apod.analytics.cdap.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-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppPreferences.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppPreferences.java
new file mode 100644
index 0000000..6dd2d38
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppPreferences.java
@@ -0,0 +1,140 @@
+/*
+ * ===============================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.apod.analytics.cdap.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;
+    }
+
+    public void setEnableAlertCEFFormat(Boolean enableAlertCEFFormat) {
+        this.enableAlertCEFFormat = enableAlertCEFFormat;
+    }
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppPreferencesTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppPreferencesTest.java
new file mode 100644
index 0000000..4272c61
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/settings/TCATestAppPreferencesTest.java
@@ -0,0 +1,41 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.settings;
+
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/16/2016.
+ */
+public class TCATestAppPreferencesTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    @Test
+    public void testToString() throws Exception {
+        final TCATestAppPreferences tcaTestAppPreferences = getTCATestAppPreferences();
+        assertTrue(tcaTestAppPreferences.toString().contains("subscriberHostName"));
+        assertTrue(tcaTestAppPreferences.toString().contains("subscriberHostPort"));
+        assertTrue(tcaTestAppPreferences.toString().contains("publisherHostName"));
+        assertTrue(tcaTestAppPreferences.toString().contains("publisherHostPort"));
+    }
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/AppPreferencesToPublisherConfigMapperTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/AppPreferencesToPublisherConfigMapperTest.java
new file mode 100644
index 0000000..79cca96
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/AppPreferencesToPublisherConfigMapperTest.java
@@ -0,0 +1,62 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.utils;
+
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Manjesh Gowda. Creation Date: 11/21/2016.
+ */
+public class AppPreferencesToPublisherConfigMapperTest extends BaseAnalyticsCDAPTCAUnitTest {
+    @Test
+    public void testMapTCAConfigToPublisherConfigFunctionGood() {
+        DMaaPMRPublisherConfig dMaaPMRPublisherConfig =
+                (new AppPreferencesToPublisherConfigMapper()).apply(getTCATestAppPreferences());
+        assertEquals(dMaaPMRPublisherConfig.getHostName(), "PUBLISHER_HOST_NAME");
+    }
+
+    @Test
+    public void testMapTCAConfigToPublisherConfigFunctionMap() {
+        DMaaPMRPublisherConfig dMaaPMRPublisherConfig = AppPreferencesToPublisherConfigMapper.map(
+                getTCATestAppPreferences());
+        assertEquals(dMaaPMRPublisherConfig.getHostName(), "PUBLISHER_HOST_NAME");
+    }
+
+    @Test
+    public void testMapTCAConfigToPublisherConfigFunction() {
+        final TCATestAppPreferences tcaAppPreferences = new TCATestAppPreferences();
+        final String publisherHostName = "publisherHostName";
+        final String publisherTopicName = "publisherTopicName";
+        tcaAppPreferences.setPublisherHostName(publisherHostName);
+        tcaAppPreferences.setPublisherTopicName(publisherTopicName);
+        DMaaPMRPublisherConfig dMaaPMRPublisherConfig =
+                (new AppPreferencesToPublisherConfigMapper()).apply(tcaAppPreferences);
+        assertTrue(publisherHostName.equals(dMaaPMRPublisherConfig.getHostName()));
+        assertTrue(publisherTopicName.equals(dMaaPMRPublisherConfig.getTopicName()));
+
+    }
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/AppPreferencesToSubscriberConfigMapperTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/AppPreferencesToSubscriberConfigMapperTest.java
new file mode 100644
index 0000000..b600caf
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/AppPreferencesToSubscriberConfigMapperTest.java
@@ -0,0 +1,62 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.utils;
+
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Manjesh Gowda. Creation Date: 11/21/2016.
+ */
+public class AppPreferencesToSubscriberConfigMapperTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    @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 testMapTCAConfigToSubscriberConfigFunction() {
+        final TCATestAppPreferences tcaAppPreferences = new TCATestAppPreferences();
+        final String subscriberHostname = "subscriberHostname";
+        tcaAppPreferences.setSubscriberHostName(subscriberHostname);
+        final String subscriberTopicName = "subscriberTopicName";
+        tcaAppPreferences.setSubscriberTopicName(subscriberTopicName);
+        DMaaPMRSubscriberConfig dMaaPMRSubscriberConfig =
+                (new AppPreferencesToSubscriberConfigMapper()).apply(tcaAppPreferences);
+        assertTrue(subscriberHostname.equals(dMaaPMRSubscriberConfig.getHostName()));
+        assertTrue(subscriberTopicName.equals(dMaaPMRSubscriberConfig.getTopicName()));
+    }
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/CDAPTCAUtilsTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/CDAPTCAUtilsTest.java
new file mode 100644
index 0000000..75f336d
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/utils/CDAPTCAUtilsTest.java
@@ -0,0 +1,70 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.utils;
+
+import co.cask.cdap.api.RuntimeContext;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 11/9/2016.
+ */
+public class CDAPTCAUtilsTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    @Test
+    public void testGetValidatedTCAAppPreferences() throws Exception {
+        RuntimeContext runtimeContext = mock(RuntimeContext.class);
+        when(runtimeContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        TCAAppPreferences validatedTCAAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(runtimeContext);
+        assertEquals(validatedTCAAppPreferences.getSubscriberHostName(), "mrlocal-mtnjftle01.homer.com");
+    }
+
+    @Test
+    public void testConvertRuntimeContextToTCAPolicy() throws Exception {
+
+        final TCAPolicy tcaPolicy =
+                CDAPTCAUtils.getValidatedTCAPolicyPreferences(getTestFlowletContextWithValidPolicy());
+        assertThat("Policy Domain must be measurementsForVfScaling",
+                tcaPolicy.getDomain(), is("measurementsForVfScaling"));
+        assertThat("Policy must have 2 metrics per functional roles",
+                tcaPolicy.getMetricsPerFunctionalRole().size(), is(2));
+    }
+
+    @Test
+    public void testConvertRuntimeContextToTCAPolicyFromJSON() throws Exception {
+
+        final TCAPolicy tcaPolicy =
+                CDAPTCAUtils.getValidatedTCAPolicyPreferences(getTestFlowletContextWithValidPolicyFromJSON());
+        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-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAAppConfigValidatorTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAAppConfigValidatorTest.java
new file mode 100644
index 0000000..fb82d33
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAAppConfigValidatorTest.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.apod.analytics.cdap.tca.validator;
+
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppConfig;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppConfig;
+import org.openecomp.dcae.apod.analytics.common.validation.GenericValidationResponse;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/16/2016.
+ */
+public class TCAAppConfigValidatorTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+
+    @Test
+    public void validateAppSettingsWhenAppConfigIsValid() throws Exception {
+        final TCAAppConfigValidator tcaAppConfigValidator = new TCAAppConfigValidator();
+        final TCATestAppConfig tcaTestAppConfig = getTCATestAppConfig();
+        final GenericValidationResponse<TCAAppConfig> validationResponse =
+                tcaAppConfigValidator.validateAppSettings(tcaTestAppConfig);
+        assertFalse(validationResponse.hasErrors());
+    }
+
+    @Test
+    public void testWhenSubscriberOutputStreamIsNull() throws Exception {
+        final TCAAppConfigValidator tcaAppConfigValidator = new TCAAppConfigValidator();
+        final TCATestAppConfig tcaTestAppConfig = getTCATestAppConfig();
+        tcaTestAppConfig.setTcaSubscriberOutputStreamName(null);
+        final GenericValidationResponse<TCAAppConfig> validationResponse =
+                tcaAppConfigValidator.validateAppSettings(tcaTestAppConfig);
+        assertTrue(validationResponse.hasErrors());
+    }
+
+    @Test
+    public void testWhenVESMessageStatusTableNameIsNull() throws Exception {
+        final TCAAppConfigValidator tcaAppConfigValidator = new TCAAppConfigValidator();
+        final TCATestAppConfig tcaTestAppConfig = getTCATestAppConfig();
+        tcaTestAppConfig.setTcaVESMessageStatusTableName(null);
+        final GenericValidationResponse<TCAAppConfig> validationResponse =
+                tcaAppConfigValidator.validateAppSettings(tcaTestAppConfig);
+        assertTrue(validationResponse.hasErrors());
+    }
+
+    @Test
+    public void testWhenVESAlertsTableNameIsNull() throws Exception {
+        final TCAAppConfigValidator tcaAppConfigValidator = new TCAAppConfigValidator();
+        final TCATestAppConfig tcaTestAppConfig = getTCATestAppConfig();
+        tcaTestAppConfig.setTcaVESAlertsTableName(null);
+        final GenericValidationResponse<TCAAppConfig> validationResponse =
+                tcaAppConfigValidator.validateAppSettings(tcaTestAppConfig);
+        assertTrue(validationResponse.hasErrors());
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAPolicyPreferencesValidatorTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAPolicyPreferencesValidatorTest.java
new file mode 100644
index 0000000..bfe288f
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAPolicyPreferencesValidatorTest.java
@@ -0,0 +1,85 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.validator;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAPolicyPreferences;
+import org.openecomp.dcae.apod.analytics.common.validation.GenericValidationResponse;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.MetricsPerFunctionalRole;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/16/2016.
+ */
+public class TCAPolicyPreferencesValidatorTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private TCAPolicyPreferencesValidator tcaPolicyPreferencesValidator;
+    private TCAPolicyPreferences tcaPolicyPreferences;
+
+    @Before
+    public void before() {
+        tcaPolicyPreferencesValidator = new TCAPolicyPreferencesValidator();
+        tcaPolicyPreferences = getSampleTCAPolicyPreferences();
+    }
+
+    @Test
+    public void testValidateAppSettingsWhenSettingsAreValid() throws Exception {
+        final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
+                tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
+        assertFalse(validationResponse.hasErrors());
+    }
+
+    @Test
+    public void testValidateAppSettingsWhenDomainIsNullAndFunctionRoleIsEmpty() throws Exception {
+        tcaPolicyPreferences.setDomain(null);
+        tcaPolicyPreferences.setMetricsPerFunctionalRole(Collections.<MetricsPerFunctionalRole>emptyList());
+        final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
+                tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
+        assertTrue(validationResponse.hasErrors());
+        assertTrue(validationResponse.getErrorMessages().size() == 2);
+    }
+
+    @Test
+    public void testValidateAppSettingsWhenThresholdIsEmpty() throws Exception {
+        tcaPolicyPreferences.getMetricsPerFunctionalRole().get(0).setThresholds(Collections.<Threshold>emptyList());
+        final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
+                tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
+        assertTrue(validationResponse.hasErrors());
+        assertTrue(validationResponse.getErrorMessages().size() == 1);
+    }
+
+    @Test
+    public void testValidateAppSettingsWhenThresholdPathIsMissing() throws Exception {
+        tcaPolicyPreferences.getMetricsPerFunctionalRole().get(0).getThresholds().get(0).setFieldPath(null);
+        final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
+                tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
+        assertTrue(validationResponse.hasErrors());
+        assertTrue(validationResponse.getErrorMessages().size() == 1);
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAPreferencesValidatorTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAPreferencesValidatorTest.java
new file mode 100644
index 0000000..1cc30b9
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/validator/TCAPreferencesValidatorTest.java
@@ -0,0 +1,74 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.validator;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;
+import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCATestAppPreferences;
+import org.openecomp.dcae.apod.analytics.common.validation.GenericValidationResponse;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/19/2016.
+ */
+public class TCAPreferencesValidatorTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private TCAPreferencesValidator tcaPreferencesValidator;
+    private TCATestAppPreferences tcaTestAppPreferences;
+
+    @Before
+    public void before() {
+        tcaPreferencesValidator = new TCAPreferencesValidator();
+        tcaTestAppPreferences = getTCATestAppPreferences();
+    }
+
+    @Test
+    public void validateAppSettingsWithValidParameters() throws Exception {
+        final GenericValidationResponse<TCAAppPreferences> validationResponse =
+                tcaPreferencesValidator.validateAppSettings(tcaTestAppPreferences);
+        assertFalse(validationResponse.hasErrors());
+    }
+
+    @Test
+    public void validateAppSettingsWhenSubscriberHostOrTopicNameIsNotPresent() throws Exception {
+        tcaTestAppPreferences.setSubscriberHostName(null);
+        tcaTestAppPreferences.setSubscriberTopicName(null);
+        final GenericValidationResponse<TCAAppPreferences> validationResponse =
+                tcaPreferencesValidator.validateAppSettings(tcaTestAppPreferences);
+        assertTrue(validationResponse.hasErrors());
+        assertTrue(validationResponse.getErrorMessages().size() == 2);
+    }
+
+    @Test
+    public void validateAppSettingsWhenPublisherHostOrTopicNameIsNotPresent() throws Exception {
+        tcaTestAppPreferences.setPublisherHostName(null);
+        tcaTestAppPreferences.setPublisherTopicName(null);
+        final GenericValidationResponse<TCAAppPreferences> validationResponse =
+                tcaPreferencesValidator.validateAppSettings(tcaTestAppPreferences);
+        assertTrue(validationResponse.hasErrors());
+        assertTrue(validationResponse.getErrorMessages().size() == 2);
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/BaseTCADMaaPMRWorkerTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/BaseTCADMaaPMRWorkerTest.java
new file mode 100644
index 0000000..fa48ea8
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/BaseTCADMaaPMRWorkerTest.java
@@ -0,0 +1,99 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.worker;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
+import org.quartz.Scheduler;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.mockito.Mockito.doNothing;
+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: 12/19/2016.
+ */
+public class BaseTCADMaaPMRWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private static final String SCHEDULER_NAME = "testSchedulerName";
+
+    private Scheduler mockScheduler;
+    private AtomicBoolean mockIsSchedulerShutdown;
+    private BaseTestTCADMaaPMRWorker baseTestTCADMaaPMRWorker;
+
+    private static class BaseTestTCADMaaPMRWorker extends BaseTCADMaaPMRWorker {
+
+        public BaseTestTCADMaaPMRWorker(final Scheduler scheduler, final AtomicBoolean isSchedulerShutdown) {
+            this.scheduler = scheduler;
+            this.isSchedulerShutdown = isSchedulerShutdown;
+        }
+
+    }
+
+    @Before
+    public void before() throws Exception {
+        mockScheduler = mock(Scheduler.class);
+        mockIsSchedulerShutdown = mock(AtomicBoolean.class);
+        baseTestTCADMaaPMRWorker = new BaseTestTCADMaaPMRWorker(mockScheduler, mockIsSchedulerShutdown);
+        when(mockScheduler.getSchedulerName()).thenReturn(SCHEDULER_NAME);
+        doNothing().when(mockScheduler).start();
+        doNothing().when(mockScheduler).shutdown();
+    }
+
+    @Test
+    public void testRun() throws Exception {
+        createShutdownHookThread();
+        baseTestTCADMaaPMRWorker.run();
+        verify(mockScheduler, times(1)).start();
+    }
+
+    @Test
+    public void testStop() throws Exception {
+        baseTestTCADMaaPMRWorker.stop();
+        verify(mockScheduler, times(1)).shutdown();
+    }
+
+    /**
+     * A helper thread which shuts down the scheduler after some time
+     */
+    public void createShutdownHookThread() {
+        new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    Thread.yield();
+                    Thread.sleep(AnalyticsConstants.TCA_DEFAULT_WORKER_SHUTDOWN_CHECK_INTERVAL_MS * 2);
+                } catch (InterruptedException e) {
+                    LOG.error("Interrupted Exception while running test: {}", e);
+                    throw new RuntimeException(e);
+                }
+                mockIsSchedulerShutdown.getAndSet(true);
+            }
+        }).start();
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMRPublisherJobTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMRPublisherJobTest.java
new file mode 100644
index 0000000..5f7e989
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMRPublisherJobTest.java
@@ -0,0 +1,158 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.worker;
+
+import co.cask.cdap.api.TxRunnable;
+import co.cask.cdap.api.metrics.Metrics;
+import co.cask.cdap.api.worker.WorkerContext;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentMatchers;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertEntity;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
+import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+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: 12/20/2016.
+ */
+public class TCADMaaPMRPublisherJobTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private JobExecutionContext jobExecutionContext;
+    private TCADMaaPMRPublisherJob publisherJob;
+    private JobDataMap jobDataMap;
+    private WorkerContext workerContext;
+    private DMaaPMRPublisher publisher;
+    private Metrics metrics;
+
+    private class TCATestDMaaPMRPublisherJob extends TCADMaaPMRPublisherJob {
+
+        private Map<String, TCAVESAlertEntity> alertEntityMap;
+
+        public TCATestDMaaPMRPublisherJob(Map<String, TCAVESAlertEntity> alertEntityMap) {
+            this.alertEntityMap = alertEntityMap;
+        }
+
+        @Override
+        protected Map<String, TCAVESAlertEntity> getNewAlertsMap(
+                String cdapAlertsTableName, WorkerContext workerContext) {
+            return alertEntityMap;
+        }
+
+        @Override
+        protected void deleteAlertsByKey(String cdapAlertsTableName, WorkerContext workerContext,
+                                         Set<String> rowKeys, Metrics metrics) {
+            // do nothing
+        }
+    }
+
+    @Before
+    public void before() throws Exception {
+
+        jobExecutionContext = mock(JobExecutionContext.class);
+        workerContext = mock(WorkerContext.class);
+
+        metrics = mock(Metrics.class);
+        doNothing().when(metrics).count(anyString(), anyInt());
+        publisher = mock(DMaaPMRPublisher.class);
+
+        jobDataMap = mock(JobDataMap.class);
+        when(jobDataMap.getString(eq(AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME))).thenReturn
+                ("testAlertTableName");
+        when(jobDataMap.get(eq(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME))).thenReturn(workerContext);
+        when(jobDataMap.get(eq(AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME))).thenReturn(publisher);
+        when(jobDataMap.get(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME)).thenReturn(metrics);
+        when(jobExecutionContext.getMergedJobDataMap()).thenReturn(jobDataMap);
+
+
+        publisherJob = new TCADMaaPMRPublisherJob();
+    }
+
+    @Test
+    public void testExecuteWhenNoAlertsFoundInAlertsTable() throws Exception {
+        doNothing().when(workerContext).execute(any(TxRunnable.class));
+        publisherJob.execute(jobExecutionContext);
+        verify(metrics, times(1))
+                .count(eq(CDAPMetricsConstants.TCA_PUBLISHER_NO_NEW_ALERTS_LOOKUP_METRIC), eq(1));
+    }
+
+    @Test
+    public void testExecuteWhenAlertsWereFoundInAlertsTable() throws Exception {
+
+        final DMaaPMRPublisherResponse publisherResponse = mock(DMaaPMRPublisherResponse.class);
+        when(publisherResponse.getResponseCode()).thenReturn(200);
+        when(publisherResponse.getResponseMessage()).thenReturn("success");
+        when(publisherResponse.getPendingMessagesCount()).thenReturn(0);
+        when(publisher.publish(ArgumentMatchers.<String>anyList())).thenReturn(publisherResponse);
+
+        final TCAVESAlertEntity tcavesAlertEntity = mock(TCAVESAlertEntity.class);
+        when(tcavesAlertEntity.getAlertMessage()).thenReturn("testAlertMessage");
+        Map<String, TCAVESAlertEntity> alertEntityMap = new HashMap<>();
+        alertEntityMap.put("key1", tcavesAlertEntity);
+        final TCATestDMaaPMRPublisherJob testPublisherJob = new TCATestDMaaPMRPublisherJob(alertEntityMap);
+        testPublisherJob.execute(jobExecutionContext);
+        verify(metrics, times(1))
+                .count(eq(CDAPMetricsConstants.TCA_PUBLISHER_NEW_ALERTS_METRIC), eq(1));
+        verify(metrics, times(1))
+                .count(eq(CDAPMetricsConstants.TCA_PUBLISHER_SUCCESSFUL_DMAAP_RESPONSE_METRIC), eq(1));
+    }
+
+    @Test
+    public void testExecuteWhenAlertsWereFoundButPublisherReturnedNon200ResponseCode() throws Exception {
+
+        final DMaaPMRPublisherResponse publisherResponse = mock(DMaaPMRPublisherResponse.class);
+        when(publisherResponse.getResponseCode()).thenReturn(500);
+        when(publisherResponse.getResponseMessage()).thenReturn("failed");
+        when(publisherResponse.getPendingMessagesCount()).thenReturn(0);
+        when(publisher.publish(ArgumentMatchers.<String>anyList())).thenReturn(publisherResponse);
+
+        final TCAVESAlertEntity tcavesAlertEntity = mock(TCAVESAlertEntity.class);
+        when(tcavesAlertEntity.getAlertMessage()).thenReturn("testAlertMessage");
+        Map<String, TCAVESAlertEntity> alertEntityMap = new HashMap<>();
+        alertEntityMap.put("key1", tcavesAlertEntity);
+        final TCATestDMaaPMRPublisherJob testPublisherJob = new TCATestDMaaPMRPublisherJob(alertEntityMap);
+        testPublisherJob.execute(jobExecutionContext);
+        verify(metrics, times(1))
+                .count(eq(CDAPMetricsConstants.TCA_PUBLISHER_NEW_ALERTS_METRIC), eq(1));
+        verify(metrics, times(1))
+                .count(eq(CDAPMetricsConstants.TCA_PUBLISHER_UNSUCCESSFUL_DMAAP_RESPONSE_METRIC), eq(1));
+    }
+
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMRSubscriberJobTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMRSubscriberJobTest.java
new file mode 100644
index 0000000..d28dff0
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMRSubscriberJobTest.java
@@ -0,0 +1,135 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.worker;
+
+import co.cask.cdap.api.metrics.Metrics;
+import co.cask.cdap.api.worker.WorkerContext;
+import com.google.common.collect.ImmutableList;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 12/20/2016.
+ */
+public class TCADMaaPMRSubscriberJobTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private JobExecutionContext jobExecutionContext;
+    private TCADMaaPMRSubscriberJob subscriberJob;
+    private JobDataMap jobDataMap;
+    private WorkerContext workerContext;
+    private DMaaPMRSubscriber subscriber;
+    private Metrics metrics;
+
+
+    @Before
+    public void before() throws Exception {
+
+        jobExecutionContext = mock(JobExecutionContext.class);
+        workerContext = mock(WorkerContext.class);
+
+        metrics = mock(Metrics.class);
+        doNothing().when(metrics).count(anyString(), anyInt());
+        subscriber = mock(DMaaPMRSubscriber.class);
+
+        jobDataMap = mock(JobDataMap.class);
+        when(jobDataMap.getString(eq(AnalyticsConstants.CDAP_STREAM_VARIABLE_NAME))).thenReturn
+                (CDAPComponentsConstants.TCA_DEFAULT_SUBSCRIBER_OUTPUT_NAME_STREAM);
+        when(jobDataMap.get(eq(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME))).thenReturn(workerContext);
+        when(jobDataMap.get(eq(AnalyticsConstants.DMAAP_SUBSCRIBER_VARIABLE_NAME))).thenReturn(subscriber);
+        when(jobDataMap.get(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME)).thenReturn(metrics);
+        when(jobExecutionContext.getMergedJobDataMap()).thenReturn(jobDataMap);
+
+        doNothing().when(workerContext).write(anyString(), anyString());
+
+        subscriberJob = new TCADMaaPMRSubscriberJob();
+    }
+
+    @Test
+    public void testExecuteWhenMessagesAreFound() throws Exception {
+        final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);
+        when(subscriberResponse.getResponseCode()).thenReturn(200);
+        when(subscriberResponse.getResponseMessage()).thenReturn("testMessage");
+        when(subscriberResponse.getFetchedMessages()).thenReturn(ImmutableList.of("testMessage1", "testMessage1"));
+        when(subscriber.fetchMessages()).thenReturn(subscriberResponse);
+        subscriberJob.execute(jobExecutionContext);
+        verify(metrics, Mockito.times(1)).count(eq(CDAPMetricsConstants
+                .DMAAP_MR_SUBSCRIBER_TOTAL_MESSAGES_PROCESSED_METRIC), eq(2));
+    }
+
+    @Test
+    public void testExecuteWhenNoMessagesFound() throws Exception {
+        final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);
+        when(subscriberResponse.getResponseCode()).thenReturn(200);
+        when(subscriberResponse.getResponseMessage()).thenReturn("no messages");
+        when(subscriberResponse.getFetchedMessages()).thenReturn(Collections.<String>emptyList());
+        when(subscriber.fetchMessages()).thenReturn(subscriberResponse);
+        subscriberJob.execute(jobExecutionContext);
+        verify(metrics, Mockito.times(1)).count(eq(CDAPMetricsConstants
+                .DMAAP_MR_SUBSCRIBER_RESPONSES_WITH_NO_MESSAGES_METRIC), eq(1));
+    }
+
+
+    @Test
+    public void testExecuteWhenSubscriberReturnNonSuccessfulReturnCode() throws Exception {
+        final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);
+        when(subscriberResponse.getResponseCode()).thenReturn(500);
+        when(subscriber.fetchMessages()).thenReturn(subscriberResponse);
+        subscriberJob.execute(jobExecutionContext);
+        verify(metrics, Mockito.times(1)).count(eq(CDAPMetricsConstants
+                .DMAAP_MR_SUBSCRIBER_UNSUCCESSFUL_RESPONSES_METRIC), eq(1));
+    }
+
+    @Test(expected = DCAEAnalyticsRuntimeException.class)
+    public void testExecuteWhenWritingToCDAPStreamThrowsException() throws Exception {
+        final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);
+        when(subscriberResponse.getResponseCode()).thenReturn(200);
+        when(subscriberResponse.getFetchedMessages()).thenReturn(Arrays.asList("TestMessage"));
+        when(subscriber.fetchMessages()).thenReturn(subscriberResponse);
+        doThrow(new IOException()).when(workerContext).write(anyString(), anyString());
+        subscriberJob.execute(jobExecutionContext);
+    }
+
+
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPPublisherWorkerTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPPublisherWorkerTest.java
new file mode 100644
index 0000000..f2eeef8
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPPublisherWorkerTest.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.apod.analytics.cdap.tca.worker;
+
+import co.cask.cdap.api.worker.WorkerConfigurer;
+import co.cask.cdap.api.worker.WorkerContext;
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+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: 12/20/2016.
+ */
+public class TCADMaaPPublisherWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private static final String VES_ALERTS_TABLE_NAME = "vesAlertsTable";
+
+    private WorkerConfigurer workerConfigurer;
+    private WorkerContext workerContext;
+    private TCADMaaPPublisherWorker publisherWorker;
+
+    @Before
+    public void before() throws Exception {
+        workerConfigurer = mock(WorkerConfigurer.class);
+        workerContext = mock(WorkerContext.class);
+        doNothing().when(workerConfigurer).setName(anyString());
+        doNothing().when(workerConfigurer).setDescription(anyString());
+        publisherWorker = new TCADMaaPPublisherWorker(VES_ALERTS_TABLE_NAME);
+
+    }
+
+    @Test
+    public void testConfigure() throws Exception {
+        publisherWorker.configure(workerConfigurer);
+        verify(workerConfigurer, times(1))
+                .setName(eq(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER));
+        verify(workerConfigurer, times(1))
+                .setDescription(eq(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_DESCRIPTION_WORKER));
+    }
+
+    @Test(expected = CDAPSettingsException.class)
+    public void testInitializeWhenSettingsHaveErrors() throws Exception {
+        publisherWorker.initialize(workerContext);
+    }
+
+    @Test
+    public void testInitializeWhenSettingsAreValid() throws Exception {
+        when(workerContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        publisherWorker.initialize(workerContext);
+    }
+
+}
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPSubscriberWorkerTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPSubscriberWorkerTest.java
new file mode 100644
index 0000000..d74669b
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPSubscriberWorkerTest.java
@@ -0,0 +1,81 @@
+/*
+ * ===============================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.apod.analytics.cdap.tca.worker;
+
+import co.cask.cdap.api.worker.WorkerConfigurer;
+import co.cask.cdap.api.worker.WorkerContext;
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+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: 12/20/2016.
+ */
+public class TCADMaaPSubscriberWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private static final String TEST_SUBSCRIBER_OUTPUT_STREAM_NAME = "testSubscriberOutputStream";
+
+    private WorkerConfigurer workerConfigurer;
+    private WorkerContext workerContext;
+    private TCADMaaPSubscriberWorker subscriberWorker;
+
+    @Before
+    public void before() throws Exception {
+        workerConfigurer = mock(WorkerConfigurer.class);
+        workerContext = mock(WorkerContext.class);
+        doNothing().when(workerConfigurer).setName(anyString());
+        doNothing().when(workerConfigurer).setDescription(anyString());
+        subscriberWorker =
+                new TCADMaaPSubscriberWorker(TEST_SUBSCRIBER_OUTPUT_STREAM_NAME);
+
+    }
+
+    @Test
+    public void testConfigure() throws Exception {
+        subscriberWorker.configure(workerConfigurer);
+        verify(workerConfigurer, times(1))
+                .setName(eq(CDAPComponentsConstants.TCA_FIXED_DMAAP_SUBSCRIBER_WORKER));
+        verify(workerConfigurer, times(1))
+                .setDescription(eq(CDAPComponentsConstants.TCA_FIXED_DMAAP_SUBSCRIBER_DESCRIPTION_WORKER));
+    }
+
+    @Test(expected = CDAPSettingsException.class)
+    public void testInitializeWhenSettingsHaveErrors() throws Exception {
+        subscriberWorker.initialize(workerContext);
+    }
+
+    @Test
+    public void testInitializeWhenSettingsAreValid() throws Exception {
+        when(workerContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        subscriberWorker.initialize(workerContext);
+    }
+
+}