Add support for ABATED alerts within CDAP TCA

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

8 files changed, 284 insertions, 32 deletions

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
index 6d84561..9c72183 100644
--- 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
@@ -65,12 +65,15 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
     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_APP_CONFIG_FILE_LOCATION = "data/json/config/controller_app_config.json";
+    protected static final String TCA_ALERT_JSON_FILE_LOCATION = "data/json/facade/tca_ves_cef_response.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";
+            "data/properties/tca_controller_policy_from_json.properties";
 
 
     protected static final String TCA_TEST_APP_CONFIG_NAME = "testTCAAppName";
@@ -87,7 +90,7 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
      *
      * @return test TCA Policy Object
      */
-    protected TCAPolicy getSampleTCAPolicy() {
+    protected static TCAPolicy getSampleTCAPolicy() {
         return deserializeJsonFileToModel(TCA_POLICY_JSON_FILE_LOCATION, TCAPolicy.class);
     }
 
@@ -96,7 +99,7 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
      *
      * @return test {@link TCAPolicyPreferences}
      */
-    protected TCAPolicyPreferences getSampleTCAPolicyPreferences() {
+    protected static TCAPolicyPreferences getSampleTCAPolicyPreferences() {
         return deserializeJsonFileToModel(TCA_POLICY_JSON_FILE_LOCATION, TCAPolicyPreferences.class);
     }
 
@@ -107,7 +110,7 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
      *
      * @throws Exception Exception
      */
-    protected List<EventListener> getCEFMessages() throws Exception {
+    protected static List<EventListener> getCEFMessages() throws Exception {
         final String cefMessageAsString = fromStream(CEF_MESSAGES_JSON_FILE_LOCATION);
         final TypeReference<List<EventListener>> eventListenerListTypeReference =
                 new TypeReference<List<EventListener>>() {
@@ -122,7 +125,7 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
      *
      * @throws Exception Exception
      */
-    protected String getValidCEFMessage() throws Exception {
+    protected static String getValidCEFMessage() throws Exception {
         return fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
     }
 
@@ -134,7 +137,7 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
      *
      * @throws Exception Exception
      */
-    protected EventListener getCEFEventListener() throws Exception {
+    protected static EventListener getCEFEventListener() throws Exception {
         final String cefMessageAsString = fromStream(CEF_MESSAGE_JSON_FILE_LOCATION);
         return ANALYTICS_MODEL_OBJECT_MAPPER.readValue(cefMessageAsString, EventListener.class);
     }
@@ -257,23 +260,16 @@ public abstract class BaseAnalyticsCDAPTCAUnitTest extends BaseDCAEAnalyticsUnit
     }
 
     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;
+        return createNewFlowletContextFromPropertiesFile(TCA_CONTROLLER_POLICY_FILE_LOCATION);
     }
 
     protected static FlowletContext getTestFlowletContextWithValidPolicyFromJSON() {
+        return createNewFlowletContextFromPropertiesFile(TCA_CONTROLLER_POLICY_FROM_JSON_FILE_LOCATION);
+    }
+
+    private static FlowletContext createNewFlowletContextFromPropertiesFile(final String propertyFileLocation) {
         final Properties controllerProperties =
-                AnalyticsModelIOUtils.loadPropertiesFile(TCA_CONTROLLER_POLICY_FROM_JSON_FILE_LOCATION,
-                        new Properties());
+                AnalyticsModelIOUtils.loadPropertiesFile(propertyFileLocation, new Properties());
 
         Map<String, String> runtimeArgs = new LinkedHashMap<>();
         for (Map.Entry<Object, Object> property : controllerProperties.entrySet()) {
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
index 05444d7..b483cdf 100644
--- 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
@@ -63,12 +63,13 @@ public class TCAVESCollectorFlowTest extends BaseAnalyticsCDAPTCAUnitTest {
         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_ABATEMENT_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));
+        assertThat("There must be four connections in VES Collector Flow", connections.size(), is(4));
 
     }
 
diff --git a/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESAlertsAbatementFlowletTest.java b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESAlertsAbatementFlowletTest.java
new file mode 100644
index 0000000..baf42cc
--- /dev/null
+++ b/dcae-analytics-cdap-tca/src/test/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESAlertsAbatementFlowletTest.java
@@ -0,0 +1,234 @@
+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 co.cask.cdap.api.flow.flowlet.OutputEmitter;
+import com.google.common.collect.ImmutableList;
+import org.apache.commons.lang3.tuple.Pair;
+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.domain.tca.ThresholdCalculatorOutput;
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAAlertsAbatementEntity;
+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.cdap.tca.settings.TCAPolicyPreferences;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.ControlLoopEventStatus;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
+import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;
+
+import java.util.Date;
+import java.util.List;
+
+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.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Author: rs153v (Rajiv Singla) . Creation Date: 9/12/2017.
+ */
+public class TCAVESAlertsAbatementFlowletTest extends BaseAnalyticsCDAPTCAUnitTest {
+
+    private static final TCAPolicyPreferences sampleTCAPolicyPreferences = getSampleTCAPolicyPreferences();
+    private static final List<MetricsPerEventName> metricsPerEventNames = sampleTCAPolicyPreferences
+            .getMetricsPerEventName();
+    private final OutputEmitter<String> mockOutputEmitter = mock(OutputEmitter.class);
+
+    private class TestTCAVESAlertsAbatementFlowlet extends TCAVESAlertsAbatementFlowlet {
+
+        public TestTCAVESAlertsAbatementFlowlet(String tcaAlertsAbatementTableName) {
+            super(tcaAlertsAbatementTableName);
+            this.alertsAbatementOutputEmitter = mockOutputEmitter;
+            doNothing().when(mockOutputEmitter).emit(any(String.class));
+        }
+    }
+
+    @Test
+    public void testConfigure() throws Exception {
+        final TCAVESAlertsAbatementFlowlet tcavesAlertsAbatementFlowlet =
+                new TCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+        assertFlowletNameAndDescription(CDAPComponentsConstants.TCA_FIXED_VES_ALERTS_ABATEMENT_NAME_FLOWLET,
+                CDAPComponentsConstants.TCA_FIXED_VES_ALERTS_ABATEMENT_DESCRIPTION_FLOWLET,
+                tcavesAlertsAbatementFlowlet);
+    }
+
+    @Test(expected = CDAPSettingsException.class)
+    public void testDetermineAbatementAlertsWhenViolatedMetricsEventNameIsBlank() throws Exception {
+
+        final TestTCAVESAlertsAbatementFlowlet tcaAlertsAbatementFlowlet =
+                new TestTCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+        final Threshold violatedThreshold = getViolatedThreshold(ControlLoopEventStatus.ONSET);
+        final ThresholdCalculatorOutput mockThresholdCalculatorOutput =
+                getMockThresholdCalculatorOutput(violatedThreshold);
+        when(mockThresholdCalculatorOutput.getViolatedMetricsPerEventName()).thenReturn("");
+
+        tcaAlertsAbatementFlowlet.determineAbatementAlerts(mockThresholdCalculatorOutput);
+    }
+
+    @Test
+    public void testDetermineAbatementAlertsWhenControlLoopTypeIsONSET() throws Exception {
+
+        final String testTCAAlertsAbatementTableName = "testTCAAlertsAbatementTableName";
+        final TestTCAVESAlertsAbatementFlowlet tcaAlertsAbatementFlowlet =
+                new TestTCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+
+        final FlowletContext mockFlowletContext = mock(FlowletContext.class);
+        final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
+        when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+        tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
+
+        doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+
+        final Threshold violatedThreshold = getViolatedThreshold(ControlLoopEventStatus.ONSET);
+        final ThresholdCalculatorOutput mockThresholdCalculatorOutput =
+                getMockThresholdCalculatorOutput(violatedThreshold);
+
+        tcaAlertsAbatementFlowlet.determineAbatementAlerts(mockThresholdCalculatorOutput);
+        verify(mockObjectMappedTable,
+                times(1)).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        verify(mockOutputEmitter, times(1)).emit(any(String.class));
+
+    }
+
+
+    @Test
+    public void testDetermineAbatementAlertsWhenControlLoopTypeIsABATEDAndNoPreviousAlertWasSent() throws Exception {
+
+        final String testTCAAlertsAbatementTableName = "testTCAAlertsAbatementTableName";
+        final TestTCAVESAlertsAbatementFlowlet tcaAlertsAbatementFlowlet =
+                new TestTCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+
+        final FlowletContext mockFlowletContext = mock(FlowletContext.class);
+        final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
+        when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+        tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
+
+        doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        final TCAAlertsAbatementEntity tcaAlertsAbatementEntity = mock(TCAAlertsAbatementEntity.class);
+        when(mockObjectMappedTable.read(any(String.class))).thenReturn(tcaAlertsAbatementEntity);
+        when(tcaAlertsAbatementEntity.getAbatementSentTS()).thenReturn(null);
+
+        final Threshold violatedThreshold = getViolatedThreshold(ControlLoopEventStatus.ABATED);
+        final ThresholdCalculatorOutput mockThresholdCalculatorOutput =
+                getMockThresholdCalculatorOutput(violatedThreshold);
+
+        tcaAlertsAbatementFlowlet.determineAbatementAlerts(mockThresholdCalculatorOutput);
+        verify(mockObjectMappedTable,
+                times(1)).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        verify(mockOutputEmitter, times(1)).emit(any(String.class));
+
+    }
+
+    @Test
+    public void testDetermineAbatementAlertsWhenControlLoopTypeIsABATEDAndPreviousAlertWasAlreadySent() throws
+            Exception {
+
+        final String testTCAAlertsAbatementTableName = "testTCAAlertsAbatementTableName";
+        final TestTCAVESAlertsAbatementFlowlet tcaAlertsAbatementFlowlet =
+                new TestTCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+
+        final FlowletContext mockFlowletContext = mock(FlowletContext.class);
+        final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
+        when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+        tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
+
+        doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        final TCAAlertsAbatementEntity tcaAlertsAbatementEntity = mock(TCAAlertsAbatementEntity.class);
+        when(mockObjectMappedTable.read(any(String.class))).thenReturn(tcaAlertsAbatementEntity);
+        final long time = new Date().getTime();
+        when(tcaAlertsAbatementEntity.getAbatementSentTS()).thenReturn(Long.toString(time));
+
+        final Threshold violatedThreshold = getViolatedThreshold(ControlLoopEventStatus.ABATED);
+        final ThresholdCalculatorOutput mockThresholdCalculatorOutput =
+                getMockThresholdCalculatorOutput(violatedThreshold);
+
+        tcaAlertsAbatementFlowlet.determineAbatementAlerts(mockThresholdCalculatorOutput);
+        verify(mockObjectMappedTable,
+                times(0)).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        verify(mockOutputEmitter, times(0)).emit(any(String.class));
+
+    }
+
+
+    @Test
+    public void testDetermineAbatementAlertsWhenControlLoopTypeIsABATEDAndNoPreviousONSETEventFound() throws
+            Exception {
+
+        final String testTCAAlertsAbatementTableName = "testTCAAlertsAbatementTableName";
+        final TestTCAVESAlertsAbatementFlowlet tcaAlertsAbatementFlowlet =
+                new TestTCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+
+        final FlowletContext mockFlowletContext = mock(FlowletContext.class);
+        final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
+        when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+        tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
+
+        doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        when(mockObjectMappedTable.read(any(String.class))).thenReturn(null);
+
+        final Threshold violatedThreshold = getViolatedThreshold(ControlLoopEventStatus.ABATED);
+        final ThresholdCalculatorOutput mockThresholdCalculatorOutput =
+                getMockThresholdCalculatorOutput(violatedThreshold);
+
+        tcaAlertsAbatementFlowlet.determineAbatementAlerts(mockThresholdCalculatorOutput);
+        verify(mockObjectMappedTable,
+                times(0)).write(any(String.class), any(TCAAlertsAbatementEntity.class));
+        verify(mockOutputEmitter, times(0)).emit(any(String.class));
+
+    }
+
+    @Test(expected = CDAPSettingsException.class)
+    public void testDetermineAbatementAlertsWhenControlLoopTypeIsNotOnsetOrAbated() throws
+            Exception {
+        final TestTCAVESAlertsAbatementFlowlet tcaAlertsAbatementFlowlet =
+                new TestTCAVESAlertsAbatementFlowlet("testTCAAlertsAbatementTableName");
+        final Threshold violatedThreshold = getViolatedThreshold(ControlLoopEventStatus.CONTINUE);
+        final ThresholdCalculatorOutput mockThresholdCalculatorOutput =
+                getMockThresholdCalculatorOutput(violatedThreshold);
+
+        tcaAlertsAbatementFlowlet.determineAbatementAlerts(mockThresholdCalculatorOutput);
+
+    }
+
+    private static Threshold getViolatedThreshold(final ControlLoopEventStatus controlLoopEventStatus) {
+        final Threshold violatedThreshold = Threshold.copy(metricsPerEventNames.get(0).getThresholds().get(0));
+        violatedThreshold.setClosedLoopEventStatus(controlLoopEventStatus);
+        return violatedThreshold;
+    }
+
+
+    private static ThresholdCalculatorOutput getMockThresholdCalculatorOutput(final Threshold violatedThreshold) throws
+            Exception {
+
+        final MetricsPerEventName violatedMetricsPerEventName =
+                MetricsPerEventName.copy(metricsPerEventNames.get(0));
+        violatedMetricsPerEventName.setThresholds(ImmutableList.of(violatedThreshold));
+        return getMockThresholdCalculatorOutput(
+                fromStream(CEF_MESSAGE_JSON_FILE_LOCATION),
+                fromStream(TCA_POLICY_JSON_FILE_LOCATION),
+                TCAUtils.writeValueAsString(violatedMetricsPerEventName),
+                fromStream(TCA_ALERT_JSON_FILE_LOCATION)
+        );
+    }
+
+
+    private static ThresholdCalculatorOutput getMockThresholdCalculatorOutput(final String cefMessage,
+                                                                              final String tcaPolicy,
+                                                                              final String violatedMetricsPerEventName,
+                                                                              final String alertMessage) {
+        final ThresholdCalculatorOutput thresholdCalculatorOutput = mock(ThresholdCalculatorOutput.class);
+        when(thresholdCalculatorOutput.getCefMessage()).thenReturn(cefMessage);
+        when(thresholdCalculatorOutput.getTcaPolicy()).thenReturn(tcaPolicy);
+        when(thresholdCalculatorOutput.getViolatedMetricsPerEventName()).thenReturn(violatedMetricsPerEventName);
+        when(thresholdCalculatorOutput.getAlertMessage()).thenReturn(alertMessage);
+        return thresholdCalculatorOutput;
+    }
+
+}
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
index b70234a..8e7884e 100644
--- 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
@@ -29,6 +29,7 @@ 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.domain.tca.ThresholdCalculatorOutput;
 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;
@@ -119,16 +120,16 @@ public class TCAVESThresholdViolationCalculatorFlowletTest extends BaseAnalytics
         final FlowletContext flowletContext =
                 initializeFlowlet(thresholdViolationCalculatorFlowlet, vesMessageStatusTable);
         final TCAPolicy policy = CDAPTCAUtils.getValidatedTCAPolicyPreferences(flowletContext);
-        final Threshold threshold = policy.getMetricsPerFunctionalRole().get(0).getThresholds().get(0);
+        final Threshold threshold = policy.getMetricsPerEventName().get(0).getThresholds().get(0);
         final Long thresholdValue = threshold.getThresholdValue();
         final EventListener thresholdViolatingMessage = getCEFEventListener();
-        thresholdViolatingMessage.getEvent().getMeasurementsForVfScalingFields().getVNicUsageArray().get(0).setBytesIn
-                (thresholdValue - 1);
+        thresholdViolatingMessage.getEvent().getMeasurementsForVfScalingFields().getVNicPerformanceArray().
+                get(0).setReceivedBroadcastPacketsAccumulated(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());
+        verify(outputEmitter, times(1)).emit(any(ThresholdCalculatorOutput.class));
     }
 
     private static TCATestVESThresholdViolationCalculatorFlowlet createTestViolationCalculator(
@@ -142,11 +143,12 @@ public class TCAVESThresholdViolationCalculatorFlowletTest extends BaseAnalytics
     }
 
     private static <T extends TCAVESThresholdViolationCalculatorFlowlet> FlowletContext initializeFlowlet(
-            T calculatorFlowlet, ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable) {
+            T calculatorFlowlet, ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable) throws Exception {
         final FlowletContext mockFlowletContext = getTestFlowletContextWithValidPolicy();
         when(mockFlowletContext.getDataset(anyString())).thenReturn(vesMessageStatusTable);
         when(mockFlowletContext.getInstanceId()).thenReturn(1);
         ApplicationSpecification mockApplicationSpecification = Mockito.mock(ApplicationSpecification.class);
+        when(mockApplicationSpecification.getConfiguration()).thenReturn(fromStream(TCA_APP_CONFIG_FILE_LOCATION));
         when(mockFlowletContext.getApplicationSpecification()).thenReturn(mockApplicationSpecification);
         when(mockApplicationSpecification.getName()).thenReturn("TestTCAAppName");
         try {
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
index 75f336d..e9a0284 100644
--- 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
@@ -21,7 +21,9 @@
 package org.openecomp.dcae.apod.analytics.cdap.tca.utils;
 
 import co.cask.cdap.api.RuntimeContext;
+import co.cask.cdap.api.app.ApplicationSpecification;
 import org.junit.Test;
+import org.mockito.Mockito;
 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;
@@ -41,8 +43,11 @@ public class CDAPTCAUtilsTest extends BaseAnalyticsCDAPTCAUnitTest {
     public void testGetValidatedTCAAppPreferences() throws Exception {
         RuntimeContext runtimeContext = mock(RuntimeContext.class);
         when(runtimeContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        ApplicationSpecification mockApplicationSpecification = Mockito.mock(ApplicationSpecification.class);
+        when(mockApplicationSpecification.getConfiguration()).thenReturn(fromStream(TCA_APP_CONFIG_FILE_LOCATION));
+        when(runtimeContext.getApplicationSpecification()).thenReturn(mockApplicationSpecification);
         TCAAppPreferences validatedTCAAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(runtimeContext);
-        assertEquals(validatedTCAAppPreferences.getSubscriberHostName(), "mrlocal-mtnjftle01.homer.com");
+        assertEquals(validatedTCAAppPreferences.getSubscriberHostName(), "HOSTNAME");
     }
 
     @Test
@@ -53,7 +58,7 @@ public class CDAPTCAUtilsTest extends BaseAnalyticsCDAPTCAUnitTest {
         assertThat("Policy Domain must be measurementsForVfScaling",
                 tcaPolicy.getDomain(), is("measurementsForVfScaling"));
         assertThat("Policy must have 2 metrics per functional roles",
-                tcaPolicy.getMetricsPerFunctionalRole().size(), is(2));
+                tcaPolicy.getMetricsPerEventName().size(), is(2));
     }
 
     @Test
@@ -64,7 +69,7 @@ public class CDAPTCAUtilsTest extends BaseAnalyticsCDAPTCAUnitTest {
         assertThat("Policy Domain must be measurementsForVfScaling",
                 tcaPolicy.getDomain(), is("measurementsForVfScaling"));
         assertThat("Policy must have 2 metrics per functional roles",
-                tcaPolicy.getMetricsPerFunctionalRole().size(), is(2));
+                tcaPolicy.getMetricsPerEventName().size(), is(2));
     }
 
 }
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
index bfe288f..947ed0f 100644
--- 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
@@ -25,7 +25,7 @@ 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.MetricsPerEventName;
 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
 
 import java.util.Collections;
@@ -57,7 +57,7 @@ public class TCAPolicyPreferencesValidatorTest extends BaseAnalyticsCDAPTCAUnitT
     @Test
     public void testValidateAppSettingsWhenDomainIsNullAndFunctionRoleIsEmpty() throws Exception {
         tcaPolicyPreferences.setDomain(null);
-        tcaPolicyPreferences.setMetricsPerFunctionalRole(Collections.<MetricsPerFunctionalRole>emptyList());
+        tcaPolicyPreferences.setMetricsPerEventName(Collections.<MetricsPerEventName>emptyList());
         final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
                 tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
         assertTrue(validationResponse.hasErrors());
@@ -66,7 +66,7 @@ public class TCAPolicyPreferencesValidatorTest extends BaseAnalyticsCDAPTCAUnitT
 
     @Test
     public void testValidateAppSettingsWhenThresholdIsEmpty() throws Exception {
-        tcaPolicyPreferences.getMetricsPerFunctionalRole().get(0).setThresholds(Collections.<Threshold>emptyList());
+        tcaPolicyPreferences.getMetricsPerEventName().get(0).setThresholds(Collections.<Threshold>emptyList());
         final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
                 tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
         assertTrue(validationResponse.hasErrors());
@@ -75,7 +75,7 @@ public class TCAPolicyPreferencesValidatorTest extends BaseAnalyticsCDAPTCAUnitT
 
     @Test
     public void testValidateAppSettingsWhenThresholdPathIsMissing() throws Exception {
-        tcaPolicyPreferences.getMetricsPerFunctionalRole().get(0).getThresholds().get(0).setFieldPath(null);
+        tcaPolicyPreferences.getMetricsPerEventName().get(0).getThresholds().get(0).setFieldPath(null);
         final GenericValidationResponse<TCAPolicyPreferences> validationResponse =
                 tcaPolicyPreferencesValidator.validateAppSettings(tcaPolicyPreferences);
         assertTrue(validationResponse.hasErrors());
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
index f2eeef8..2324b38 100644
--- 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
@@ -20,10 +20,12 @@
 
 package org.openecomp.dcae.apod.analytics.cdap.tca.worker;
 
+import co.cask.cdap.api.app.ApplicationSpecification;
 import co.cask.cdap.api.worker.WorkerConfigurer;
 import co.cask.cdap.api.worker.WorkerContext;
 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.exception.CDAPSettingsException;
 import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
@@ -46,6 +48,7 @@ public class TCADMaaPPublisherWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
     private WorkerConfigurer workerConfigurer;
     private WorkerContext workerContext;
     private TCADMaaPPublisherWorker publisherWorker;
+    private ApplicationSpecification mockApplicationSpecification;
 
     @Before
     public void before() throws Exception {
@@ -53,6 +56,8 @@ public class TCADMaaPPublisherWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
         workerContext = mock(WorkerContext.class);
         doNothing().when(workerConfigurer).setName(anyString());
         doNothing().when(workerConfigurer).setDescription(anyString());
+        mockApplicationSpecification = Mockito.mock(ApplicationSpecification.class);
+        when(workerContext.getApplicationSpecification()).thenReturn(mockApplicationSpecification);
         publisherWorker = new TCADMaaPPublisherWorker(VES_ALERTS_TABLE_NAME);
 
     }
@@ -68,12 +73,14 @@ public class TCADMaaPPublisherWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
 
     @Test(expected = CDAPSettingsException.class)
     public void testInitializeWhenSettingsHaveErrors() throws Exception {
+        when(mockApplicationSpecification.getConfiguration()).thenReturn("{}");
         publisherWorker.initialize(workerContext);
     }
 
     @Test
     public void testInitializeWhenSettingsAreValid() throws Exception {
         when(workerContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        when(mockApplicationSpecification.getConfiguration()).thenReturn(fromStream(TCA_APP_CONFIG_FILE_LOCATION));
         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
index d74669b..e5b1f9a 100644
--- 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
@@ -20,10 +20,12 @@
 
 package org.openecomp.dcae.apod.analytics.cdap.tca.worker;
 
+import co.cask.cdap.api.app.ApplicationSpecification;
 import co.cask.cdap.api.worker.WorkerConfigurer;
 import co.cask.cdap.api.worker.WorkerContext;
 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.exception.CDAPSettingsException;
 import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;
@@ -46,11 +48,14 @@ public class TCADMaaPSubscriberWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
     private WorkerConfigurer workerConfigurer;
     private WorkerContext workerContext;
     private TCADMaaPSubscriberWorker subscriberWorker;
+    private ApplicationSpecification mockApplicationSpecification;
 
     @Before
     public void before() throws Exception {
         workerConfigurer = mock(WorkerConfigurer.class);
         workerContext = mock(WorkerContext.class);
+        mockApplicationSpecification = Mockito.mock(ApplicationSpecification.class);
+        when(workerContext.getApplicationSpecification()).thenReturn(mockApplicationSpecification);
         doNothing().when(workerConfigurer).setName(anyString());
         doNothing().when(workerConfigurer).setDescription(anyString());
         subscriberWorker =
@@ -69,12 +74,14 @@ public class TCADMaaPSubscriberWorkerTest extends BaseAnalyticsCDAPTCAUnitTest {
 
     @Test(expected = CDAPSettingsException.class)
     public void testInitializeWhenSettingsHaveErrors() throws Exception {
+        when(mockApplicationSpecification.getConfiguration()).thenReturn("{}");
         subscriberWorker.initialize(workerContext);
     }
 
     @Test
     public void testInitializeWhenSettingsAreValid() throws Exception {
         when(workerContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+        when(mockApplicationSpecification.getConfiguration()).thenReturn(fromStream(TCA_APP_CONFIG_FILE_LOCATION));
         subscriberWorker.initialize(workerContext);
     }