[babeltrace.git] / tests / bindings / python / bt2 / test_trace_collection_notification_iterator.py

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')


@unittest.skip("this is broken")
class ComponentSpecTestCase(unittest.TestCase):
    def test_create_good_no_params(self):
        spec = bt2.ComponentSpec('plugin', 'compcls')

    def test_create_good_with_params(self):
        spec = bt2.ComponentSpec('plugin', 'compcls', {'salut': 23})

    def test_create_good_with_path_params(self):
        spec = bt2.ComponentSpec('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.ComponentSpec(23, 'compcls')

    def test_create_wrong_component_class_name_type(self):
        with self.assertRaises(TypeError):
            spec = bt2.ComponentSpec('plugin', 190)

    def test_create_wrong_params_type(self):
        with self.assertRaises(TypeError):
            spec = bt2.ComponentSpec('dwdw', 'compcls', datetime.datetime.now())


@unittest.skip("this is broken")
class TraceCollectionNotificationIteratorTestCase(unittest.TestCase):
    def test_create_wrong_stream_intersection_mode_type(self):
        specs = [bt2.ComponentSpec('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.ComponentSpec('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.ComponentSpec('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.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]

        with self.assertRaises(bt2.Error):
            notif_iter = bt2.TraceCollectionNotificationIterator(specs)

    def test_create_begin_s(self):
        specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
        notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=19457.918232)

    def test_create_end_s(self):
        specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
        notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=123.12312)

    def test_create_begin_datetime(self):
        specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
        notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=datetime.datetime.now())

    def test_create_end_datetime(self):
        specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
        notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=datetime.datetime.now())

    def test_iter_no_intersection(self):
        specs = [bt2.ComponentSpec('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.ComponentSpec('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_specs_not_list(self):
        spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
        notif_iter = bt2.TraceCollectionNotificationIterator(spec,
                                                             notification_types=[bt2.EventNotification])
        self.assertEqual(len(list(notif_iter)), 8)

    def test_iter_custom_filter(self):
        src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
        flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
            'end': 13515309000000075,
        })
        notif_iter = bt2.TraceCollectionNotificationIterator(src_spec, flt_spec,
                                                             notification_types=[bt2.EventNotification])
        self.assertEqual(len(list(notif_iter)), 5)

    def test_iter_intersection(self):
        specs = [bt2.ComponentSpec('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.ComponentSpec('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.ComponentSpec('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.ComponentSpec('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.ComponentSpec('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.ComponentSpec('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)
Commit	Line	Data
cc261e29 PP	1	import unittest
	2	import datetime
	3	import copy
	4	import uuid
	5	import bt2
	6	import os
	7	import os.path
	8
	9
	10	_TEST_CTF_TRACES_PATH = os.environ['TEST_CTF_TRACES_PATH']
	11	_3EVENTS_INTERSECT_TRACE_PATH = os.path.join(_TEST_CTF_TRACES_PATH,
	12	'intersection',
	13	'3eventsintersect')
	14
	15
90cfc012	16	@unittest.skip("this is broken")
88fdcc33	17	class ComponentSpecTestCase(unittest.TestCase):
cc261e29	18	def test_create_good_no_params(self):
88fdcc33	19	spec = bt2.ComponentSpec('plugin', 'compcls')
cc261e29 PP	20
cc261e29 PP	21	def test_create_good_with_params(self):
88fdcc33	22	spec = bt2.ComponentSpec('plugin', 'compcls', {'salut': 23})
cc261e29 PP	23
cc261e29 PP	24	def test_create_good_with_path_params(self):
88fdcc33	25	spec = bt2.ComponentSpec('plugin', 'compcls', 'a path')
cc261e29 PP	26	self.assertEqual(spec.params['path'], 'a path')
	27
	28	def test_create_wrong_plugin_name_type(self):
	29	with self.assertRaises(TypeError):
88fdcc33	30	spec = bt2.ComponentSpec(23, 'compcls')
cc261e29 PP	31
	32	def test_create_wrong_component_class_name_type(self):
	33	with self.assertRaises(TypeError):
88fdcc33	34	spec = bt2.ComponentSpec('plugin', 190)
cc261e29 PP	35
	36	def test_create_wrong_params_type(self):
	37	with self.assertRaises(TypeError):
88fdcc33	38	spec = bt2.ComponentSpec('dwdw', 'compcls', datetime.datetime.now())
cc261e29 PP	39
cc261e29 PP	40
90cfc012	41	@unittest.skip("this is broken")
cc261e29 PP	42	class TraceCollectionNotificationIteratorTestCase(unittest.TestCase):
cc261e29 PP	43	def test_create_wrong_stream_intersection_mode_type(self):
88fdcc33	44	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	45
	46	with self.assertRaises(TypeError):
	47	notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=23)
	48
	49	def test_create_wrong_begin_type(self):
88fdcc33	50	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	51
	52	with self.assertRaises(TypeError):
	53	notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin='hi')
	54
	55	def test_create_wrong_end_type(self):
88fdcc33	56	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	57
	58	with self.assertRaises(TypeError):
	59	notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin='lel')
	60
	61	def test_create_no_such_plugin(self):
