Integrate split and filter in main App

Change-Id: I9da0852072409854118e46aba63c491bdd53fd28
Issue-ID: DCAEGEN2-1038
Signed-off-by: emartin <ephraim.martin@est.tech>

11 files changed, 382 insertions, 54 deletions

diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
index 6ebc61c..9abe086 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
@@ -33,12 +33,17 @@ import org.onap.dcaegen2.services.pmmapper.exceptions.CBSConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
+import org.onap.dcaegen2.services.pmmapper.exceptions.ProcessEventException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.TooManyTriesException;
 import org.onap.dcaegen2.services.pmmapper.filtering.MetadataFilter;
+import org.onap.dcaegen2.services.pmmapper.filtering.MeasFilterHandler;
 import org.onap.dcaegen2.services.pmmapper.mapping.Mapper;
 import org.onap.dcaegen2.services.pmmapper.model.Event;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
 import org.onap.dcaegen2.services.pmmapper.healthcheck.HealthCheckHandler;
+import org.onap.dcaegen2.services.pmmapper.utils.DataRouterUtils;
+import org.onap.dcaegen2.services.pmmapper.utils.MeasConverter;
+import org.onap.dcaegen2.services.pmmapper.utils.MeasSplitter;
 import org.onap.dcaegen2.services.pmmapper.utils.XMLValidator;
 import org.onap.logging.ref.slf4j.ONAPLogAdapter;
 import org.slf4j.LoggerFactory;
@@ -50,6 +55,7 @@ import reactor.core.scheduler.Schedulers;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
+import java.util.List;
 
 public class App {
     private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(App.class));
