Add `bt2.TraceCollectionNotificationIterator` tests

Signed-off-by: Philippe Proulx <eeppeliteloop@gmail.com>
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>

diff --git a/tests/bindings/python/bt2/Makefile.am b/tests/bindings/python/bt2/Makefile.am
index 557cc7b4118846e4fd7aa0650c9f7d4f441224ca..e966da9c8e9131ca81a532664de4e2401c244fc5 100644 (file)
--- a/tests/bindings/python/bt2/Makefile.am
+++ b/tests/bindings/python/bt2/Makefile.am
@@ -1,22 +1,23 @@
-EXTRA_DIST =                                   \
-       test_clock_class.py                     \
-       test_clock_class_priority_map.py        \
-       test_component.py                       \
-       test_component_class.py                 \
-       test_connection.py                      \
-       test_ctf_writer_clock.py                \
-       test_event.py                           \
-       test_event_class.py                     \
-       test_field_types.py                     \
-       test_fields.py                          \
-       test_graph.py                           \
-       test_notification.py                    \
-       test_notification_iterator.py           \
-       test_packet.py                          \
-       test_plugin.py                          \
-       test_port.py                            \
-       test_stream.py                          \
-       test_stream_class.py                    \
-       test_trace.py                           \
-       test_values.py                          \
+EXTRA_DIST =                                           \
+       test_clock_class.py                             \
+       test_clock_class_priority_map.py                \
+       test_component.py                               \
+       test_component_class.py                         \
+       test_connection.py                              \
+       test_ctf_writer_clock.py                        \
+       test_event.py                                   \
+       test_event_class.py                             \
+       test_field_types.py                             \
+       test_fields.py                                  \
+       test_graph.py                                   \
+       test_notification.py                            \
+       test_notification_iterator.py                   \
+       test_packet.py                                  \
+       test_plugin.py                                  \
+       test_port.py                                    \
+       test_stream.py                                  \
+       test_stream_class.py                            \
+       test_trace.py                                   \
+       test_trace_collection_notification_iterator.py  \
+       test_values.py                                  \
        .coveragerc

diff --git a/tests/bindings/python/bt2/test_python_bt2.in b/tests/bindings/python/bt2/test_python_bt2.in
index 4fa569e88be6c0c5efa08830e293ed007db2743c..d9c23f2b8398796e7c1a4768c80ad8f2cda7179f 100644 (file)
--- a/tests/bindings/python/bt2/test_python_bt2.in
+++ b/tests/bindings/python/bt2/test_python_bt2.in
@@ -28,6 +28,7 @@ export BABELTRACE_PYTHON_BT2_NO_TRACEBACK=1
 export TEST_PLUGIN_PLUGINS_PATH="${BT_BUILD_PATH}/plugins"
 export BABELTRACE_PLUGIN_PATH="${BT_BUILD_PATH}/plugins/ctf:${BT_BUILD_PATH}/plugins/utils:${BT_BUILD_PATH}/plugins/text"
 export LD_LIBRARY_PATH="${BT_BUILD_PATH}/lib/.libs"
+export TEST_CTF_TRACES_PATH="${BT_BUILD_PATH}/tests/ctf-traces"
 PYTHON_BUILD_DIR="${BT_BUILD_PATH}/bindings/python/bt2/build/build_lib"
 TESTS_UTILS_PYTHON_DIR="${BT_SRC_PATH}/tests/utils/python"
 TESTRUNNER_PY="${BT_SRC_PATH}/tests/utils/python/testrunner.py"