88fdcc33	62	specs = [bt2.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	63
	64	with self.assertRaises(bt2.Error):
	65	notif_iter = bt2.TraceCollectionNotificationIterator(specs)
	66
	67	def test_create_begin_s(self):
88fdcc33	68	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	69	notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=19457.918232)
	70
	71	def test_create_end_s(self):
88fdcc33	72	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	73	notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=123.12312)
	74
	75	def test_create_begin_datetime(self):
88fdcc33	76	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	77	notif_iter = bt2.TraceCollectionNotificationIterator(specs, begin=datetime.datetime.now())
	78
	79	def test_create_end_datetime(self):
88fdcc33	80	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	81	notif_iter = bt2.TraceCollectionNotificationIterator(specs, end=datetime.datetime.now())
	82
	83	def test_iter_no_intersection(self):
88fdcc33	84	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	85	notif_iter = bt2.TraceCollectionNotificationIterator(specs)
	86	self.assertEqual(len(list(notif_iter)), 28)
	87
	88	def test_iter_no_intersection_subscribe(self):
88fdcc33	89	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	90	notif_iter = bt2.TraceCollectionNotificationIterator(specs,
	91	notification_types=[bt2.EventNotification])
	92	self.assertEqual(len(list(notif_iter)), 8)
	93
88fdcc33 PP	94	def test_iter_specs_not_list(self):
	95	spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
	96	notif_iter = bt2.TraceCollectionNotificationIterator(spec,
	97	notification_types=[bt2.EventNotification])
	98	self.assertEqual(len(list(notif_iter)), 8)
	99
	100	def test_iter_custom_filter(self):
	101	src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
	102	flt_spec = bt2.ComponentSpec('utils', 'trimmer', {
	103	'end': 13515309000000075,
	104	})
	105	notif_iter = bt2.TraceCollectionNotificationIterator(src_spec, flt_spec,
	106	notification_types=[bt2.EventNotification])
	107	self.assertEqual(len(list(notif_iter)), 5)
	108
cc261e29	109	def test_iter_intersection(self):
88fdcc33	110	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	111	notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True)
	112	self.assertEqual(len(list(notif_iter)), 15)
	113
	114	def test_iter_intersection_subscribe(self):
88fdcc33	115	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	116	notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True,
	117	notification_types=[bt2.EventNotification])
	118	self.assertEqual(len(list(notif_iter)), 3)
	119
	120	def test_iter_intersection_no_path_param(self):
88fdcc33	121	specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
cc261e29 PP	122
	123	with self.assertRaises(bt2.Error):
	124	notif_iter = bt2.TraceCollectionNotificationIterator(specs, stream_intersection_mode=True,
	125	notification_types=[bt2.EventNotification])
	126
	127	def test_iter_no_intersection_two_traces(self):
88fdcc33	128	spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
cc261e29 PP	129	specs = [spec, spec]
	130	notif_iter = bt2.TraceCollectionNotificationIterator(specs)
	131	self.assertEqual(len(list(notif_iter)), 56)
	132
	133	def test_iter_no_intersection_begin(self):
88fdcc33	134	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	135	notif_iter = bt2.TraceCollectionNotificationIterator(specs,
	136	notification_types=[bt2.EventNotification],
	137	begin=13515309.000000023)
	138	self.assertEqual(len(list(notif_iter)), 6)
	139
	140	def test_iter_no_intersection_end(self):
88fdcc33	141	specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
cc261e29 PP	142	notif_iter = bt2.TraceCollectionNotificationIterator(specs,
	143	notification_types=[bt2.EventNotification],
	144	end=13515309.000000075)
	145	self.assertEqual(len(list(notif_iter)), 5)