@@ -62,8 +68,12 @@ public class App {
         HealthCheckHandler healthCheckHandler = new HealthCheckHandler();
         MapperConfig mapperConfig = new ConfigHandler().getMapperConfig();
         MetadataFilter metadataFilter = new MetadataFilter(mapperConfig);
+        MeasConverter measConverter = new MeasConverter();
+        MeasFilterHandler filterHandler = new MeasFilterHandler(measConverter);
         Mapper mapper = new Mapper(mappingTemplate);
+        MeasSplitter splitter = new MeasSplitter(measConverter);
         XMLValidator validator = new XMLValidator(xmlSchema);
+
         flux.onBackpressureDrop(App::handleBackPressure)
                 .doOnNext(App::receiveRequest)
                 .limitRate(1)
@@ -71,9 +81,13 @@ public class App {
                 .runOn(Schedulers.newParallel(""), 1)
                 .doOnNext(event -> MDC.setContextMap(event.getMdc()))
                 .filter(metadataFilter::filter)
+                .filter(filterHandler::filterByFileType)
                 .filter(validator::validate)
-                .map(mapper::map)
-                .subscribe(event -> logger.unwrap().info("Event Processed"));
+                .concatMap(event -> App.split(splitter,event, mapperConfig))
+                .filter(events -> App.filter(filterHandler, events, mapperConfig))
+                .concatMap(events -> App.map(mapper, events, mapperConfig))
+                .subscribe(events -> logger.unwrap().info("Event Processed"));
+
         DataRouterSubscriber dataRouterSubscriber = new DataRouterSubscriber(fluxSink::next, mapperConfig);
         dataRouterSubscriber.start();
         ArrayList<Configurable> configurables = new ArrayList<>();
@@ -89,6 +103,53 @@ public class App {
                 .build().start();
     }
 
+    public static boolean filter(MeasFilterHandler filterHandler, List<Event> events, MapperConfig config) {
+        Event event = events.get(0);
+        boolean hasMatchingFilter = false;
+        try {
+            hasMatchingFilter = filterHandler.filterByMeasType(events);
+            if(!hasMatchingFilter) {
+                sendEventProcessed(config,event);
+            }
+        } catch (Exception exception) {
+            logger.unwrap().error(exception.getMessage(),exception);
+            sendEventProcessed(config,event);
+        }
+        return hasMatchingFilter;
+    }
+
+    public static Flux<List<Event>> map(Mapper mapper, List<Event> events, MapperConfig config) {
+        List<Event> mappedEvents  = new ArrayList<>();
+        try {
+            mappedEvents = mapper.mapEvents(events);
+        } catch (Exception exception) {
+            logger.unwrap().error(exception.getMessage(),exception);
+            sendEventProcessed(config,events.get(0));
+            return Flux.<List<Event>>empty();
+        }
+        return Flux.just(mappedEvents);
+    }
+
+    public static Flux<List<Event>> split(MeasSplitter splitter, Event event, MapperConfig config) {
+        List<Event> splitEvents  = new ArrayList<>();
+        try {
+            splitEvents = splitter.split(event);
+        } catch (Exception exception) {
+            logger.unwrap().error(exception.getMessage(),exception);
+            sendEventProcessed(config,event);
+            return Flux.<List<Event>>empty();
+        }
+        return Flux.just(splitEvents);
+    }
+
+    public static void sendEventProcessed(MapperConfig config, Event event) {
+      try {
+          DataRouterUtils.processEvent(config, event);
+      } catch (ProcessEventException exception) {
+          logger.unwrap().error("Process event failure", exception);
+      }
+    }
+
     /**
      * Takes the exchange from an event, responds with a 429 and un-dispatches the exchange.
      * @param event to be ignored.
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java
index a6017b6..7aec457 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java
@@ -24,8 +24,10 @@ import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+
 import org.onap.dcaegen2.services.pmmapper.model.Event;
 import org.onap.dcaegen2.services.pmmapper.model.MeasCollecFile;
 import org.onap.dcaegen2.services.pmmapper.model.MeasCollecFile.MeasData;
@@ -43,28 +45,30 @@ import org.slf4j.LoggerFactory;
  **/
 public class MeasFilterHandler {
     private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(MeasFilterHandler.class));
-    private Filter filter;
     private MeasConverter converter;
 
     public MeasFilterHandler(MeasConverter converter) {
         this.converter = converter;
     }
 
-    public void setFilter(Filter filter) {
-       this.filter = filter;
-    }
-
     /**
      * Filters each measInfo node for measTypes that match the given measTypes from filters.
      **/
     public boolean filterByMeasType(Event event) {
-      logger.unwrap().debug("Filtering the measurement file.");
-
+        Optional<Filter> filter = Optional.ofNullable(event.getFilter());
         MeasCollecFile measCollecFile = event.getMeasCollecFile();
-        if(filter.getMeasTypes().isEmpty() || measCollecFile.getMeasData().isEmpty()) {
+
+        if(hasNoFilters(filter)) {
+            logger.unwrap().info("Skipping filtering by measTypes as filter config does not contain measTypes.");
+            return true;
+        }
+
+        if(measCollecFile.getMeasData().isEmpty()) {
+            logger.unwrap().info("Measurement file will not be processed further as MeasData is empty.");
             return false;
         }
 
+        logger.unwrap().info("Filtering the measurement file by measTypes.");
         MeasData measData = measCollecFile.getMeasData().get(0);
         List<MeasInfo> measInfos = measData.getMeasInfo();
         List<MeasInfo> filteredMeasInfos = new ArrayList<>();
@@ -73,23 +77,62 @@ public class MeasFilterHandler {
             MeasInfo currentMeasInfo = measInfos.get(i);
             List<String> measTypesNode = currentMeasInfo.getMeasTypes();
             if(!measTypesNode.isEmpty()) {
-                setMeasInfosFromMeasTypes(currentMeasInfo,filteredMeasInfos);
+                setMeasInfosFromMeasTypes(currentMeasInfo,filteredMeasInfos, filter.get());
             }else {
-                setMeasInfoFromMeasType(currentMeasInfo,filteredMeasInfos);
+                setMeasInfoFromMeasType(currentMeasInfo,filteredMeasInfos, filter.get());
             }
         }
 
         if (filteredMeasInfos.isEmpty()) {
+            logger.unwrap().info("No filter match from the current measurement file.");
             return false;
         }
-
         measData.setMeasInfo(filteredMeasInfos);
         String filteredXMl = converter.convert(measCollecFile);
         event.setBody(filteredXMl);
+        logger.unwrap().info("Successfully filtered the measurement by measTypes.");
+        return true;
+    }
+
+    /**
+     * Filters each measInfo node in the list for measTypes that match the given measTypes from filters.
+     **/
+    public boolean filterByMeasType(List<Event> events) {
+        boolean hasMatchAnyFilter = false;
+        for (int i = 0; i < events.size(); i++) {
+            Event currentEvent = events.get(i);
+            boolean hasMatchingFilter = filterByMeasType(currentEvent);
+            if (hasMatchingFilter) {
+                hasMatchAnyFilter = true;
+            } else {
+                events.remove(events.get(i));
+            }
+        }
+
+        if (!hasMatchAnyFilter) {
+            logger.unwrap().info("No filter match from all measurement files.");
+            return false;
+        }
+
         return true;
     }
 
-    private void setMeasInfoFromMeasType(MeasInfo currentMeasInfo,  List<MeasInfo> filteredMeasInfos) {
+    private boolean hasNoFilters(Optional<Filter> filter) {
+        return !filter.isPresent() || filter.get().getMeasTypes().isEmpty();
+    }
+
+
+    /**
+     * Filters the measurement by file type. Measurement files starting with A or C are valid.
+     **/
+    public boolean filterByFileType(Event event) {
+        logger.unwrap().debug("Filtering the measurement by file type.");
+        String requestPath  = event.getHttpServerExchange().getRequestPath();
+        String fileName = requestPath.substring(requestPath.lastIndexOf('/')+1);
+        return (fileName.startsWith("C") || fileName.startsWith("A"));
+    }
+
+    private void setMeasInfoFromMeasType(MeasInfo currentMeasInfo,  List<MeasInfo> filteredMeasInfos, Filter filter) {
         MeasValue currentMeasValue = currentMeasInfo.getMeasValue()
                 .get(0);
         List<R> measResultsRNodes = currentMeasValue.getR();
@@ -114,7 +157,7 @@ public class MeasFilterHandler {
 
     }
 
-    private void setMeasInfosFromMeasTypes(MeasInfo currentMeasInfo, List<MeasInfo> filteredMeasInfos) {
+    private void setMeasInfosFromMeasTypes(MeasInfo currentMeasInfo, List<MeasInfo> filteredMeasInfos, Filter filter) {
         MeasValue currentMeasValue = currentMeasInfo.getMeasValue()
                 .get(0);
         List<String> measTypesNode = currentMeasInfo.getMeasTypes();
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/mapping/Mapper.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/mapping/Mapper.java
index 0ad26e3..44bbc27 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/mapping/Mapper.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/mapping/Mapper.java
@@ -42,6 +42,7 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
@@ -62,6 +63,11 @@ public class Mapper {
         }
     }
 
+    public List<Event> mapEvents(List<Event> events) {
+        events.forEach(event -> event.setVes(this.map(event)));
+        return events;
+    }
+
     public String map(@NonNull Event event) {
         logger.unwrap().info("Mapping event");
         NodeModel pmNodeModel;
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/Event.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/Event.java
index fa41142..9eadb8a 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/Event.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/Event.java
@@ -45,4 +45,6 @@ public class Event {
     private MeasCollecFile measCollecFile;
 
     private Filter filter;
+
+    private String ves;
 }
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DataRouterUtils.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DataRouterUtils.java
index 9525ec7..f30fb96 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DataRouterUtils.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DataRouterUtils.java
@@ -42,7 +42,7 @@ public class DataRouterUtils {
         logger.unwrap().info("Sending processed to DataRouter");
         String baseDelete = config.getDmaapDRDeleteEndpoint();
         String subscriberIdentity = config.getSubscriberIdentity();
-        String delete = String.format("https://%s/%s/%s", baseDelete, subscriberIdentity, event.getPublishIdentity());
+        String delete = String.format("%s/%s/%s", baseDelete, subscriberIdentity, event.getPublishIdentity());
         try {
             return new RequestSender().send("DELETE", delete);
         } catch (Exception exception) {
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitter.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitter.java
index 009e454..1821803 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitter.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitter.java
@@ -22,18 +22,16 @@ package org.onap.dcaegen2.services.pmmapper.utils;
 

 import java.util.Arrays;

 import java.util.List;

-import java.util.Map;

 import java.util.stream.Collectors;

 

 import org.onap.dcaegen2.services.pmmapper.model.Event;

-import org.onap.dcaegen2.services.pmmapper.model.EventMetadata;

+import java.util.NoSuchElementException;

+

 import org.onap.dcaegen2.services.pmmapper.model.MeasCollecFile;

 import org.onap.dcaegen2.services.pmmapper.model.MeasCollecFile.MeasData;

 import org.onap.logging.ref.slf4j.ONAPLogAdapter;

 import org.slf4j.LoggerFactory;

 

-import io.undertow.server.HttpServerExchange;

-

 /**

  * Splits the MeasCollecFile based on MeasData.

  **/

@@ -44,11 +42,17 @@ public class MeasSplitter {
     public MeasSplitter(MeasConverter converter) {

         this.converter = converter;

     }

-

+
+    /**

+     * Splits the MeasCollecFile to multiple MeasCollecFile based on the number of MeasData

+     **/

     public List<Event> split(Event event) {

-        logger.unwrap().debug("Splitting 3GPP xml MeasData to MeasCollecFile");

+        logger.unwrap().debug("Splitting 3GPP xml MeasData to individual MeasCollecFile");

         MeasCollecFile currentMeasurement = converter.convert(event.getBody());

-

+        event.setMeasCollecFile(currentMeasurement);

+        if(currentMeasurement.getMeasData().isEmpty()) {

+            throw new NoSuchElementException("MeasData is empty.");

+        }

         return currentMeasurement.getMeasData().stream().map( measData -> {

             Event newEvent  = generateNewEvent(event);

             MeasCollecFile newMeasCollec = generateNewMeasCollec(newEvent,measData);

@@ -70,6 +74,7 @@ public class MeasSplitter {
                 event.getBody(), event.getMetadata(), event.getMdc(),

                 event.getPublishIdentity());

         modifiedEvent.setMeasCollecFile(event.getMeasCollecFile());

+        modifiedEvent.setFilter(event.getFilter());

         return modifiedEvent;

     }

 }

diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
index 82650f3..e82d90f 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
@@ -20,22 +20,58 @@
 
 package org.onap.dcaegen2.services.pmmapper;
 
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockserver.integration.ClientAndServer.startClientAndServer;
+import static org.mockserver.model.HttpResponse.response;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.List;
 
 import io.undertow.util.StatusCodes;
+import reactor.core.publisher.Flux;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockserver.client.server.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.HttpRequest;
+import org.onap.dcaegen2.services.pmmapper.filtering.MeasFilterHandler;
+import org.onap.dcaegen2.services.pmmapper.mapping.Mapper;
 import org.onap.dcaegen2.services.pmmapper.model.Event;
 import org.onap.dcaegen2.services.pmmapper.model.EventMetadata;
+import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
+import org.onap.dcaegen2.services.pmmapper.utils.MeasConverter;
+import org.onap.dcaegen2.services.pmmapper.utils.MeasSplitter;
 
 
 @ExtendWith(MockitoExtension.class)
 class AppTest {
 
+    static ClientAndServer mockServer;
+    static MockServerClient client;
+
+    @BeforeAll
+    public static void setup() {
+        mockServer =  startClientAndServer(1080);
+        client = new MockServerClient("127.0.0.1", 1080);
+    }
+
+    @AfterAll
+    public static void teardown() {
+        mockServer.stop();
+    }
+
     @Test
     void testHandleBackPressureNullValue() {
         assertThrows(NullPointerException.class, () -> App.handleBackPressure(null));
@@ -62,5 +98,65 @@ class AppTest {
         verify(event.getHttpServerExchange(), times(1)).unDispatch();
     }
 
+    @Test
+    public void testFilter_success() {
+        Event mockEvent = Mockito.mock(Event.class);
+        List<Event> mockEvents = Arrays.asList(mockEvent);
+        MapperConfig mockConfig = Mockito.mock(MapperConfig.class);
+        boolean result = App.filter(new MeasFilterHandler(new MeasConverter()), mockEvents, mockConfig);
+        assertTrue(result);
+    }
+
+    @Test
+    public void testFilter_throwException() {
+        HttpRequest req = HttpRequest.request();
+        client.when(req).respond( response().withStatusCode(200));
+
+        Event mockEvent = Mockito.mock(Event.class);
+        List<Event> mockEvents = Arrays.asList(mockEvent);
+        MeasFilterHandler mockFilter = Mockito.mock(MeasFilterHandler.class);
+        MapperConfig mockConfig = Mockito.mock(MapperConfig.class);
+
+        Mockito.when(mockConfig.getDmaapDRDeleteEndpoint()).thenReturn("http://127.0.0.1:1080");
+        Mockito.when(mockConfig.getSubscriberIdentity()).thenReturn("sid");
+        Mockito.when(mockEvent.getPublishIdentity()).thenReturn("pid");
+        Mockito.when(mockFilter.filterByMeasType(mockEvent)).thenThrow(RuntimeException.class);
+
+        boolean x = App.filter(mockFilter, mockEvents, mockConfig);
+        assertFalse(x);
+
+        client.clear(req);
+    }
+
+    @Test
+    public void testSplit_empty_success() {
+        Event mockEvent = Mockito.mock(Event.class);
+        MapperConfig mockConfig = Mockito.mock(MapperConfig.class);
+        Flux<List<Event>> splitResult = App.split(new MeasSplitter(new MeasConverter()), mockEvent, mockConfig);
+        splitResult.equals(Flux.<List<Event>>empty());
+    }
+
+    @Test
+    public void testSplit_success() {
+        Event mockEvent = Mockito.mock(Event.class);
+        List<Event> mockEvents = Arrays.asList(mockEvent,mockEvent);
+        MapperConfig mockConfig = Mockito.mock(MapperConfig.class);
+        MeasSplitter mockSplitter  = Mockito.mock(MeasSplitter.class);
+        Mockito.when(mockSplitter.split(mockEvent)).thenReturn(mockEvents);
+
+        Flux<List<Event>> splitResult = App.split(mockSplitter, mockEvent, mockConfig);
+
+        splitResult.equals(Flux.just(mockEvents));
+    }
+
+    @Test
+    public void testMapping_empty_success() {
+        Event mockEvent = Mockito.mock(Event.class);
+        List<Event> mockEvents = Arrays.asList(mockEvent);
+        MapperConfig mockConfig = Mockito.mock(MapperConfig.class);
+        Path mappingTemplate = Paths.get("src/main/resources/mapping.ftl");
+        Flux<List<Event>> mappingResult = App.map(new Mapper(mappingTemplate), mockEvents, mockConfig);
+        mappingResult.equals(Flux.<List<Event>>empty());
+    }
 
 }
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandlerTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandlerTest.java
index 8b1f8aa..436984e 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandlerTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandlerTest.java
@@ -22,10 +22,14 @@ package org.onap.dcaegen2.services.pmmapper.filtering;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -51,11 +55,9 @@ import utils.EventUtils;
 @ExtendWith(MockitoExtension.class)
 class MeasFilterHandlerTest {
 
-    private static MeasFilterConfig filterConfig;
     private static final String baseDir = "src/test/resources/filter_test/";
     private static final Path dataDirectory = Paths.get("src/test/resources/mapper_test/mapping_data/");
-    private static List<String> counters = Arrays.asList("a", "b");
-    private static  MeasConverter converter = new MeasConverter();
+    private static MeasConverter converter = new MeasConverter();
     private MeasFilterHandler objUnderTest;
     @Mock
     private HttpServerExchange exchange;
@@ -64,23 +66,14 @@ class MeasFilterHandlerTest {
 
     @BeforeEach
     void setup() {
-        filterConfig = new MeasFilterConfig();
-        Filter filter = filterConfig.new Filter();
-        filter.setDictionaryVersion("");
-        filter.setMeasTypes(counters);
-        filterConfig.setFilters(Arrays.asList(filter));
         objUnderTest = new MeasFilterHandler(new MeasConverter());
-        objUnderTest.setFilter(filterConfig.getFilters()
-                .get(0));
     }
 
     @Test
     void measTypes_byCommaSeparation() throws IOException {
         String inputPath = baseDir + "meas_results";
-        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));
         String expected = EventUtils.fileContentsToString(Paths.get(inputPath + "_filtered.xml"));
-        Event event = new Event(exchange, inputXml, metaData, new HashMap<String, String>(), "");
-        event.setMeasCollecFile(converter.convert(inputXml));
+        Event event = generateEvent(inputPath, generateValidFilter());
 
         objUnderTest.filterByMeasType(event);
 
@@ -91,12 +84,11 @@ class MeasFilterHandlerTest {
     @Test
     void measType_byID() throws IOException {
         String inputPath = baseDir + "meas_type_and_r";
-        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));
         String filteredString = EventUtils.fileContentsToString(Paths.get(inputPath + "_filtered.xml"));
-        Event event = new Event(exchange, inputXml, metaData, new HashMap<String, String>(), "");
-        event.setMeasCollecFile(converter.convert(inputXml));
+        Event event = generateEvent(inputPath, generateValidFilter());
         MeasCollecFile f = converter.convert(filteredString);
         String expected = converter.convert(f);
+
         objUnderTest.filterByMeasType(event);
 
         String actual = converter.convert(event.getMeasCollecFile());
@@ -104,26 +96,70 @@ class MeasFilterHandlerTest {
     }
 
     @Test
-    void no_Filters_match() {
+    void skip_mapping_when_no_Filters_match() {
         String inputPath = baseDir + "meas_results";
-        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));
-
-        Filter noMatchFilter = filterConfig.new Filter();
+        Filter noMatchFilter = new MeasFilterConfig().new Filter();
         noMatchFilter.setMeasTypes(Arrays.asList("nomatch1", "nomatch2"));
-        objUnderTest.setFilter(noMatchFilter);
+        Event event = generateEvent(inputPath, noMatchFilter);
+        List<Event> events = new ArrayList<>();
+        events.add(event);
+        events.add(event);
+
+        assertFalse(objUnderTest.filterByMeasType(event));
+        assertFalse(objUnderTest.filterByMeasType(events));
+    }
+
+    @Test
+    void remove_events_that_does_not_match_filter() {
+        String inputPath = baseDir + "meas_type_and_r_manyInfo";
+
+        Filter matchFilter = new MeasFilterConfig().new Filter();
+        matchFilter.setMeasTypes(Arrays.asList("a", "b"));
+        Event eventMatch = generateEvent(inputPath, matchFilter);
+        Filter noMatchFilter = new MeasFilterConfig().new Filter();
+        noMatchFilter.setMeasTypes(Arrays.asList("ad", "bs"));
+        Event eventNoMatch = generateEvent(inputPath, noMatchFilter);
+
+        List<Event> events = new ArrayList<>();
+        events.add(eventMatch);
+        events.add(eventNoMatch);
+        assertTrue(objUnderTest.filterByMeasType(events));
+        assertEquals(1, events.size());
+    }
+
+    @Test
+    void skip_mapping_when_MeasData_isEmpty() {
+        String inputPath = baseDir + "meas_results";
+        Event event = generateEvent(inputPath, generateValidFilter());
+        event.getMeasCollecFile().replaceMeasData(Arrays.asList());
 
-        Event event = new Event(exchange, inputXml, metaData, new HashMap<String, String>(), "");
-        event.setMeasCollecFile(converter.convert(inputXml));
         assertFalse(objUnderTest.filterByMeasType(event));
     }
 
     @Test
+    void skip_filtering_if_filter_or_meastypes_isEmpty() {
+        String inputPath = baseDir + "meas_results";
+
+        Filter emptyMeastypesFilter = new MeasFilterConfig().new Filter();
+        emptyMeastypesFilter.setMeasTypes(Arrays.asList());
+
+        Event event = generateEvent(inputPath, emptyMeastypesFilter);
+        MeasCollecFile originalMeasCollec = event.getMeasCollecFile();
+
+        assertTrue(objUnderTest.filterByMeasType(event));
+        assertEquals(originalMeasCollec,event.getMeasCollecFile());
+
+        event.setFilter(null);
+        assertTrue(objUnderTest.filterByMeasType(event));
+        assertEquals(originalMeasCollec,event.getMeasCollecFile());
+    }
+
+    @Test
     void multiple_measInfos_measResults() {
         String inputPath = baseDir + "meas_results_manyInfo";
-        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));
         String filteredString = EventUtils.fileContentsToString(Paths.get(inputPath + "_filtered.xml"));
-        Event event = new Event(exchange, inputXml, metaData, new HashMap<String, String>(), "");
-        event.setMeasCollecFile(converter.convert(inputXml));
+        Event event = generateEvent(inputPath, generateValidFilter());
+
         MeasCollecFile f = converter.convert(filteredString);
         String expected = converter.convert(f);
         objUnderTest.filterByMeasType(event);
@@ -135,10 +171,9 @@ class MeasFilterHandlerTest {
     @Test
     void multiple_measInfos_measTypeAndR() {
         String inputPath = baseDir + "meas_type_and_r_manyInfo";
-        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));
         String filteredString = EventUtils.fileContentsToString(Paths.get(inputPath + "_filtered.xml"));
-        Event event = new Event(exchange, inputXml, metaData, new HashMap<String, String>(), "");
-        event.setMeasCollecFile(converter.convert(inputXml));
+        Event event = generateEvent(inputPath, generateValidFilter());
+
         MeasCollecFile f = converter.convert(filteredString);
         String expected = converter.convert(f);
         objUnderTest.filterByMeasType(event);
@@ -147,21 +182,58 @@ class MeasFilterHandlerTest {
         assertEquals(expected, actual);
     }
 
+    @Test
+    void valid_fileType() {
+        Event event = mock(Event.class);
+        when(event.getHttpServerExchange()).thenReturn(exchange);
+        when(exchange.getRequestPath()).thenReturn("Apm.xml","Cpm.xml");
+        assertTrue(objUnderTest.filterByFileType(event));
+        assertTrue(objUnderTest.filterByFileType(event));
+    }
+
+    @Test
+    void invalid_fileType() {
+        Event event = mock(Event.class);
+        List<String> invalidFiletypes = Arrays.asList("Bpm.xml","Dpm.xml","asdf","bsdf");
+        when(event.getHttpServerExchange()).thenReturn(exchange);
+        when(exchange.getRequestPath())
+            .thenReturn(invalidFiletypes.toString());
+
+        invalidFiletypes.forEach(c -> {
+            assertFalse(objUnderTest.filterByFileType(event));
+        });
+    }
+
+
     @ParameterizedTest
     @MethodSource("getValidMeas")
     void applyFilterToValidMeasurements(Event testEvent) {
         objUnderTest.filterByMeasType(testEvent);
     }
 
+    private Event generateEvent(String inputPath, Filter filter) {
+        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));
+        Event event = new Event(exchange, inputXml, metaData, new HashMap<String, String>(), "");
+        event.setMeasCollecFile(converter.convert(inputXml));
+        event.setFilter(filter);
+        return event;
+    }
+
+    private Filter generateValidFilter() {
+        Filter filter;
+        filter = new MeasFilterConfig().new Filter();
+        filter.setDictionaryVersion("1.0");
+        filter.setMeasTypes(Arrays.asList("a", "b"));
+        return filter;
+    }
+
     static List<Event> getValidMeas() throws IOException {
         final Path metadata = Paths.get("src/test/resources/valid_metadata.json");
         List<Event> events = EventUtils
                 .eventsFromDirectory(Paths.get(dataDirectory.toString() + "/valid_data/"), metadata)
                 .stream()
                 .map(e -> {
-                    System.out.println(e.getBody());
                     MeasCollecFile m = converter.convert(e.getBody());
-                    System.out.println(m.getMeasData());
                     e.setMeasCollecFile(m);
                     return e;
                 })
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/mapping/MapperTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/mapping/MapperTest.java
index eeaa39e..e7c8f3a 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/mapping/MapperTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/mapping/MapperTest.java
@@ -20,6 +20,7 @@
 
 package org.onap.dcaegen2.services.pmmapper.mapping;
 
+import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
@@ -35,6 +36,7 @@ import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.everit.json.schema.Schema;
@@ -46,6 +48,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MappingException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.XMLParseException;
@@ -133,6 +136,15 @@ class MapperTest {
         assertThrows(NullPointerException.class, () -> objUnderTest.map(mock(Event.class)));
     }
 
+    @Test
+    void testMapEvents() throws IOException {
+        List<Event> events = getValidEvents();
+        List<Event> expectedEvents = objUnderTest.mapEvents(events);
+        expectedEvents.forEach(event->{
+            assertTrue(event.getVes() != null);
+        });
+    }
+
     static List<Event> getValidEvents() throws IOException {
         return EventUtils.eventsFromDirectory(Paths.get(dataDirectory.toString() + "/valid_data/"), metadata);
     }
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitterTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitterTest.java
index 962303c..54c5091 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitterTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/MeasSplitterTest.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertEquals;
 import java.nio.file.Paths;

 import java.util.HashMap;

 import java.util.List;

+import java.util.NoSuchElementException;

 

 import javax.xml.bind.JAXBException;

 

@@ -35,6 +36,7 @@ import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;

 import org.onap.dcaegen2.services.pmmapper.model.Event;

 import org.onap.dcaegen2.services.pmmapper.model.EventMetadata;

+import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;

 import org.onap.dcaegen2.services.pmmapper.model.MeasCollecFile;

 

 import io.undertow.server.HttpServerExchange;

@@ -51,11 +53,16 @@ public class MeasSplitterTest {
     EventMetadata meta;

     @Mock

     Event event;

+    @Mock

+    MapperConfig config;

 

     @BeforeEach

     public void setup() {

         converter =  new MeasConverter();

         objUnderTest = new MeasSplitter(converter);

+    }

+

+    public void setupBaseEvent() {

         Mockito.when(event.getHttpServerExchange()).thenReturn(exchange);

         Mockito.when(event.getMetadata()).thenReturn(meta);

         Mockito.when(event.getMdc()).thenReturn(new HashMap<String, String>());

@@ -63,11 +70,24 @@ public class MeasSplitterTest {
         Mockito.when(event.getPublishIdentity()).thenReturn("");

     }

 

+

+    @Test

+    public void no_measData() {

+        String inputPath = baseDir + "no_measdata";

+        String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));

+        Mockito.when(event.getBody()).thenReturn(inputXml);

+

+        Assertions.assertThrows(NoSuchElementException.class, ()->{

+            objUnderTest.split(event);

+        });

+    }

+

     @Test

     public void typeA_returns_only_one_event() throws JAXBException {

         String inputPath = baseDir + "meas_results_typeA";

         String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));

         MeasCollecFile measToBeSplit = converter.convert(inputXml);

+        setupBaseEvent();

         Mockito.when(event.getBody()).thenReturn(inputXml);

         Mockito.when(event.getMeasCollecFile()).thenReturn(measToBeSplit);

 

@@ -80,6 +100,7 @@ public class MeasSplitterTest {
     public void typeC_returns_multiple_events() throws JAXBException {

         String inputPath = baseDir + "meas_results_typeC";

         String inputXml = EventUtils.fileContentsToString(Paths.get(inputPath + ".xml"));

+        setupBaseEvent();

         Mockito.when(event.getBody()).thenReturn(inputXml);

         MeasCollecFile measToBeSplit = converter.convert(inputXml);

         Mockito.when(event.getMeasCollecFile()).thenReturn(measToBeSplit);

diff --git a/src/test/resources/split_test/no_measdata.xml b/src/test/resources/split_test/no_measdata.xml
new file mode 100644
index 0000000..5b8eb5a
--- /dev/null
+++ b/src/test/resources/split_test/no_measdata.xml
@@ -0,0 +1,10 @@
+<measCollecFile xmlns="http://www.3gpp.org/ftp/specs/archive/32_series/32.435#measCollec">
+    <fileHeader dnPrefix="some dnPrefix" vendorName="FooBar Ltd"
+                fileFormatVersion="32.435 V10.0">
+        <fileSender localDn="Dublin"/>
+        <measCollec beginTime="2018-10-02T12:00:00+01:00"/>
+    </fileHeader>
+    <fileFooter>
+        <measCollec endTime="2018-10-02T12:15:00+01:00"/>
+    </fileFooter>
+</measCollecFile>