diff --git a/tests/bindings/python/bt2/test_trace_collection_notification_iterator.py b/tests/bindings/python/bt2/test_trace_collection_notification_iterator.py
new file mode 100644 (file)
index 0000000..459607e
--- /dev/null
+++ b/tests/bindings/python/bt2/test_trace_collection_notification_iterator.py
@@ -0,0 +1,128 @@
+import unittest
+import datetime
+import copy
+import uuid
+import bt2
+import os
+import os.path
+
+
+_TEST_CTF_TRACES_PATH = os.environ['TEST_CTF_TRACES_PATH']
+_3EVENTS_INTERSECT_TRACE_PATH = os.path.join(_TEST_CTF_TRACES_PATH,
+                                             'intersection',
+                                             '3eventsintersect')
+
+
+class SourceComponentSpecTestCase(unittest.TestCase):
+    def test_create_good_no_params(self):
+        spec = bt2.SourceComponentSpec('plugin', 'compcls')
+
+    def test_create_good_with_params(self):
+        spec = bt2.SourceComponentSpec('plugin', 'compcls', {'salut': 23})
+
+    def test_create_good_with_path_params(self):
+        spec = bt2.SourceComponentSpec('plugin', 'compcls', 'a path')
+        self.assertEqual(spec.params['path'], 'a path')
+
+    def test_create_wrong_plugin_name_type(self):
+        with self.assertRaises(TypeError):
+            spec = bt2.SourceComponentSpec(23, 'compcls')
+
+    def test_create_wrong_component_class_name_type(self):
+        with self.assertRaises(TypeError):
+            spec = bt2.SourceComponentSpec('plugin', 190)
+
+    def test_create_wrong_params_type(self):
+        with self.assertRaises(TypeError):
+            spec = bt2.SourceComponentSpec('dwdw', 'compcls', datetime.datetime.now())
+
+
+class TraceCollectionNotificationIteratorTestCase(unittest.TestCase):
+    def test_create_wrong_stream_intersection_mode_type(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+        with self.assertRaises(TypeError):
+            notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=23)
+
+    def test_create_wrong_begin_type(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+        with self.assertRaises(TypeError):
+            notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin='hi')
+
+    def test_create_wrong_end_type(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+        with self.assertRaises(TypeError):
+            notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin='lel')
+
+    def test_create_no_such_plugin(self):
+        specs = [bt2.SourceComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
+
+        with self.assertRaises(bt2.Error):
+            notif_iter = bt2.TraceCollectionNotificationIterator(specs)
+
+    def test_create_begin_s(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=19457.918232)
+
+    def test_create_end_s(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=123.12312)
+
+    def test_create_begin_datetime(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=datetime.datetime.now())
+
+    def test_create_end_datetime(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=datetime.datetime.now())
+
+    def test_iter_no_intersection(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs)
+        self.assertEqual(len(list(notif_iter)), 28)
+
+    def test_iter_no_intersection_subscribe(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs,
+                                                             notification_types=[bt2.EventNotification])
+        self.assertEqual(len(list(notif_iter)), 8)
+
+    def test_iter_intersection(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True)
+        self.assertEqual(len(list(notif_iter)), 15)
+
+    def test_iter_intersection_subscribe(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True,
+                                                             notification_types=[bt2.EventNotification])
+        self.assertEqual(len(list(notif_iter)), 3)
+
+    def test_iter_intersection_no_path_param(self):
+        specs = [bt2.SourceComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
+
+        with self.assertRaises(bt2.Error):
+            notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True,
+                                                                 notification_types=[bt2.EventNotification])
+
+    def test_iter_no_intersection_two_traces(self):
+        spec = bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
+        specs = [spec, spec]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs)
+        self.assertEqual(len(list(notif_iter)), 56)
+
+    def test_iter_no_intersection_begin(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs,
+                                                             notification_types=[bt2.EventNotification],
+                                                             begin=13515309.000000023)
+        self.assertEqual(len(list(notif_iter)), 6)
+
+    def test_iter_no_intersection_end(self):
+        specs = [bt2.SourceComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
+        notif_iter = bt2.TraceCollectionNotificationIterator(specs,
+                                                             notification_types=[bt2.EventNotification],
+                                                             end=13515309.000000075)
+        self.assertEqual(len(list(notif_iter)), 5)