bug: bot library mode: clear internal messages before run and add test

before a message is processed in library mode, clear the internal queue
of the pipeline and the cached message of the bot instance.

in case of leftovers, which can happen when the process method raised
exception, the leftover message would take precendence over any new
messages and thus keep the bot in a endless loop of wrong errors

this commit also adds a unittest to cover this behaviour

Author: wagner-intevation

intelmq/lib/bot.py

@@ -1023,13 +1023,17 @@ def process_message(self, *messages: Union[libmessage.Message, dict]):
 
         Access the output queue e.g. with return_value['output']
         """
-        print('process_message, destination state:', self.__destination_pipeline.state, 'self.destination_queues', self.destination_queues)
         if self.bottype == BotType.COLLECTOR:
             if messages:
                 raise exceptions.InvalidArgument('Collector Bots take no messages as processing input')
         else:
             # reset source queue
             self.__source_pipeline.state[self.source_queue] = []
+            # reset internal queue
+            if self.__source_pipeline._has_message:
+                self.__source_pipeline.acknowledge()
+            self.__current_message = None
+
             for message in messages:
                 # convert to Message objects, it the message is a dict
                 # use an appropriate default message type, not requiring __type keys in the message

intelmq/lib/pipeline.py

@@ -15,7 +15,7 @@
 
 import time
 from itertools import chain
-from typing import Dict, Optional
+from typing import Optional
 import ssl
 
 import redis

intelmq/tests/lib/test_bot_library_mode.py

@@ -7,7 +7,7 @@
 
 Software engineering by BSI & Intevation GmbH
 
-This tests IntelMQ bots in library mode (IEP007)
+This file tests IntelMQ bots in library mode (IEP007)
 """
 import json
 import unittest
@@ -28,13 +28,32 @@
                                            'source.urlpath': '/',
                                            'protocol.application': 'http',
                                            'protocol.transport': 'tcp'}
+EXAMPLE_IP_INPUT = {"source.ip": "192.0.43.7",  # icann.org.
+                    "destination.ip": "192.0.43.8",  # iana.org.
+                    "time.observation": "2015-01-01T00:00:00+00:00",
+                    }
 
 
 class BrokenInitExpertBot(ExpertBot):
     def init(self):
         raise ValueError('This initialization intionally raises an error!')
 
 
+class RaisesOnFirstRunExpertBot(ExpertBot):
+    counter = 0
+
+    def init(self):
+        self.counter = 0
+
+    def process(self):
+        event = self.receive_message()
+        self.counter += 1
+        if self.counter == 1:
+            raise ValueError('This initialization intionally raises an error!')
+        self.send_message(event)
+        self.acknowledge_message()
+
+
 def assertMessageEqual(actual, expected):
     """
     Compare two messages as dicts.
@@ -109,5 +128,18 @@ def test_bot_multi_message():
     assert queues['output'] == [EXAMPLE_DATA_URL_OUT] * 2
 
 
+def test_bot_raises_and_second_message():
+    """
+    The first message raises an error and the second message
+    This test is based on an issue where the exception-raising message was not cleared from the internal message store of the Bot/Pipeline instance and thus re-used on the second run
+    """
+    raises_on_first_run = RaisesOnFirstRunExpertBot('raises', settings=BotLibSettings)
+    with raises(ValueError):
+        raises_on_first_run.process_message(EXAMPLE_DATA_URL)
+    queues = raises_on_first_run.process_message(EXAMPLE_IP_INPUT)
+    assert len(queues['output']) == 1
+    assertMessageEqual(queues['output'][0], EXAMPLE_IP_INPUT)
+
+
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()