Add basic Processor/EventHandler test

tracetools_analysis/test/tracetools_analysis/test_processor.py

@@ -0,0 +1,76 @@
+# Copyright 2019 Robert Bosch GmbH
+#
+# 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.
+
+from inspect import signature
+from typing import Dict
+import unittest
+
+from tracetools_analysis.processor import EventHandler
+from tracetools_analysis.processor import EventMetadata
+from tracetools_analysis.processor import Processor
+
+
+class StubHandler1(EventHandler):
+
+    def __init__(self) -> None:
+        handler_map = {
+            'myeventname': self._handler_whatever,
+        }
+        super().__init__(handler_map=handler_map)
+        self.handler_called = False
+
+    def _handler_whatever(
+        self, event: Dict, metadata: EventMetadata
+    ) -> None:
+        self.handler_called = True
+
+
+class StubHandler2(EventHandler):
+
+    def __init__(self) -> None:
+        handler_map = {
+            'myeventname': self._handler_whatever,
+        }
+        super().__init__(handler_map=handler_map)
+        self.handler_called = False
+
+    def _handler_whatever(
+        self, event: Dict, metadata: EventMetadata
+    ) -> None:
+        self.handler_called = True
+
+
+class TestProcessor(unittest.TestCase):
+
+    def __init__(self, *args) -> None:
+        super().__init__(
+            *args,
+        )
+
+    def test_handler_method_with_merge(self) -> None:
+        handler1 = StubHandler1()
+        handler2 = StubHandler2()
+        mock_event = {
+            '_name': 'myeventname',
+            '_timestamp': 0,
+            'cpu_id': 0,
+        }
+        processor = Processor(handler1, handler2)
+        processor.process([mock_event])
+        self.assertTrue(handler1.handler_called, 'event handler not called')
+        self.assertTrue(handler2.handler_called, 'event handler not called')
+
+
+if __name__ == '__main__':
+    unittest.main()

tracetools_analysis/tracetools_analysis/processor/__init__.py

@@ -246,7 +246,6 @@ class Processor():
         self._handlers = self._expand_dependencies(*handlers, **kwargs)
         print('handlers after:', [type(handler).__name__ for handler in self._handlers])
         self._register_with_handlers()
-        input()
 
     def _register_with_handlers(self) -> None:
         """Register this processor with its `EventHandler`s."""
@@ -324,4 +323,3 @@ class Processor():
                     raise_if_not_found=False)
                 metadata = EventMetadata(event_name, timestamp, cpu_id, procname, pid, tid)
                 handler_function(event, metadata)
-                input()