metron-platform/metron-parsing/metron-parsers-common/src/test/java/org/apache/metron/parsers/ParserRunnerImplTest.java - metron - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.
  */
 package org.apache.metron.parsers;

 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.powermock.api.mockito.PowerMockito.mockStatic;
 import static org.powermock.api.mockito.PowerMockito.when;

 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
 import org.adrianwalker.multilinestring.Multiline;
 import org.apache.metron.common.Constants;
 import org.apache.metron.common.Constants.Fields;
 import org.apache.metron.common.configuration.ParserConfigurations;
 import org.apache.metron.common.configuration.SensorParserConfig;
 import org.apache.metron.common.error.MetronError;
 import org.apache.metron.common.message.metadata.RawMessage;
 import org.apache.metron.common.utils.JSONUtils;
 import org.apache.metron.common.utils.ReflectionUtils;
 import org.apache.metron.parsers.ParserRunnerImpl.ProcessResult;
 import org.apache.metron.parsers.filters.Filters;
 import org.apache.metron.parsers.filters.StellarFilter;
 import org.apache.metron.parsers.interfaces.MessageFilter;
 import org.apache.metron.parsers.interfaces.MessageParser;
 import org.apache.metron.parsers.interfaces.MessageParserResult;
 import org.apache.metron.stellar.dsl.Context;
 import org.json.simple.JSONObject;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;

 @RunWith(PowerMockRunner.class)
 @PrepareForTest({ParserRunnerImpl.class, ReflectionUtils.class, Filters.class})
 public class ParserRunnerImplTest {

   @Rule
   public final ExpectedException exception = ExpectedException.none();

   /**
    {
    "fieldValidations" : [
      {
        "input" : [ "ip_src_addr", "ip_dst_addr"],
        "validation" : "IP"
      }
    ]
    }
    */
   @Multiline
   private String globalConfigString;

   /**
    {
      "parserClassName":"org.apache.metron.parsers.bro.BasicBroParser",
      "filterClassName":"org.apache.metron.parsers.filters.StellarFilter",
      "sensorTopic":"bro",
      "parserConfig": {
        "field": "value"
      },
      "fieldTransformations" : [
        {
          "input" : "field1",
          "transformation" : "REMOVE"
        }
      ]
    }
    */
   @Multiline
   private String broConfigString;

   /**
    {
      "parserClassName":"org.apache.metron.parsers.snort.BasicSnortParser",
      "sensorTopic":"snort",
      "parserConfig": {}
    }
    */
   @Multiline
   private String snortConfigString;

   private ParserConfigurations parserConfigurations;
   private MessageParser<JSONObject> broParser;
   private MessageParser<JSONObject> snortParser;
   private MessageFilter<JSONObject> stellarFilter;
   private ParserRunnerImpl parserRunner;


   @Before
   public void setup() throws IOException {
     parserConfigurations = new ParserConfigurations();
     SensorParserConfig broConfig = SensorParserConfig.fromBytes(broConfigString.getBytes());
     SensorParserConfig snortConfig = SensorParserConfig.fromBytes(snortConfigString.getBytes());
     parserConfigurations.updateSensorParserConfig("bro", broConfig);
     parserConfigurations.updateSensorParserConfig("snort", snortConfig);
     parserConfigurations.updateGlobalConfig(JSONUtils.INSTANCE.load(globalConfigString, JSONUtils.MAP_SUPPLIER));
     parserRunner = new ParserRunnerImpl(new HashSet<>(Arrays.asList("bro", "snort")));
     broParser = mock(MessageParser.class);
     snortParser = mock(MessageParser.class);
     stellarFilter = mock(StellarFilter.class);
     mockStatic(ReflectionUtils.class);
     mockStatic(Filters.class);

     when(ReflectionUtils.createInstance("org.apache.metron.parsers.bro.BasicBroParser")).thenReturn(broParser);
     when(ReflectionUtils.createInstance("org.apache.metron.parsers.snort.BasicSnortParser")).thenReturn(snortParser);
     when(Filters.get("org.apache.metron.parsers.filters.StellarFilter", broConfig.getParserConfig()))
             .thenReturn(stellarFilter);

   }

   @Test
   public void shouldThrowExceptionOnEmptyParserSupplier() {
     exception.expect(IllegalStateException.class);
     exception.expectMessage("A parser config supplier must be set before initializing the ParserRunner.");

     parserRunner.init(null, null);
   }

   @Test
   public void shouldThrowExceptionOnEmptyStellarContext() {
     exception.expect(IllegalStateException.class);
     exception.expectMessage("A stellar context must be set before initializing the ParserRunner.");

     parserRunner.init(() -> parserConfigurations, null);
   }

   @Test
   public void initShouldThrowExceptionOnMissingSensorParserConfig() {
     exception.expect(IllegalStateException.class);
     exception.expectMessage("Could not initialize parsers.  Cannot find configuration for sensor test.");

     parserRunner = new ParserRunnerImpl(new HashSet<String>() {{
       add("test");
     }});

     parserRunner.init(() -> parserConfigurations, mock(Context.class));
   }

   @Test
   public void executeShouldThrowExceptionOnMissingSensorParserConfig() {
     exception.expect(IllegalStateException.class);
     exception.expectMessage("Could not execute parser.  Cannot find configuration for sensor test.");

     parserRunner = new ParserRunnerImpl(new HashSet<String>() {{
       add("test");
     }});

     parserRunner.execute("test", mock(RawMessage.class), parserConfigurations);
   }

   @Test
   public void shouldInit() throws Exception {
     Context stellarContext = mock(Context.class);
     Map<String, Object> broParserConfig = parserConfigurations.getSensorParserConfig("bro").getParserConfig();
     Map<String, Object> snortParserConfig = parserConfigurations.getSensorParserConfig("snort").getParserConfig();

     parserRunner.init(() -> parserConfigurations, stellarContext);

     {
       // Verify Stellar context
       Assert.assertEquals(stellarContext, parserRunner.getStellarContext());
     }

     Map<String, ParserComponent> sensorToParserComponentMap = parserRunner.getSensorToParserComponentMap();
     {
       // Verify Bro parser initialization
       Assert.assertEquals(2, sensorToParserComponentMap.size());
       ParserComponent broComponent = sensorToParserComponentMap.get("bro");
       Assert.assertEquals(broParser, broComponent.getMessageParser());
       Assert.assertEquals(stellarFilter, broComponent.getFilter());
       verify(broParser, times(1)).init();
       verify(broParser, times(1)).configure(broParserConfig);
       verifyNoMoreInteractions(broParser);
       verifyNoMoreInteractions(stellarFilter);
     }
     {
       // Verify Snort parser initialization
       ParserComponent snortComponent = sensorToParserComponentMap.get("snort");
       Assert.assertEquals(snortParser, snortComponent.getMessageParser());
       Assert.assertNull(snortComponent.getFilter());
       verify(snortParser, times(1)).init();
       verify(snortParser, times(1)).configure(snortParserConfig);
       verifyNoMoreInteractions(snortParser);
     }
   }

   /**
    * This is only testig the execute method. It mocks out processMessage().
    */
   @Test
   public void shouldExecute() {
     parserRunner = spy(parserRunner);
     RawMessage rawMessage = new RawMessage("raw_message".getBytes(), new HashMap<>());
     JSONObject parsedMessage1 = new JSONObject();
     parsedMessage1.put("field", "parsedMessage1");
     JSONObject parsedMessage2 = new JSONObject();
     parsedMessage2.put("field", "parsedMessage2");
     Object rawMessage1 = new RawMessage("raw_message1".getBytes(), new HashMap<>());
     Object rawMessage2 = new RawMessage("raw_message2".getBytes(), new HashMap<>());
     Throwable throwable1 = mock(Throwable.class);
     Throwable throwable2 = mock(Throwable.class);
     MessageParserResult<JSONObject> messageParserResult = new DefaultMessageParserResult<>(Arrays.asList(parsedMessage1, parsedMessage2),
             new HashMap<Object, Throwable>(){{
               put(rawMessage1, throwable1);
               put(rawMessage2, throwable2);
             }});
     JSONObject processedMessage = new JSONObject();
     processedMessage.put("field", "processedMessage1");
     MetronError processedError = new MetronError().withMessage("processedError");
     ProcessResult processedMessageResult = mock(ProcessResult.class);
     ProcessResult processedErrorResult = mock(ProcessResult.class);

     when(broParser.parseOptionalResult(rawMessage.getMessage())).thenReturn(Optional.of(messageParserResult));
     when(processedMessageResult.getMessage()).thenReturn(processedMessage);
     when(processedErrorResult.isError()).thenReturn(true);
     when(processedErrorResult.getError()).thenReturn(processedError);
     doReturn(Optional.of(processedMessageResult)).when(parserRunner)
             .processMessage("bro", parsedMessage1, rawMessage, broParser, parserConfigurations);
     doReturn(Optional.of(processedErrorResult)).when(parserRunner)
             .processMessage("bro", parsedMessage2, rawMessage, broParser, parserConfigurations);

     MetronError expectedParseError1 = new MetronError()
             .withErrorType(Constants.ErrorType.PARSER_ERROR)
             .withThrowable(throwable1)
             .withSensorType(Collections.singleton("bro"))
             .addRawMessage(rawMessage1);
     MetronError expectedParseError2 = new MetronError()
             .withErrorType(Constants.ErrorType.PARSER_ERROR)
             .withThrowable(throwable2)
             .withSensorType(Collections.singleton("bro"))
             .addRawMessage(rawMessage2);

     parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
       put("bro", new ParserComponent(broParser, stellarFilter));
     }});
     ParserRunnerResults<JSONObject> parserRunnerResults = parserRunner.execute("bro", rawMessage, parserConfigurations);

     Assert.assertEquals(1, parserRunnerResults.getMessages().size());
     Assert.assertTrue(parserRunnerResults.getMessages().contains(processedMessage));
     Assert.assertEquals(3, parserRunnerResults.getErrors().size());
     Assert.assertTrue(parserRunnerResults.getErrors().contains(processedError));
     Assert.assertTrue(parserRunnerResults.getErrors().contains(expectedParseError1));
     Assert.assertTrue(parserRunnerResults.getErrors().contains(expectedParseError2));
   }

   @Test
   public void shouldExecuteWithMasterThrowable() {
     parserRunner = spy(parserRunner);
     RawMessage rawMessage = new RawMessage("raw_message".getBytes(), new HashMap<>());
     Throwable masterThrowable = mock(Throwable.class);
     MessageParserResult<JSONObject> messageParserResult = new DefaultMessageParserResult<>(masterThrowable);


     when(broParser.parseOptionalResult(rawMessage.getMessage())).thenReturn(Optional.of(messageParserResult));

     parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
       put("bro", new ParserComponent(broParser, stellarFilter));
     }});
     ParserRunnerResults<JSONObject> parserRunnerResults = parserRunner.execute("bro", rawMessage, parserConfigurations);

     verify(parserRunner, times(0))
             .processMessage(any(), any(), any(), any(), any());

     MetronError expectedError = new MetronError()
             .withErrorType(Constants.ErrorType.PARSER_ERROR)
             .withThrowable(masterThrowable)
             .withSensorType(Collections.singleton("bro"))
             .addRawMessage(rawMessage.getMessage());
     Assert.assertEquals(1, parserRunnerResults.getErrors().size());
     Assert.assertTrue(parserRunnerResults.getErrors().contains(expectedError));
   }

   /**
    * This is only testing the processMessage method
    */
   @Test
   public void shouldPopulateMessagesOnProcessMessage() {
     JSONObject inputMessage = new JSONObject();
     inputMessage.put("guid", "guid");
     inputMessage.put("ip_src_addr", "192.168.1.1");
     inputMessage.put("ip_dst_addr", "192.168.1.2");
     RawMessage rawMessage = new RawMessage("raw_message_for_testing".getBytes(), new HashMap<>());

     JSONObject expectedOutput  = new JSONObject();
     expectedOutput.put("guid", "guid");
     expectedOutput.put("source.type", "bro");
     expectedOutput.put("ip_src_addr", "192.168.1.1");
     expectedOutput.put("ip_dst_addr", "192.168.1.2");
     expectedOutput.put(Fields.ORIGINAL.getName(), "raw_message_for_testing");

     when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
     when(broParser.validate(expectedOutput)).thenReturn(true);

     parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
       put("bro", new ParserComponent(broParser, stellarFilter));
     }});

     Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);
     Assert.assertTrue(processResult.isPresent());
     Assert.assertFalse(processResult.get().isError());
     Assert.assertEquals(expectedOutput, processResult.get().getMessage());
   }

   /**
    * This is only testing the processMessage method
    */
   @Test
   public void shouldNotOverwriteOriginalStringAddedByParser() {
     JSONObject inputMessage = new JSONObject();
     inputMessage.put("guid", "guid");
     inputMessage.put("ip_src_addr", "192.168.1.1");
     inputMessage.put("ip_dst_addr", "192.168.1.2");
     inputMessage.put(Fields.ORIGINAL.getName(), "original_string_added_by_parser");
     RawMessage rawMessage = new RawMessage("raw_message_for_testing".getBytes(), new HashMap<>());

     JSONObject expectedOutput  = new JSONObject();
     expectedOutput.put("guid", "guid");
     expectedOutput.put("source.type", "bro");
     expectedOutput.put("ip_src_addr", "192.168.1.1");
     expectedOutput.put("ip_dst_addr", "192.168.1.2");
     expectedOutput.put(Fields.ORIGINAL.getName(), "original_string_added_by_parser");

     when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
     when(broParser.validate(expectedOutput)).thenReturn(true);

     parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
       put("bro", new ParserComponent(broParser, stellarFilter));
     }});

     Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);
     Assert.assertTrue(processResult.isPresent());
     Assert.assertFalse(processResult.get().isError());
     Assert.assertEquals(expectedOutput, processResult.get().getMessage());
   }

   @Test
   public void shouldReturnMetronErrorOnInvalidMessage() {
     Map<String, Object> metadata = new HashMap<>();
     metadata.put("metron.metadata.topic", "bro");
     metadata.put("metron.metadata.partition", 0);
     metadata.put("metron.metadata.offset", 123);

     JSONObject inputMessage = new JSONObject();
     inputMessage.put("guid", "guid");
     RawMessage rawMessage = new RawMessage("raw_message".getBytes(), metadata);

     JSONObject expectedOutput  = new JSONObject();
     expectedOutput.put("guid", "guid");
     expectedOutput.put("source.type", "bro");
     expectedOutput.put(Fields.ORIGINAL.getName(), "raw_message");
     MetronError expectedMetronError = new MetronError()
             .withErrorType(Constants.ErrorType.PARSER_INVALID)
             .withSensorType(Collections.singleton("bro"))
             .withMetadata(metadata)
             .addRawMessage(inputMessage);

     when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
     // This is the important switch. Not to be confused with field validators.
     when(broParser.validate(expectedOutput)).thenReturn(false);

     parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
       put("bro", new ParserComponent(broParser, stellarFilter));
     }});

     Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);

     Assert.assertTrue(processResult.isPresent());
     Assert.assertTrue(processResult.get().isError());
     Assert.assertEquals(expectedMetronError, processResult.get().getError());
   }

   @Test
   public void shouldReturnMetronErrorOnFailedFieldValidator() {
     Map<String, Object> metadata = new HashMap<>();
     metadata.put("metron.metadata.topic", "bro");
     metadata.put("metron.metadata.partition", 0);
     metadata.put("metron.metadata.offset", 123);

     JSONObject inputMessage = new JSONObject();
     inputMessage.put("guid", "guid");
     inputMessage.put("ip_src_addr", "test");
     inputMessage.put("ip_dst_addr", "test");
     RawMessage rawMessage = new RawMessage("raw_message".getBytes(), metadata);

     JSONObject expectedOutput  = new JSONObject();
     expectedOutput.put("guid", "guid");
     expectedOutput.put("ip_src_addr", "test");
     expectedOutput.put("ip_dst_addr", "test");
     expectedOutput.put("source.type", "bro");
     expectedOutput.put(Fields.ORIGINAL.getName(), "raw_message");
     MetronError expectedMetronError = new MetronError()
             .withErrorType(Constants.ErrorType.PARSER_INVALID)
             .withSensorType(Collections.singleton("bro"))
             .addRawMessage(inputMessage)
             .withMetadata(metadata)
             .withErrorFields(new HashSet<>(Arrays.asList("ip_src_addr", "ip_dst_addr")));

     when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
     when(broParser.validate(expectedOutput)).thenReturn(true);

     parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
       put("bro", new ParserComponent(broParser, stellarFilter));
     }});

     Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);

     Assert.assertTrue(processResult.isPresent());
     Assert.assertTrue(processResult.get().isError());
     Assert.assertEquals(expectedMetronError, processResult.get().getError());
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.
	*/
	package org.apache.metron.parsers;

	import static org.mockito.Matchers.any;
	import static org.mockito.Mockito.doReturn;
	import static org.mockito.Mockito.mock;
	import static org.mockito.Mockito.spy;
	import static org.mockito.Mockito.times;
	import static org.mockito.Mockito.verify;
	import static org.mockito.Mockito.verifyNoMoreInteractions;
	import static org.powermock.api.mockito.PowerMockito.mockStatic;
	import static org.powermock.api.mockito.PowerMockito.when;

	import java.io.IOException;
	import java.util.Arrays;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.Map;
	import java.util.Optional;
	import org.adrianwalker.multilinestring.Multiline;
	import org.apache.metron.common.Constants;
	import org.apache.metron.common.Constants.Fields;
	import org.apache.metron.common.configuration.ParserConfigurations;
	import org.apache.metron.common.configuration.SensorParserConfig;
	import org.apache.metron.common.error.MetronError;
	import org.apache.metron.common.message.metadata.RawMessage;
	import org.apache.metron.common.utils.JSONUtils;
	import org.apache.metron.common.utils.ReflectionUtils;
	import org.apache.metron.parsers.ParserRunnerImpl.ProcessResult;
	import org.apache.metron.parsers.filters.Filters;
	import org.apache.metron.parsers.filters.StellarFilter;
	import org.apache.metron.parsers.interfaces.MessageFilter;
	import org.apache.metron.parsers.interfaces.MessageParser;
	import org.apache.metron.parsers.interfaces.MessageParserResult;
	import org.apache.metron.stellar.dsl.Context;
	import org.json.simple.JSONObject;
	import org.junit.Assert;
	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;
	import org.junit.rules.ExpectedException;
	import org.junit.runner.RunWith;
	import org.powermock.core.classloader.annotations.PrepareForTest;
	import org.powermock.modules.junit4.PowerMockRunner;

	@RunWith(PowerMockRunner.class)
	@PrepareForTest({ParserRunnerImpl.class, ReflectionUtils.class, Filters.class})
	public class ParserRunnerImplTest {

	@Rule
	public final ExpectedException exception = ExpectedException.none();

	/**
	{
	"fieldValidations" : [
	{
	"input" : [ "ip_src_addr", "ip_dst_addr"],
	"validation" : "IP"
	}
	]
	}
	*/
	@Multiline
	private String globalConfigString;

	/**
	{
	"parserClassName":"org.apache.metron.parsers.bro.BasicBroParser",
	"filterClassName":"org.apache.metron.parsers.filters.StellarFilter",
	"sensorTopic":"bro",
	"parserConfig": {
	"field": "value"
	},
	"fieldTransformations" : [
	{
	"input" : "field1",
	"transformation" : "REMOVE"
	}
	]
	}
	*/
	@Multiline
	private String broConfigString;

	/**
	{
	"parserClassName":"org.apache.metron.parsers.snort.BasicSnortParser",
	"sensorTopic":"snort",
	"parserConfig": {}
	}
	*/
	@Multiline
	private String snortConfigString;

	private ParserConfigurations parserConfigurations;
	private MessageParser<JSONObject> broParser;
	private MessageParser<JSONObject> snortParser;
	private MessageFilter<JSONObject> stellarFilter;
	private ParserRunnerImpl parserRunner;


	@Before
	public void setup() throws IOException {
	parserConfigurations = new ParserConfigurations();
	SensorParserConfig broConfig = SensorParserConfig.fromBytes(broConfigString.getBytes());
	SensorParserConfig snortConfig = SensorParserConfig.fromBytes(snortConfigString.getBytes());
	parserConfigurations.updateSensorParserConfig("bro", broConfig);
	parserConfigurations.updateSensorParserConfig("snort", snortConfig);
	parserConfigurations.updateGlobalConfig(JSONUtils.INSTANCE.load(globalConfigString, JSONUtils.MAP_SUPPLIER));
	parserRunner = new ParserRunnerImpl(new HashSet<>(Arrays.asList("bro", "snort")));
	broParser = mock(MessageParser.class);
	snortParser = mock(MessageParser.class);
	stellarFilter = mock(StellarFilter.class);
	mockStatic(ReflectionUtils.class);
	mockStatic(Filters.class);

	when(ReflectionUtils.createInstance("org.apache.metron.parsers.bro.BasicBroParser")).thenReturn(broParser);
	when(ReflectionUtils.createInstance("org.apache.metron.parsers.snort.BasicSnortParser")).thenReturn(snortParser);
	when(Filters.get("org.apache.metron.parsers.filters.StellarFilter", broConfig.getParserConfig()))
	.thenReturn(stellarFilter);

	}

	@Test
	public void shouldThrowExceptionOnEmptyParserSupplier() {
	exception.expect(IllegalStateException.class);
	exception.expectMessage("A parser config supplier must be set before initializing the ParserRunner.");

	parserRunner.init(null, null);
	}

	@Test
	public void shouldThrowExceptionOnEmptyStellarContext() {
	exception.expect(IllegalStateException.class);
	exception.expectMessage("A stellar context must be set before initializing the ParserRunner.");

	parserRunner.init(() -> parserConfigurations, null);
	}

	@Test
	public void initShouldThrowExceptionOnMissingSensorParserConfig() {
	exception.expect(IllegalStateException.class);
	exception.expectMessage("Could not initialize parsers. Cannot find configuration for sensor test.");

	parserRunner = new ParserRunnerImpl(new HashSet<String>() {{
	add("test");
	}});

	parserRunner.init(() -> parserConfigurations, mock(Context.class));
	}

	@Test
	public void executeShouldThrowExceptionOnMissingSensorParserConfig() {
	exception.expect(IllegalStateException.class);
	exception.expectMessage("Could not execute parser. Cannot find configuration for sensor test.");

	parserRunner = new ParserRunnerImpl(new HashSet<String>() {{
	add("test");
	}});

	parserRunner.execute("test", mock(RawMessage.class), parserConfigurations);
	}

	@Test
	public void shouldInit() throws Exception {
	Context stellarContext = mock(Context.class);
	Map<String, Object> broParserConfig = parserConfigurations.getSensorParserConfig("bro").getParserConfig();
	Map<String, Object> snortParserConfig = parserConfigurations.getSensorParserConfig("snort").getParserConfig();

	parserRunner.init(() -> parserConfigurations, stellarContext);

	{
	// Verify Stellar context
	Assert.assertEquals(stellarContext, parserRunner.getStellarContext());
	}

	Map<String, ParserComponent> sensorToParserComponentMap = parserRunner.getSensorToParserComponentMap();
	{
	// Verify Bro parser initialization
	Assert.assertEquals(2, sensorToParserComponentMap.size());
	ParserComponent broComponent = sensorToParserComponentMap.get("bro");
	Assert.assertEquals(broParser, broComponent.getMessageParser());
	Assert.assertEquals(stellarFilter, broComponent.getFilter());
	verify(broParser, times(1)).init();
	verify(broParser, times(1)).configure(broParserConfig);
	verifyNoMoreInteractions(broParser);
	verifyNoMoreInteractions(stellarFilter);
	}
	{
	// Verify Snort parser initialization
	ParserComponent snortComponent = sensorToParserComponentMap.get("snort");
	Assert.assertEquals(snortParser, snortComponent.getMessageParser());
	Assert.assertNull(snortComponent.getFilter());
	verify(snortParser, times(1)).init();
	verify(snortParser, times(1)).configure(snortParserConfig);
	verifyNoMoreInteractions(snortParser);
	}
	}

	/**
	* This is only testig the execute method. It mocks out processMessage().
	*/
	@Test
	public void shouldExecute() {
	parserRunner = spy(parserRunner);
	RawMessage rawMessage = new RawMessage("raw_message".getBytes(), new HashMap<>());
	JSONObject parsedMessage1 = new JSONObject();
	parsedMessage1.put("field", "parsedMessage1");
	JSONObject parsedMessage2 = new JSONObject();
	parsedMessage2.put("field", "parsedMessage2");
	Object rawMessage1 = new RawMessage("raw_message1".getBytes(), new HashMap<>());
	Object rawMessage2 = new RawMessage("raw_message2".getBytes(), new HashMap<>());
	Throwable throwable1 = mock(Throwable.class);
	Throwable throwable2 = mock(Throwable.class);
	MessageParserResult<JSONObject> messageParserResult = new DefaultMessageParserResult<>(Arrays.asList(parsedMessage1, parsedMessage2),
	new HashMap<Object, Throwable>(){{
	put(rawMessage1, throwable1);
	put(rawMessage2, throwable2);
	}});
	JSONObject processedMessage = new JSONObject();
	processedMessage.put("field", "processedMessage1");
	MetronError processedError = new MetronError().withMessage("processedError");
	ProcessResult processedMessageResult = mock(ProcessResult.class);
	ProcessResult processedErrorResult = mock(ProcessResult.class);

	when(broParser.parseOptionalResult(rawMessage.getMessage())).thenReturn(Optional.of(messageParserResult));
	when(processedMessageResult.getMessage()).thenReturn(processedMessage);
	when(processedErrorResult.isError()).thenReturn(true);
	when(processedErrorResult.getError()).thenReturn(processedError);
	doReturn(Optional.of(processedMessageResult)).when(parserRunner)
	.processMessage("bro", parsedMessage1, rawMessage, broParser, parserConfigurations);
	doReturn(Optional.of(processedErrorResult)).when(parserRunner)
	.processMessage("bro", parsedMessage2, rawMessage, broParser, parserConfigurations);

	MetronError expectedParseError1 = new MetronError()
	.withErrorType(Constants.ErrorType.PARSER_ERROR)
	.withThrowable(throwable1)
	.withSensorType(Collections.singleton("bro"))
	.addRawMessage(rawMessage1);
	MetronError expectedParseError2 = new MetronError()
	.withErrorType(Constants.ErrorType.PARSER_ERROR)
	.withThrowable(throwable2)
	.withSensorType(Collections.singleton("bro"))
	.addRawMessage(rawMessage2);

	parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
	put("bro", new ParserComponent(broParser, stellarFilter));
	}});
	ParserRunnerResults<JSONObject> parserRunnerResults = parserRunner.execute("bro", rawMessage, parserConfigurations);

	Assert.assertEquals(1, parserRunnerResults.getMessages().size());
	Assert.assertTrue(parserRunnerResults.getMessages().contains(processedMessage));
	Assert.assertEquals(3, parserRunnerResults.getErrors().size());
	Assert.assertTrue(parserRunnerResults.getErrors().contains(processedError));
	Assert.assertTrue(parserRunnerResults.getErrors().contains(expectedParseError1));
	Assert.assertTrue(parserRunnerResults.getErrors().contains(expectedParseError2));
	}

	@Test
	public void shouldExecuteWithMasterThrowable() {
	parserRunner = spy(parserRunner);
	RawMessage rawMessage = new RawMessage("raw_message".getBytes(), new HashMap<>());
	Throwable masterThrowable = mock(Throwable.class);
	MessageParserResult<JSONObject> messageParserResult = new DefaultMessageParserResult<>(masterThrowable);


	when(broParser.parseOptionalResult(rawMessage.getMessage())).thenReturn(Optional.of(messageParserResult));

	parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
	put("bro", new ParserComponent(broParser, stellarFilter));
	}});
	ParserRunnerResults<JSONObject> parserRunnerResults = parserRunner.execute("bro", rawMessage, parserConfigurations);

	verify(parserRunner, times(0))
	.processMessage(any(), any(), any(), any(), any());

	MetronError expectedError = new MetronError()
	.withErrorType(Constants.ErrorType.PARSER_ERROR)
	.withThrowable(masterThrowable)
	.withSensorType(Collections.singleton("bro"))
	.addRawMessage(rawMessage.getMessage());
	Assert.assertEquals(1, parserRunnerResults.getErrors().size());
	Assert.assertTrue(parserRunnerResults.getErrors().contains(expectedError));
	}

	/**
	* This is only testing the processMessage method
	*/
	@Test
	public void shouldPopulateMessagesOnProcessMessage() {
	JSONObject inputMessage = new JSONObject();
	inputMessage.put("guid", "guid");
	inputMessage.put("ip_src_addr", "192.168.1.1");
	inputMessage.put("ip_dst_addr", "192.168.1.2");
	RawMessage rawMessage = new RawMessage("raw_message_for_testing".getBytes(), new HashMap<>());

	JSONObject expectedOutput = new JSONObject();
	expectedOutput.put("guid", "guid");
	expectedOutput.put("source.type", "bro");
	expectedOutput.put("ip_src_addr", "192.168.1.1");
	expectedOutput.put("ip_dst_addr", "192.168.1.2");
	expectedOutput.put(Fields.ORIGINAL.getName(), "raw_message_for_testing");

	when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
	when(broParser.validate(expectedOutput)).thenReturn(true);

	parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
	put("bro", new ParserComponent(broParser, stellarFilter));
	}});

	Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);
	Assert.assertTrue(processResult.isPresent());
	Assert.assertFalse(processResult.get().isError());
	Assert.assertEquals(expectedOutput, processResult.get().getMessage());
	}

	/**
	* This is only testing the processMessage method
	*/
	@Test
	public void shouldNotOverwriteOriginalStringAddedByParser() {
	JSONObject inputMessage = new JSONObject();
	inputMessage.put("guid", "guid");
	inputMessage.put("ip_src_addr", "192.168.1.1");
	inputMessage.put("ip_dst_addr", "192.168.1.2");
	inputMessage.put(Fields.ORIGINAL.getName(), "original_string_added_by_parser");
	RawMessage rawMessage = new RawMessage("raw_message_for_testing".getBytes(), new HashMap<>());

	JSONObject expectedOutput = new JSONObject();
	expectedOutput.put("guid", "guid");
	expectedOutput.put("source.type", "bro");
	expectedOutput.put("ip_src_addr", "192.168.1.1");
	expectedOutput.put("ip_dst_addr", "192.168.1.2");
	expectedOutput.put(Fields.ORIGINAL.getName(), "original_string_added_by_parser");

	when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
	when(broParser.validate(expectedOutput)).thenReturn(true);

	parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
	put("bro", new ParserComponent(broParser, stellarFilter));
	}});

	Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);
	Assert.assertTrue(processResult.isPresent());
	Assert.assertFalse(processResult.get().isError());
	Assert.assertEquals(expectedOutput, processResult.get().getMessage());
	}

	@Test
	public void shouldReturnMetronErrorOnInvalidMessage() {
	Map<String, Object> metadata = new HashMap<>();
	metadata.put("metron.metadata.topic", "bro");
	metadata.put("metron.metadata.partition", 0);
	metadata.put("metron.metadata.offset", 123);

	JSONObject inputMessage = new JSONObject();
	inputMessage.put("guid", "guid");
	RawMessage rawMessage = new RawMessage("raw_message".getBytes(), metadata);

	JSONObject expectedOutput = new JSONObject();
	expectedOutput.put("guid", "guid");
	expectedOutput.put("source.type", "bro");
	expectedOutput.put(Fields.ORIGINAL.getName(), "raw_message");
	MetronError expectedMetronError = new MetronError()
	.withErrorType(Constants.ErrorType.PARSER_INVALID)
	.withSensorType(Collections.singleton("bro"))
	.withMetadata(metadata)
	.addRawMessage(inputMessage);

	when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
	// This is the important switch. Not to be confused with field validators.
	when(broParser.validate(expectedOutput)).thenReturn(false);

	parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
	put("bro", new ParserComponent(broParser, stellarFilter));
	}});

	Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);

	Assert.assertTrue(processResult.isPresent());
	Assert.assertTrue(processResult.get().isError());
	Assert.assertEquals(expectedMetronError, processResult.get().getError());
	}

	@Test
	public void shouldReturnMetronErrorOnFailedFieldValidator() {
	Map<String, Object> metadata = new HashMap<>();
	metadata.put("metron.metadata.topic", "bro");
	metadata.put("metron.metadata.partition", 0);
	metadata.put("metron.metadata.offset", 123);

	JSONObject inputMessage = new JSONObject();
	inputMessage.put("guid", "guid");
	inputMessage.put("ip_src_addr", "test");
	inputMessage.put("ip_dst_addr", "test");
	RawMessage rawMessage = new RawMessage("raw_message".getBytes(), metadata);

	JSONObject expectedOutput = new JSONObject();
	expectedOutput.put("guid", "guid");
	expectedOutput.put("ip_src_addr", "test");
	expectedOutput.put("ip_dst_addr", "test");
	expectedOutput.put("source.type", "bro");
	expectedOutput.put(Fields.ORIGINAL.getName(), "raw_message");
	MetronError expectedMetronError = new MetronError()
	.withErrorType(Constants.ErrorType.PARSER_INVALID)
	.withSensorType(Collections.singleton("bro"))
	.addRawMessage(inputMessage)
	.withMetadata(metadata)
	.withErrorFields(new HashSet<>(Arrays.asList("ip_src_addr", "ip_dst_addr")));

	when(stellarFilter.emit(expectedOutput, parserRunner.getStellarContext())).thenReturn(true);
	when(broParser.validate(expectedOutput)).thenReturn(true);

	parserRunner.setSensorToParserComponentMap(new HashMap<String, ParserComponent>() {{
	put("bro", new ParserComponent(broParser, stellarFilter));
	}});

	Optional<ParserRunnerImpl.ProcessResult> processResult = parserRunner.processMessage("bro", inputMessage, rawMessage, broParser, parserConfigurations);

	Assert.assertTrue(processResult.isPresent());
	Assert.assertTrue(processResult.get().isError());
	Assert.assertEquals(expectedMetronError, processResult.get().getError());
	}
	}