<version>2.18.0</version>\r
<scope>test</scope>\r
</dependency>\r
+ <dependency>\r
+ <groupId>org.assertj</groupId>\r
+ <artifactId>assertj-core</artifactId>\r
+ <version>3.8.0</version>\r
+ <scope>test</scope>\r
+ </dependency>\r
</dependencies>\r
\r
<reporting>\r
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * Copyright (C) 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.onap.dcae.commonFunction;
+
+import com.google.gson.JsonObject;
+
+import java.util.List;
+
+class Event {
+ final JsonObject filter;
+ final List<Processor> processors;
+
+ Event(JsonObject filter, List<Processor> processors) {
+ this.filter = filter;
+ this.processors = processors;
+ }
+}
import com.att.nsa.clock.SaClock;\r
import com.att.nsa.logging.LoggingContext;\r
import com.att.nsa.logging.log4j.EcompFields;\r
-import com.google.gson.JsonArray;\r
-import com.google.gson.JsonParser;\r
-import org.json.JSONArray;\r
+import com.google.common.reflect.TypeToken;\r
+import com.google.gson.Gson;\r
import org.json.JSONObject;\r
import org.slf4j.Logger;\r
import org.slf4j.LoggerFactory;\r
\r
-import java.io.IOException;\r
import java.io.FileReader;\r
-import java.lang.reflect.Constructor;\r
+import java.io.IOException;\r
import java.lang.reflect.Method;\r
+import java.lang.reflect.Type;\r
import java.text.SimpleDateFormat;\r
import java.util.Arrays;\r
import java.util.Date;\r
import java.util.HashMap;\r
+import java.util.List;\r
import java.util.TimeZone;\r
\r
+\r
public class EventProcessor implements Runnable {\r
\r
- private static final Logger log = LoggerFactory.getLogger(EventProcessor.class);\r
- private static final String EVENT_LITERAL = "event";\r
- private static final String COMMON_EVENT_HEADER = "commonEventHeader";\r
-\r
- private static HashMap<String, String[]> streamidHash = new HashMap<>();\r
- public JSONObject event;\r
-\r
- public EventProcessor() {\r
- log.debug("EventProcessor: Default Constructor");\r
-\r
- String[] list = CommonStartup.streamid.split("\\|");\r
- for (String aList : list) {\r
- String domain = aList.split("=")[0];\r
-\r
- String[] streamIdList = aList.substring(aList.indexOf('=') + 1).split(",");\r
-\r
- log.debug(String.format("Domain: %s streamIdList:%s", domain, Arrays.toString(streamIdList)));\r
- streamidHash.put(domain, streamIdList);\r
- }\r
-\r
- }\r
-\r
- @Override\r
- public void run() {\r
-\r
- try {\r
-\r
- event = CommonStartup.fProcessingInputQueue.take();\r
-\r
- while (event != null) {\r
- // As long as the producer is running we remove elements from\r
- // the queue.\r
- log.info("QueueSize:" + CommonStartup.fProcessingInputQueue.size()+ "\tEventProcessor\tRemoving element: " + event );\r
- \r
- String uuid = event.get("VESuniqueId").toString();\r
- LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);\r
- localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());\r
-\r
- log.debug("event.VESuniqueId" + event.get("VESuniqueId") + "event.commonEventHeader.domain:"\r
- + event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER).getString("domain"));\r
- String[] streamIdList = streamidHash\r
- .get(event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER).getString("domain"));\r
- log.debug("streamIdList:" + streamIdList);\r
-\r
- if (streamIdList.length == 0) {\r
- log.error("No StreamID defined for publish - Message dropped" + event);\r
- } else {\r
- for (String aStreamIdList : streamIdList) {\r
- log.info("Invoking publisher for streamId:" + aStreamIdList);\r
- this.overrideEvent();\r
-\r
- EventPublisherHash.getInstance().sendEvent(event, aStreamIdList);\r
-\r
- }\r
- }\r
- log.debug("Message published" + event);\r
- event = CommonStartup.fProcessingInputQueue.take();\r
-\r
- }\r
- } catch (InterruptedException e) {\r
- log.error("EventProcessor InterruptedException" + e.getMessage());\r
- Thread.currentThread().interrupt();\r
- }\r
-\r
- }\r
-\r
- @SuppressWarnings({ "unchecked", "rawtypes" })\r
- public void overrideEvent() {\r
- // Set collector timestamp in event payload before publish\r
- final Date currentTime = new Date();\r
- final SimpleDateFormat sdf = new SimpleDateFormat("EEE, MM dd yyyy hh:mm:ss z");\r
- sdf.setTimeZone(TimeZone.getTimeZone("GMT"));\r
-\r
- JSONObject collectorTimeStamp = new JSONObject().put("collectorTimeStamp", sdf.format(currentTime));\r
- JSONObject commonEventHeaderkey = event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER);\r
- commonEventHeaderkey.put("internalHeaderFields", collectorTimeStamp);\r
- event.getJSONObject(EVENT_LITERAL).put(COMMON_EVENT_HEADER, commonEventHeaderkey);\r
-\r
- if (CommonStartup.eventTransformFlag == 1) {\r
- // read the mapping json file\r
- final JsonParser parser = new JsonParser();\r
- FileReader fr = null;\r
- try {\r
- fr = new FileReader("./etc/eventTransform.json");\r
- final JsonArray jo = (JsonArray) parser.parse(fr);\r
- log.info("parse eventTransform.json");\r
- // now convert to org.json\r
- final String jsonText = jo.toString();\r
- final JSONArray topLevel = new JSONArray(jsonText);\r
-\r
- Class[] paramJSONObject = new Class[1];\r
- paramJSONObject[0] = JSONObject.class;\r
- // load VESProcessors class at runtime\r
- Class cls = Class.forName("org.onap.dcae.commonFunction.ConfigProcessors");\r
- Constructor constr = cls.getConstructor(paramJSONObject);\r
- Object obj = constr.newInstance(event);\r
-\r
- for (int j = 0; j < topLevel.length(); j++) {\r
- JSONObject filterObj = topLevel.getJSONObject(j).getJSONObject("filter");\r
- Method method = cls.getDeclaredMethod("isFilterMet", paramJSONObject);\r
- boolean filterMet = (boolean) method.invoke(obj, filterObj);\r
- if (filterMet) {\r
- final JSONArray processors = topLevel.getJSONObject(j).getJSONArray("processors");\r
-\r
- // call the processor method\r
- for (int i = 0; i < processors.length(); i++) {\r
- final JSONObject processorList = processors.getJSONObject(i);\r
- final String functionName = processorList.getString("functionName");\r
- final JSONObject args = processorList.getJSONObject("args");\r
- \r
-\r
- log.info(String.format("functionName==%s | args==%s", functionName, args));\r
- // reflect method call\r
- method = cls.getDeclaredMethod(functionName, paramJSONObject);\r
- method.invoke(obj, args);\r
- }\r
- }\r
- }\r
-\r
- } catch (Exception e) {\r
-\r
- log.error("EventProcessor Exception" + e.getMessage() + e + e.getCause());\r
- } finally {\r
- // close the file\r
- if (fr != null) {\r
- try {\r
- fr.close();\r
- } catch (IOException e) {\r
- log.error("Error closing file reader stream : " + e.toString());\r
- }\r
-\r
- }\r
- }\r
- }\r
- // Remove VESversion from event. This field is for internal use and must\r
- // be removed after use.\r
- if (event.has("VESversion"))\r
- event.remove("VESversion");\r
-\r
- log.debug("Modified event:" + event);\r
-\r
- }\r
+ private static final Logger log = LoggerFactory.getLogger(EventProcessor.class);\r
+ private static final String EVENT_LITERAL = "event";\r
+ private static final String COMMON_EVENT_HEADER = "commonEventHeader";\r
+ static final Type EVENT_LIST_TYPE = new TypeToken<List<Event>>() {\r
+ }.getType();\r
+ private final SimpleDateFormat dateFormat = new SimpleDateFormat("EEE, MM dd yyyy hh:mm:ss z");\r
+\r
+ private static HashMap<String, String[]> streamidHash = new HashMap<>();\r
+ JSONObject event;\r
+\r
+ public EventProcessor() {\r
+ dateFormat.setTimeZone(TimeZone.getTimeZone("GMT"));\r
+ String[] list = CommonStartup.streamid.split("\\|");\r
+ for (String aList : list) {\r
+ String domain = aList.split("=")[0];\r
+\r
+ String[] streamIdList = aList.substring(aList.indexOf('=') + 1).split(",");\r
+\r
+ log.debug(String.format("Domain: %s streamIdList:%s", domain, Arrays.toString(streamIdList)));\r
+ streamidHash.put(domain, streamIdList);\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public void run() {\r
+\r
+ try {\r
+\r
+ event = CommonStartup.fProcessingInputQueue.take();\r
+\r
+ while (event != null) {\r
+ // As long as the producer is running we remove elements from\r
+ // the queue.\r
+ log.info("QueueSize:" + CommonStartup.fProcessingInputQueue.size() + "\tEventProcessor\tRemoving element: " + event);\r
+\r
+ String uuid = event.get("VESuniqueId").toString();\r
+ LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);\r
+ localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());\r
+\r
+ String domain = event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER).getString("domain");\r
+ log.debug("event.VESuniqueId" + event.get("VESuniqueId") + "event.commonEventHeader.domain:" + domain);\r
+ String[] streamIdList = streamidHash.get(domain);\r
+ log.debug("streamIdList:" + Arrays.toString(streamIdList));\r
+\r
+ if (streamIdList.length == 0) {\r
+ log.error("No StreamID defined for publish - Message dropped" + event);\r
+ } else {\r
+ sendEventsToStreams(streamIdList);\r
+ }\r
+ log.debug("Message published" + event);\r
+ event = CommonStartup.fProcessingInputQueue.take();\r
+\r
+ }\r
+ } catch (InterruptedException e) {\r
+ log.error("EventProcessor InterruptedException" + e.getMessage());\r
+ Thread.currentThread().interrupt();\r
+ }\r
+\r
+ }\r
+\r
+ public void overrideEvent() {\r
+ // Set collector timestamp in event payload before publish\r
+ addCurrentTimeToEvent(event);\r
+\r
+ if (CommonStartup.eventTransformFlag == 1) {\r
+ // read the mapping json file\r
+ try (FileReader fr = new FileReader("./etc/eventTransform.json")) {\r
+ log.info("parse eventTransform.json");\r
+ List<Event> events = new Gson().fromJson(fr, EVENT_LIST_TYPE);\r
+ parseEventsJson(events, new ConfigProcessorAdapter(new ConfigProcessors(event)));\r
+ } catch (IOException e) {\r
+ log.error("Couldn't find file ./etc/eventTransform.json" + e.toString());\r
+ }\r
+ }\r
+ // Remove VESversion from event. This field is for internal use and must\r
+ // be removed after use.\r
+ if (event.has("VESversion"))\r
+ event.remove("VESversion");\r
+\r
+ log.debug("Modified event:" + event);\r
+\r
+ }\r
+\r
+ private void sendEventsToStreams(String[] streamIdList) {\r
+ for (String aStreamIdList : streamIdList) {\r
+ log.info("Invoking publisher for streamId:" + aStreamIdList);\r
+ this.overrideEvent();\r
+ EventPublisherHash.getInstance().sendEvent(event, aStreamIdList);\r
+\r
+ }\r
+ }\r
+\r
+ private void addCurrentTimeToEvent(JSONObject event) {\r
+ final Date currentTime = new Date();\r
+ JSONObject collectorTimeStamp = new JSONObject().put("collectorTimeStamp", dateFormat.format(currentTime));\r
+ JSONObject commonEventHeaderkey = event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER);\r
+ commonEventHeaderkey.put("internalHeaderFields", collectorTimeStamp);\r
+ event.getJSONObject(EVENT_LITERAL).put(COMMON_EVENT_HEADER, commonEventHeaderkey);\r
+ }\r
+\r
+ void parseEventsJson(List<Event> eventsTransform, ConfigProcessorAdapter configProcessorAdapter) {\r
+\r
+ // load VESProcessors class at runtime\r
+ for (Event eventTransform : eventsTransform) {\r
+ JSONObject filterObj = new JSONObject(eventTransform.filter.toString());\r
+ if (configProcessorAdapter.isFilterMet(filterObj)) {\r
+ callProcessorsMethod(configProcessorAdapter, eventTransform.processors);\r
+ }\r
+ }\r
+ }\r
+\r
+ private void callProcessorsMethod(ConfigProcessorAdapter configProcessorAdapter, List<Processor> processors) {\r
+ // call the processor method\r
+ for (Processor processor : processors) {\r
+ final String functionName = processor.functionName;\r
+ final JSONObject args = new JSONObject(processor.args.toString());\r
+\r
+ log.info(String.format("functionName==%s | args==%s", functionName, args));\r
+ // reflect method call\r
+ try {\r
+ configProcessorAdapter.runConfigProcessorFunctionByName(functionName, args);\r
+ } catch (ReflectiveOperationException e) {\r
+ log.error("EventProcessor Exception" + e.getMessage() + e + e.getCause());\r
+ }\r
+ }\r
+ }\r
+\r
+ static class ConfigProcessorAdapter {\r
+ private final ConfigProcessors configProcessors;\r
+\r
+ ConfigProcessorAdapter(ConfigProcessors configProcessors) {\r
+ this.configProcessors = configProcessors;\r
+ }\r
+\r
+ boolean isFilterMet(JSONObject parameter) {\r
+ return configProcessors.isFilterMet(parameter);\r
+ }\r
+\r
+ void runConfigProcessorFunctionByName(String functionName, JSONObject parameter) throws ReflectiveOperationException {\r
+ Method method = configProcessors.getClass().getDeclaredMethod(functionName, parameter.getClass());\r
+ method.invoke(configProcessors, parameter);\r
+ }\r
+ }\r
}\r
+\r
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * Copyright (C) 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.onap.dcae.commonFunction;
+
+import com.google.gson.JsonObject;
+
+class Processor {
+ final String functionName;
+ final JsonObject args;
+
+ Processor(String functionName, JsonObject args) {
+ this.functionName = functionName;
+ this.args = args;
+ }
+}
--- /dev/null
+//
+// ================================================================================
+// Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
+// Copyright (c) 2018 Nokia. 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.onap.dcae.commonFunction;
+
+import org.json.JSONObject;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ConfigProcessorAdapterTest {
+
+ @Mock
+ private ConfigProcessors configProcessors;
+
+ @InjectMocks
+ private EventProcessor.ConfigProcessorAdapter configProcessorAdapter;
+
+
+ @Test
+ public void shouldCallIsFilterMetOnAdapter() throws Exception {
+ //given
+ JSONObject parameter = new JSONObject();
+ when(configProcessors.isFilterMet(parameter)).thenReturn(true);
+ //when
+ boolean actualReturn = configProcessorAdapter.isFilterMet(parameter);
+ //then
+ assertTrue(actualReturn);
+ verify(configProcessors, times(1)).isFilterMet(parameter);
+ }
+
+ @Test
+ public void shouldCallGivenMethodFromConfigProcessor() throws Exception {
+ JSONObject parameter = new JSONObject();
+ String exampleFunction = "concatenateValue";
+ //when
+ configProcessorAdapter.runConfigProcessorFunctionByName(exampleFunction, parameter);
+ //then
+ verify(configProcessors, times(1)).concatenateValue(parameter);
+ }
+
+}
\ No newline at end of file
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. 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.onap.dcae.commonFunction;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.google.gson.Gson;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.io.FileNotFoundException;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.onap.dcae.commonFunction.EventProcessor.EVENT_LIST_TYPE;
+
+public class EventProcessorTest {
+
+ private final String ev = "{\"event\": {\"commonEventHeader\": { \"reportingEntityName\": \"VM name will be provided by ECOMP\", \"startEpochMicrosec\": 1477012779802988,\"lastEpochMicrosec\": 1477012789802988,\"eventId\": \"83\",\"sourceName\": \"Dummy VM name - No Metadata available\",\"sequence\": 83,\"priority\": \"Normal\",\"functionalRole\": \"vFirewall\",\"domain\": \"measurementsForVfScaling\",\"reportingEntityId\": \"VM UUID will be provided by ECOMP\",\"sourceId\": \"Dummy VM UUID - No Metadata available\",\"version\": 1.1},\"measurementsForVfScalingFields\": {\"measurementInterval\": 10,\"measurementsForVfScalingVersion\": 1.1,\"vNicUsageArray\": [{\"multicastPacketsIn\": 0,\"bytesIn\": 3896,\"unicastPacketsIn\": 0, \"multicastPacketsOut\": 0,\"broadcastPacketsOut\": 0, \"packetsOut\": 28,\"bytesOut\": 12178,\"broadcastPacketsIn\": 0,\"packetsIn\": 58,\"unicastPacketsOut\": 0,\"vNicIdentifier\": \"eth0\"}]}}}";
+ private String testinput = "src/test/resources/testDmaapConfig_ip.json";
+
+ @Before
+ public void setUp() throws Exception {
+ CommonStartup.streamid = "fault=sec_fault|syslog=sec_syslog|heartbeat=sec_heartbeat|measurementsForVfScaling=sec_measurement|mobileFlow=sec_mobileflow|other=sec_other|stateChange=sec_statechange|thresholdCrossingAlert=sec_thresholdCrossingAlert|voiceQuality=ves_voicequality|sipSignaling=ves_sipsignaling";
+ CommonStartup.eventTransformFlag = 1;
+ }
+
+ @Test
+ public void testLoad() {
+ //given
+ EventProcessor ec = new EventProcessor();
+ ec.event = new org.json.JSONObject(ev);
+ //when
+ ec.overrideEvent();
+
+ //then
+ Boolean hasSourceNameNode = ec.event.getJSONObject("event").getJSONObject("commonEventHeader").has("sourceName");
+ assertTrue(hasSourceNameNode);
+ }
+
+ @Test
+ public void shouldParseJsonEvents() throws FileNotFoundException, ReflectiveOperationException {
+ //given
+ EventProcessor eventProcessor = new EventProcessor();
+ String event_json = "[{ \"filter\": {\"event.commonEventHeader.domain\":\"heartbeat\",\"VESversion\":\"v4\"},\"processors\":[" +
+ "{\"functionName\": \"concatenateValue\",\"args\":{\"field\":\"event.commonEventHeader.eventName\",\"concatenate\": [\"$event.commonEventHeader.domain\",\"$event.commonEventHeader.eventType\",\"$event.faultFields.alarmCondition\"], \"delimiter\":\"_\"}}" +
+ ",{\"functionName\": \"addAttribute\",\"args\":{\"field\": \"event.heartbeatFields.heartbeatFieldsVersion\",\"value\": \"1.0\",\"fieldType\": \"number\"}}" +
+ ",{\"functionName\": \"map\",\"args\":{\"field\": \"event.commonEventHeader.nfNamingCode\",\"oldField\": \"event.commonEventHeader.functionalRole\"}}]}]";
+ List<Event> events = new Gson().fromJson(event_json, EVENT_LIST_TYPE);
+ EventProcessor.ConfigProcessorAdapter configProcessorAdapter = mock(EventProcessor.ConfigProcessorAdapter.class);
+
+ when(configProcessorAdapter.isFilterMet(any(JSONObject.class))).thenReturn(true);
+ ArgumentCaptor<String> stringArgumentCaptor = ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor<JSONObject> jsonObjectArgumentCaptor = ArgumentCaptor.forClass(JSONObject.class);
+ //when
+ eventProcessor.parseEventsJson(events, configProcessorAdapter);
+
+ //then
+ verify(configProcessorAdapter, times(3)).runConfigProcessorFunctionByName(stringArgumentCaptor.capture(), jsonObjectArgumentCaptor.capture());
+ assertThat(stringArgumentCaptor.getAllValues()).contains("concatenateValue", "addAttribute", "map");
+ }
+
+ @Test
+ public void shouldCreateDmaapPublisher() {
+
+ //given
+ EventPublisherHash eph = EventPublisherHash.getInstance();
+ EventProcessor ec = new EventProcessor();
+ ec.event = new org.json.JSONObject(ev);
+ CommonStartup.cambriaConfigFile = "src/test/resources/testDmaapConfig_ip.json";
+
+ //when
+ CambriaBatchingPublisher pub = eph.getDmaapPublisher("sec_fault_ueb");
+
+ //then
+ assertNotNull(pub);
+ }
+
+ @Test
+ public void shouldSendEventWithNoError() {
+
+ EventPublisherHash eph = EventPublisherHash.getInstance();
+ EventProcessor eventProcessor = new EventProcessor();
+ eventProcessor.event = new org.json.JSONObject(ev);
+ CommonStartup.cambriaConfigFile = "src/test/resources/testDmaapConfig_ip.json";
+
+ //when
+ eph.sendEvent(eventProcessor.event, "sec_fault_ueb");
+ }
+}
+
+++ /dev/null
-/*-\r
- * ============LICENSE_START=======================================================\r
- * org.onap.dcaegen2.collectors.ves\r
- * ================================================================================\r
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.\r
- * Copyright (C) 2018 Nokia. All rights reserved.\r
- * ================================================================================\r
- * Licensed under the Apache License, Version 2.0 (the "License");\r
- * you may not use this file except in compliance with the License.\r
- * You may obtain a copy of the License at\r
- *\r
- * http://www.apache.org/licenses/LICENSE-2.0\r
- *\r
- * Unless required by applicable law or agreed to in writing, software\r
- * distributed under the License is distributed on an "AS IS" BASIS,\r
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * See the License for the specific language governing permissions and\r
- * limitations under the License.\r
- * ============LICENSE_END=========================================================\r
- */\r
-package org.onap.dcae.vestest;\r
-\r
-import static org.junit.Assert.assertEquals;\r
-import static org.junit.Assert.assertNotNull;\r
-import static org.junit.Assert.assertTrue;\r
-\r
-import org.junit.After;\r
-import org.junit.Before;\r
-import org.junit.Test;\r
-import org.junit.experimental.theories.suppliers.TestedOn;\r
-import org.onap.dcae.commonFunction.CommonStartup;\r
-import org.onap.dcae.commonFunction.DmaapPropertyReader;\r
-import org.onap.dcae.commonFunction.EventProcessor;\r
-import org.onap.dcae.commonFunction.EventPublisherHash;\r
-\r
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;\r
-\r
-public class TestEventProcessor {\r
-\r
- EventProcessor ec;\r
- String ev= "{\"event\": {\"commonEventHeader\": { \"reportingEntityName\": \"VM name will be provided by ECOMP\", \"startEpochMicrosec\": 1477012779802988,\"lastEpochMicrosec\": 1477012789802988,\"eventId\": \"83\",\"sourceName\": \"Dummy VM name - No Metadata available\",\"sequence\": 83,\"priority\": \"Normal\",\"functionalRole\": \"vFirewall\",\"domain\": \"measurementsForVfScaling\",\"reportingEntityId\": \"VM UUID will be provided by ECOMP\",\"sourceId\": \"Dummy VM UUID - No Metadata available\",\"version\": 1.1},\"measurementsForVfScalingFields\": {\"measurementInterval\": 10,\"measurementsForVfScalingVersion\": 1.1,\"vNicUsageArray\": [{\"multicastPacketsIn\": 0,\"bytesIn\": 3896,\"unicastPacketsIn\": 0, \"multicastPacketsOut\": 0,\"broadcastPacketsOut\": 0, \"packetsOut\": 28,\"bytesOut\": 12178,\"broadcastPacketsIn\": 0,\"packetsIn\": 58,\"unicastPacketsOut\": 0,\"vNicIdentifier\": \"eth0\"}]}}}";\r
- String testinput;\r
-\r
- @Before\r
- public void setUp() throws Exception {\r
- CommonStartup.streamid="fault=sec_fault|syslog=sec_syslog|heartbeat=sec_heartbeat|measurementsForVfScaling=sec_measurement|mobileFlow=sec_mobileflow|other=sec_other|stateChange=sec_statechange|thresholdCrossingAlert=sec_thresholdCrossingAlert|voiceQuality=ves_voicequality|sipSignaling=ves_sipsignaling";\r
- CommonStartup.eventTransformFlag = 1;\r
- testinput = "src/test/resources/testDmaapConfig_ip.json";\r
-\r
- }\r
-\r
- @After\r
- public void tearDown() throws Exception {\r
- }\r
-\r
- @Test\r
- public void testLoad() {\r
-\r
-\r
- EventProcessor ec = new EventProcessor();\r
-\r
- ec.event=new org.json.JSONObject(ev);\r
-\r
- ec.overrideEvent();\r
- //event.commonEventHeader.sourceName\r
- Boolean flag = ec.event.getJSONObject("event").getJSONObject("commonEventHeader").has("sourceName");\r
- assertEquals(true, flag);\r
- }\r
-\r
-\r
- @Test\r
- public void testpublisherhashclass() {\r
-\r
- DmaapPropertyReader dr = null;\r
- EventPublisherHash eph = null;\r
-\r
- Boolean flag = false;\r
- dr = new DmaapPropertyReader(testinput);\r
- eph = EventPublisherHash.getInstance();\r
-\r
-\r
- if (eph.equals(null))\r
- {\r
- flag = false;\r
- }\r
- else\r
- {\r
- flag = true;\r
- }\r
- assertEquals(true, flag);\r
-\r
-\r
- }\r
-\r
- @Test\r
- public void testpublisherhashclassload() {\r
-\r
- DmaapPropertyReader dr;\r
- EventPublisherHash eph = null;\r
-\r
- dr = new DmaapPropertyReader(testinput);\r
- eph = EventPublisherHash.getInstance();\r
- EventProcessor ec = new EventProcessor();\r
- ec.event=new org.json.JSONObject(ev);\r
- CommonStartup.cambriaConfigFile="src/test/resources/testDmaapConfig_ip.json";\r
-\r
- CambriaBatchingPublisher pub = eph.getDmaapPublisher("sec_fault_ueb");\r
-\r
- assertNotNull(pub);\r
- }\r
-\r
- @Test\r
- public void testpublisherhashSend() {\r
-\r
- DmaapPropertyReader dr;\r
- EventPublisherHash eph = null;\r
-\r
- Boolean flag = true;\r
- dr = new DmaapPropertyReader(testinput);\r
- eph = EventPublisherHash.getInstance();\r
-\r
-\r
- EventProcessor ec = new EventProcessor();\r
- ec.event=new org.json.JSONObject(ev);\r
- CommonStartup.cambriaConfigFile="src/test/resources/testDmaapConfig_ip.json";\r
- eph.sendEvent(ec.event, "sec_fault_ueb");\r
-\r
- assertEquals(true, flag);\r
-\r
- }\r
-}\r
-\r
Code Review / dcaegen2 / collectors / ves.git / commitdiff