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

from collections import OrderedDict
import unittest
import copy
import bt2.notification_iterator
import bt2


def _create_ec():
    # clock class
    cc = bt2.ClockClass('salut_clock')

    # event header
    eh = bt2.StructureFieldType()
    eh += OrderedDict((
        ('id', bt2.IntegerFieldType(8)),
        ('ts', bt2.IntegerFieldType(32, mapped_clock_class=cc)),
    ))

    # packet context
    pc = bt2.StructureFieldType()
    pc += OrderedDict((
        ('something', bt2.IntegerFieldType(8)),
    ))

    # stream class
    sc = bt2.StreamClass()
    sc.packet_context_field_type = pc
    sc.event_header_field_type = eh
    sc.event_context_field_type = None

    # event payload
    ep = bt2.StructureFieldType()
    ep += OrderedDict((
        ('mosquito', bt2.StringFieldType()),
    ))

    # event class
    event_class = bt2.EventClass('ec', id=0)
    event_class.context_field_type = None
    event_class.payload_field_type = ep
    sc.add_event_class(event_class)

    # packet header
    ph = bt2.StructureFieldType()
    ph += OrderedDict((
        ('magic', bt2.IntegerFieldType(32)),
        ('stream_id', bt2.IntegerFieldType(16)),
    ))

    # trace
    trace = bt2.Trace()
    trace.packet_header_field_type = ph
    trace.add_clock_class(cc)
    trace.add_stream_class(sc)
    return event_class


def _create_event(event_class, msg):
    ev = event_class()
    ev.header_field['id'] = 0
    ev.header_field['ts'] = 19487
    ev.payload_field['mosquito'] = msg
    return ev


def _create_packet(stream):
    packet = stream.create_packet()
    packet.header_field['magic'] = 0xc1fc1fc1
    packet.header_field['stream_id'] = 0
    packet.context_field['something'] = 194
    return packet


def _create_source():
    class MyIter(bt2.UserNotificationIterator):
        def __init__(self):
            self._event_class = self.component._event_class
            self._stream = self.component._stream
            self._packet = _create_packet(self._stream)
            self._at = 0
            self._cur_notif = None

        def _get(self):
            if self._cur_notif is None:
                raise bt2.Error('nothing here!')

            return self._cur_notif

        def _next(self):
            if self._at == 0:
                notif = bt2.BeginningOfPacketNotification(self._packet)
            elif self._at < 5:
                ev = _create_event(self._event_class, 'at {}'.format(self._at))
                ev.packet = self._packet
                notif = bt2.TraceEventNotification(ev)
            elif self._at == 5:
                notif = bt2.EndOfPacketNotification(self._packet)
            elif self._at == 6:
                notif = bt2.EndOfStreamNotification(self._stream)
            else:
                raise bt2.Stop

            self._at += 1
            self._cur_notif = notif

    class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
        def __init__(self):
            self._event_class = _create_ec()
            self._stream = self._event_class.stream_class(name='abcdef')

    return MySource()


class GenCompClassTestCase(unittest.TestCase):
    def test_attr_name(self):
        class MySink(bt2.UserSinkComponent):
            def _consume(self):
                pass

        sink = MySink()
        self.assertEqual(sink.component_class.name, 'MySink')

    def test_attr_description(self):
        class MySink(bt2.UserSinkComponent):
            '''This is the description.

            This is the help.

            This too:

                * And this.
                * And also that.

            Voilà.
            '''

            def _consume(self):
                pass

        sink = MySink()
        self.assertEqual(sink.component_class.description,
                         'This is the description.')

    def test_attr_help(self):
        class MySink(bt2.UserSinkComponent):
            '''This is the description.

            This is the help.

            This too:

                * And this.
                * And also that.

            Voilà.
            '''

            def _consume(self):
                pass

        sink = MySink()
        expected_help = '''This is the help.

This too:

    * And this.
    * And also that.

Voilà.'''
        self.assertEqual(sink.component_class.help, expected_help)

    def test_instantiate(self):
        class MySink(bt2.UserSinkComponent):
            'hello'

            def __init__(self, params=None, name=None):
                if params is None or name is None:
                    return

                nonlocal nl_params_a, nl_name
                nl_params_a = params['a']
                nl_name = name

            def _consume(self):
                pass

        nl_params_a = None
        nl_name = None
        sink = MySink()
        self.assertIsNone(nl_params_a)
        self.assertIsNone(nl_name)
        gen_comp_class = sink.component_class
        sink2 = gen_comp_class(params={'a': 23}, name='salut')
        self.assertEqual(nl_params_a, 23)
        self.assertEqual(nl_name, 'salut')


class UserCompClassTestCase(unittest.TestCase):
    def test_attr_name(self):
        class MySink(bt2.UserSinkComponent):
            def _consume(self):
                pass

        self.assertEqual(MySink.name, 'MySink')

    def test_attr_name_param(self):
        class MySink(bt2.UserSinkComponent, name='my custom name'):
            def _consume(self):
                pass

        self.assertEqual(MySink.name, 'my custom name')

    def test_attr_description(self):
        class MySink(bt2.UserSinkComponent):
            '''This is the description.

            This is the help.

            This too:

                * And this.
                * And also that.

            Voilà.
            '''

            def _consume(self):
                pass

        self.assertEqual(MySink.description, 'This is the description.')

    def test_attr_help(self):
        class MySink(bt2.UserSinkComponent):
            '''This is the description.

            This is the help.

            This too:

                * And this.
                * And also that.

            Voilà.
            '''

            def _consume(self):
                pass

        expected_help = '''This is the help.

This too:

    * And this.
    * And also that.

Voilà.'''
        self.assertEqual(MySink.help, expected_help)

    def test_init(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                nonlocal inited
                inited = True

            def _consume(self):
                pass

        inited = False
        sink = MySink()
        self.assertTrue(inited)

    def test_init_raises(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                raise RuntimeError('oops')

            def _consume(self):
                pass

        with self.assertRaises(RuntimeError):
            sink = MySink()

    def test_destroy(self):
        class MySink(bt2.UserSinkComponent):
            def _destroy(self):
                nonlocal destroyed
                destroyed = True

            def _consume(self):
                pass

        destroyed = False
        sink = MySink()
        del sink
        self.assertTrue(destroyed)

    def test_destroy_raises(self):
        class MySink(bt2.UserSinkComponent):
            def _destroy(self):
                raise RuntimeError('oh oh')
                nonlocal destroyed
                destroyed = True

            def _consume(self):
                pass

        destroyed = False
        sink = MySink()
        del sink
        self.assertFalse(destroyed)

    def test_comp_attr_name(self):
        class MySink(bt2.UserSinkComponent):
            def _consume(self):
                pass

        sink = MySink(name='salut')
        self.assertEqual(sink.name, 'salut')

    def test_comp_attr_no_name(self):
        class MySink(bt2.UserSinkComponent):
            def _consume(self):
                pass

        sink = MySink()
        self.assertIsNone(sink.name)

    def test_comp_attr_class(self):
        class MySink(bt2.UserSinkComponent):
            def _consume(self):
                pass

        sink = MySink()
        self.assertEqual(sink.component_class.name, 'MySink')


class UserSinkCompClassTestCase(unittest.TestCase):
    def test_missing_consume(self):
        with self.assertRaises(bt2.IncompleteUserClassError):
            class MySink(bt2.UserSinkComponent):
                pass

    def test_set_min_input(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                self._maximum_input_notification_iterator_count = 10
                self._minimum_input_notification_iterator_count = 5

            def _consume(self):
                pass

        sink = MySink()

    def test_set_max_input(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                self._maximum_input_notification_iterator_count = 5

            def _consume(self):
                pass

        sink = MySink()

    def test_consume(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                pass

            def _consume(self):
                nonlocal consumed
                consumed = True

        sink = MySink()
        consumed = False
        source = _create_source()
        notif_iter = source.create_notification_iterator()
        sink.add_notification_iterator(notif_iter)
        sink.consume()
        self.assertTrue(consumed)

    def test_consume_raises(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                pass

            def _consume(self):
                raise RuntimeError('hello')

        sink = MySink()
        source = _create_source()
        notif_iter = source.create_notification_iterator()
        sink.add_notification_iterator(notif_iter)

        with self.assertRaises(bt2.Error):
            sink.consume()

    def test_add_notification_iterator(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                pass

            def _consume(self):
                pass

            def _add_notification_iterator(self, notif_iter):
                nonlocal added
                added = True

        sink = MySink()
        added = False
        source = _create_source()
        notif_iter = source.create_notification_iterator()
        sink.add_notification_iterator(notif_iter)
        self.assertTrue(added)

    def test_input_notif_iterators(self):
        class MySink(bt2.UserSinkComponent):
            def __init__(self):
                self._maximum_input_notification_iterator_count = 5

            def _consume(self):
                nonlocal count
                count = 0

                for notif_iter in self._input_notification_iterators:
                    count += 1

        sink = MySink()
        count = 0
        source = _create_source()
        notif_iter1 = source.create_notification_iterator()
        notif_iter2 = source.create_notification_iterator()
        sink.add_notification_iterator(notif_iter1)
        sink.add_notification_iterator(notif_iter2)
        sink.consume()
        self.assertEqual(count, 2)


class UserSourceCompClassTestCase(unittest.TestCase):
    def test_missing_notif_iterator_class(self):
        with self.assertRaises(bt2.IncompleteUserClassError):
            class MySource(bt2.UserSourceComponent):
                pass

    def test_invalid_notif_iterator_class(self):
        class Lel:
            pass

        with self.assertRaises(bt2.IncompleteUserClassError):
            class MySource(bt2.UserSourceComponent, notification_iterator_class=Lel):
                pass

    def test_notif_iterator_class_missing_get(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

        with self.assertRaises(bt2.IncompleteUserClassError):
            class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
                pass

    def test_notif_iterator_class_missing_next(self):
        class MyIter(bt2.UserNotificationIterator):
            def _get(self):
                pass

        with self.assertRaises(bt2.IncompleteUserClassError):
            class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
                pass

    def test_create_notification_iterator(self):
        class MyIter(bt2.UserNotificationIterator):
            def __init__(self):
                nonlocal created
                created = True

            def _next(self):
                pass

            def _get(self):
                pass

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        created = False
        source = MySource()
        notif_iter = source.create_notification_iterator()
        self.assertTrue(created)


class UserSourceCompClassTestCase(unittest.TestCase):
    def test_missing_notif_iterator_class(self):
        with self.assertRaises(bt2.IncompleteUserClassError):
            class MyFilter(bt2.UserFilterComponent):
                pass

    def test_invalid_notif_iterator_class(self):
        class Lel:
            pass

        with self.assertRaises(bt2.IncompleteUserClassError):
            class MyFilter(bt2.UserFilterComponent, notification_iterator_class=Lel):
                pass

    def test_notif_iterator_class_missing_get(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

        with self.assertRaises(bt2.IncompleteUserClassError):
            class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
                pass

    def test_notif_iterator_class_missing_next(self):
        class MyIter(bt2.UserNotificationIterator):
            def _get(self):
                pass

        with self.assertRaises(bt2.IncompleteUserClassError):
            class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
                pass

    def test_create_notification_iterator(self):
        class MyIter(bt2.UserNotificationIterator):
            def __init__(self):
                nonlocal created
                created = True

            def _next(self):
                pass

            def _get(self):
                pass

        class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
            pass

        created = False
        filter = MyFilter()
        notif_iter = filter.create_notification_iterator()
        self.assertTrue(created)

    def test_set_min_input(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

            def _get(self):
                pass

        class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
            def __init__(self):
                self._maximum_input_notification_iterator_count = 10
                self._minimum_input_notification_iterator_count = 5

        filter = MyFilter()

    def test_set_max_input(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

            def _get(self):
                pass

        class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
            def __init__(self):
                self._maximum_input_notification_iterator_count = 5

        filter = MyFilter()

    def test_add_notification_iterator(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

            def _get(self):
                pass

        class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
            def __init__(self):
                pass

            def _add_notification_iterator(self, notif_iter):
                nonlocal added
                added = True

        filter = MyFilter()
        added = False
        source = _create_source()
        notif_iter = source.create_notification_iterator()
        filter.add_notification_iterator(notif_iter)
        self.assertTrue(added)

    def test_input_notif_iterators(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

            def _get(self):
                pass

        class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
            def __init__(self):
                self._maximum_input_notification_iterator_count = 5

            def _test(self):
                nonlocal count
                count = 0

                for notif_iter in self._input_notification_iterators:
                    count += 1

        filter = MyFilter()
        count = 0
        source = _create_source()
        notif_iter1 = source.create_notification_iterator()
        notif_iter2 = source.create_notification_iterator()
        filter.add_notification_iterator(notif_iter1)
        filter.add_notification_iterator(notif_iter2)
        filter._test()
        self.assertEqual(count, 2)


class UserNotificationIteratorTestCase(unittest.TestCase):
    def test_init(self):
        class MyIter(bt2.UserNotificationIterator):
            def __init__(self):
                nonlocal inited
                inited = True

            def _next(self):
                pass

            def _get(self):
                pass

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        inited = False
        source = MySource()
        notif_iter = source.create_notification_iterator()
        self.assertTrue(inited)

    def test_destroy(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                pass

            def _get(self):
                pass

            def _destroy(self):
                nonlocal destroyed
                destroyed = True

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        source = MySource()
        notif_iter = source.create_notification_iterator()
        destroyed = False
        del notif_iter
        self.assertTrue(destroyed)

    def test_attr_component(self):
        class MyIter(bt2.UserNotificationIterator):
            def __init__(self):
                nonlocal source, is_same
                is_same = self.component is source

            def _next(self):
                pass

            def _get(self):
                pass

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        source = MySource()
        is_same = False
        notif_iter = source.create_notification_iterator()
        self.assertTrue(is_same)

    def test_next_raises_stop(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                raise bt2.Stop

            def _get(self):
                pass

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        source = MySource()
        is_same = False
        notif_iter = source.create_notification_iterator()

        with self.assertRaises(bt2.Stop):
            notif_iter.next()

    def test_next_raises_unsupported_feature(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                raise bt2.UnsupportedFeature

            def _get(self):
                pass

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        source = MySource()
        is_same = False
        notif_iter = source.create_notification_iterator()

        with self.assertRaises(bt2.UnsupportedFeature):
            notif_iter.next()

    def test_next_raises_unknown(self):
        class MyIter(bt2.UserNotificationIterator):
            def _next(self):
                raise TypeError('lel')

            def _get(self):
                pass

        class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
            pass

        source = MySource()
        is_same = False
        notif_iter = source.create_notification_iterator()

        with self.assertRaises(bt2.Error):
            notif_iter.next()

    def test_iteration(self):
        source = _create_source()
        notif_iter = source.create_notification_iterator()
        pkt_addr = None

        for index, notif in enumerate(notif_iter):
            if index == 0:
                self.assertIsInstance(notif, bt2.BeginningOfPacketNotification)
                self.assertEqual(notif.packet.header_field['magic'], 0xc1fc1fc1)
                self.assertEqual(notif.packet.context_field['something'], 194)
                pkt_addr = notif.packet.addr
            elif index < 5:
                self.assertIsInstance(notif, bt2.TraceEventNotification)
                self.assertEqual(notif.event.header_field['ts'], 19487)
                self.assertEqual(notif.event.payload_field['mosquito'], 'at {}'.format(index))
            elif index == 5:
                self.assertIsInstance(notif, bt2.EndOfPacketNotification)
                self.assertEqual(notif.packet.header_field['magic'], 0xc1fc1fc1)
                self.assertEqual(notif.packet.context_field['something'], 194)
                self.assertEqual(notif.packet.addr, pkt_addr)
            elif index == 6:
                self.assertIsInstance(notif, bt2.EndOfStreamNotification)
                self.assertEqual(notif.stream.name, 'abcdef')
            else:
                raise RuntimeError('FAIL')


class GenNotificationIteratorTestCase(unittest.TestCase):
    def test_attr_component(self):
        source = _create_source()
        notif_iter = source.create_notification_iterator()
        self.assertIsInstance(notif_iter, bt2.notification_iterator._GenericNotificationIterator)
        self.assertEqual(notif_iter.component.addr, source.addr)
Commit	Line	Data
9cf643d1 PP	1	from collections import OrderedDict
	2	import unittest
	3	import copy
	4	import bt2.notification_iterator
	5	import bt2
	6
	7
	8	def _create_ec():
	9	# clock class
	10	cc = bt2.ClockClass('salut_clock')
	11
	12	# event header
	13	eh = bt2.StructureFieldType()
	14	eh += OrderedDict((
	15	('id', bt2.IntegerFieldType(8)),
	16	('ts', bt2.IntegerFieldType(32, mapped_clock_class=cc)),
	17	))
	18
	19	# packet context
	20	pc = bt2.StructureFieldType()
	21	pc += OrderedDict((
	22	('something', bt2.IntegerFieldType(8)),
	23	))
	24
	25	# stream class
	26	sc = bt2.StreamClass()
	27	sc.packet_context_field_type = pc
	28	sc.event_header_field_type = eh
	29	sc.event_context_field_type = None
	30
	31	# event payload
	32	ep = bt2.StructureFieldType()
	33	ep += OrderedDict((
	34	('mosquito', bt2.StringFieldType()),
	35	))
	36
	37	# event class
	38	event_class = bt2.EventClass('ec', id=0)
	39	event_class.context_field_type = None
	40	event_class.payload_field_type = ep
	41	sc.add_event_class(event_class)
	42
	43	# packet header
	44	ph = bt2.StructureFieldType()
	45	ph += OrderedDict((
	46	('magic', bt2.IntegerFieldType(32)),
	47	('stream_id', bt2.IntegerFieldType(16)),
	48	))
	49
	50	# trace
	51	trace = bt2.Trace()
	52	trace.packet_header_field_type = ph
	53	trace.add_clock_class(cc)
	54	trace.add_stream_class(sc)
	55	return event_class
	56
	57
	58	def _create_event(event_class, msg):
	59	ev = event_class()
	60	ev.header_field['id'] = 0
	61	ev.header_field['ts'] = 19487
	62	ev.payload_field['mosquito'] = msg
	63	return ev
	64
65
66	def _create_packet(stream):
67	packet = stream.create_packet()
68	packet.header_field['magic'] = 0xc1fc1fc1
69	packet.header_field['stream_id'] = 0
70	packet.context_field['something'] = 194
71	return packet
72
73
74	def _create_source():
75	class MyIter(bt2.UserNotificationIterator):
76	def __init__(self):
77	self._event_class = self.component._event_class
78	self._stream = self.component._stream
79	self._packet = _create_packet(self._stream)
80	self._at = 0
81	self._cur_notif = None
82
83	def _get(self):
84	if self._cur_notif is None:
85	raise bt2.Error('nothing here!')
86
87	return self._cur_notif
88
89	def _next(self):
90	if self._at == 0:
91	notif = bt2.BeginningOfPacketNotification(self._packet)
92	elif self._at < 5:
93	ev = _create_event(self._event_class, 'at {}'.format(self._at))
94	ev.packet = self._packet
95	notif = bt2.TraceEventNotification(ev)
96	elif self._at == 5:
97	notif = bt2.EndOfPacketNotification(self._packet)
98	elif self._at == 6:
99	notif = bt2.EndOfStreamNotification(self._stream)
100	else:
101	raise bt2.Stop
102
103	self._at += 1
104	self._cur_notif = notif
105
106	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
107	def __init__(self):
108	self._event_class = _create_ec()
109	self._stream = self._event_class.stream_class(name='abcdef')
110
111	return MySource()
112
113
114	class GenCompClassTestCase(unittest.TestCase):
115	def test_attr_name(self):
116	class MySink(bt2.UserSinkComponent):
117	def _consume(self):
118	pass
119
120	sink = MySink()
121	self.assertEqual(sink.component_class.name, 'MySink')
122
123	def test_attr_description(self):
124	class MySink(bt2.UserSinkComponent):
feb5fe23 PP	125	'''This is the description.
	126
	127	This is the help.
	128
	129	This too:
	130
	131	* And this.
	132	* And also that.
	133
	134	Voilà.
	135	'''
	136
	137	def _consume(self):
	138	pass
	139
	140	sink = MySink()
	141	self.assertEqual(sink.component_class.description,
	142	'This is the description.')
	143
	144	def test_attr_help(self):
	145	class MySink(bt2.UserSinkComponent):
	146	'''This is the description.
	147
	148	This is the help.
	149
	150	This too:
	151
	152	* And this.
	153	* And also that.
	154
	155	Voilà.
	156	'''
9cf643d1 PP	157
	158	def _consume(self):
	159	pass
	160
	161	sink = MySink()
feb5fe23 PP	162	expected_help = '''This is the help.
	163
	164	This too:
	165
	166	* And this.
	167	* And also that.
	168
	169	Voilà.'''
	170	self.assertEqual(sink.component_class.help, expected_help)
9cf643d1 PP	171
	172	def test_instantiate(self):
	173	class MySink(bt2.UserSinkComponent):
	174	'hello'
	175
	176	def __init__(self, params=None, name=None):
	177	if params is None or name is None:
	178	return
	179
	180	nonlocal nl_params_a, nl_name
	181	nl_params_a = params['a']
	182	nl_name = name
	183
	184	def _consume(self):
	185	pass
	186
	187	nl_params_a = None
	188	nl_name = None
	189	sink = MySink()
	190	self.assertIsNone(nl_params_a)
	191	self.assertIsNone(nl_name)
	192	gen_comp_class = sink.component_class
	193	sink2 = gen_comp_class(params={'a': 23}, name='salut')
	194	self.assertEqual(nl_params_a, 23)
	195	self.assertEqual(nl_name, 'salut')
	196
	197
	198	class UserCompClassTestCase(unittest.TestCase):
	199	def test_attr_name(self):
	200	class MySink(bt2.UserSinkComponent):
	201	def _consume(self):
	202	pass
	203
	204	self.assertEqual(MySink.name, 'MySink')
	205
	206	def test_attr_name_param(self):
	207	class MySink(bt2.UserSinkComponent, name='my custom name'):
	208	def _consume(self):
	209	pass
	210
	211	self.assertEqual(MySink.name, 'my custom name')
	212
	213	def test_attr_description(self):
	214	class MySink(bt2.UserSinkComponent):
feb5fe23 PP	215	'''This is the description.
	216
	217	This is the help.
	218
	219	This too:
	220
	221	* And this.
	222	* And also that.
	223
	224	Voilà.
	225	'''
	226
	227	def _consume(self):
	228	pass
	229
	230	self.assertEqual(MySink.description, 'This is the description.')
	231
	232	def test_attr_help(self):
	233	class MySink(bt2.UserSinkComponent):
	234	'''This is the description.
	235
	236	This is the help.
	237
	238	This too:
	239
	240	* And this.
	241	* And also that.
	242
	243	Voilà.
	244	'''
9cf643d1 PP	245
	246	def _consume(self):
	247	pass
	248
feb5fe23 PP	249	expected_help = '''This is the help.
	250
	251	This too:
	252
	253	* And this.
	254	* And also that.
	255
	256	Voilà.'''
	257	self.assertEqual(MySink.help, expected_help)
9cf643d1 PP	258
	259	def test_init(self):
	260	class MySink(bt2.UserSinkComponent):
	261	def __init__(self):
	262	nonlocal inited
	263	inited = True
	264
	265	def _consume(self):
	266	pass
	267
	268	inited = False
	269	sink = MySink()
	270	self.assertTrue(inited)
	271
	272	def test_init_raises(self):
	273	class MySink(bt2.UserSinkComponent):
	274	def __init__(self):
	275	raise RuntimeError('oops')
	276
	277	def _consume(self):
	278	pass
	279
	280	with self.assertRaises(RuntimeError):
	281	sink = MySink()
	282
	283	def test_destroy(self):
	284	class MySink(bt2.UserSinkComponent):
	285	def _destroy(self):
	286	nonlocal destroyed
	287	destroyed = True
	288
	289	def _consume(self):
	290	pass
	291
	292	destroyed = False
	293	sink = MySink()
	294	del sink
	295	self.assertTrue(destroyed)
	296
	297	def test_destroy_raises(self):
	298	class MySink(bt2.UserSinkComponent):
	299	def _destroy(self):
	300	raise RuntimeError('oh oh')
	301	nonlocal destroyed
	302	destroyed = True
	303
	304	def _consume(self):
	305	pass
	306
	307	destroyed = False
	308	sink = MySink()
	309	del sink
	310	self.assertFalse(destroyed)
	311
	312	def test_comp_attr_name(self):
	313	class MySink(bt2.UserSinkComponent):
	314	def _consume(self):
	315	pass
	316
	317	sink = MySink(name='salut')
	318	self.assertEqual(sink.name, 'salut')
	319
	320	def test_comp_attr_no_name(self):
	321	class MySink(bt2.UserSinkComponent):
322	def _consume(self):
323	pass
324
325	sink = MySink()
326	self.assertIsNone(sink.name)
327
328	def test_comp_attr_class(self):
329	class MySink(bt2.UserSinkComponent):
330	def _consume(self):
331	pass
332
333	sink = MySink()
334	self.assertEqual(sink.component_class.name, 'MySink')
335
336
337	class UserSinkCompClassTestCase(unittest.TestCase):
338	def test_missing_consume(self):
339	with self.assertRaises(bt2.IncompleteUserClassError):
340	class MySink(bt2.UserSinkComponent):
341	pass
342
343	def test_set_min_input(self):
344	class MySink(bt2.UserSinkComponent):
345	def __init__(self):
346	self._maximum_input_notification_iterator_count = 10
347	self._minimum_input_notification_iterator_count = 5
348
349	def _consume(self):
350	pass
351
352	sink = MySink()
353
354	def test_set_max_input(self):
355	class MySink(bt2.UserSinkComponent):
356	def __init__(self):
357	self._maximum_input_notification_iterator_count = 5
358
359	def _consume(self):
360	pass
361
362	sink = MySink()
363
364	def test_consume(self):
365	class MySink(bt2.UserSinkComponent):
366	def __init__(self):
367	pass
368
369	def _consume(self):
370	nonlocal consumed
371	consumed = True
372
373	sink = MySink()
374	consumed = False
375	source = _create_source()
376	notif_iter = source.create_notification_iterator()
377	sink.add_notification_iterator(notif_iter)
378	sink.consume()
379	self.assertTrue(consumed)
380
381	def test_consume_raises(self):
382	class MySink(bt2.UserSinkComponent):
383	def __init__(self):
384	pass
385
386	def _consume(self):
387	raise RuntimeError('hello')
388
389	sink = MySink()
390	source = _create_source()
391	notif_iter = source.create_notification_iterator()
392	sink.add_notification_iterator(notif_iter)
393
394	with self.assertRaises(bt2.Error):
395	sink.consume()
396
397	def test_add_notification_iterator(self):
398	class MySink(bt2.UserSinkComponent):
399	def __init__(self):
400	pass
401
402	def _consume(self):
403	pass
404
405	def _add_notification_iterator(self, notif_iter):
406	nonlocal added
407	added = True
408
409	sink = MySink()
410	added = False
411	source = _create_source()
412	notif_iter = source.create_notification_iterator()
413	sink.add_notification_iterator(notif_iter)
414	self.assertTrue(added)
415
416	def test_input_notif_iterators(self):
417	class MySink(bt2.UserSinkComponent):
418	def __init__(self):
419	self._maximum_input_notification_iterator_count = 5
420
421	def _consume(self):
422	nonlocal count
423	count = 0
424
425	for notif_iter in self._input_notification_iterators:
426	count += 1
427
428	sink = MySink()
429	count = 0
430	source = _create_source()
431	notif_iter1 = source.create_notification_iterator()
432	notif_iter2 = source.create_notification_iterator()
433	sink.add_notification_iterator(notif_iter1)
434	sink.add_notification_iterator(notif_iter2)
435	sink.consume()
436	self.assertEqual(count, 2)
437
438
439	class UserSourceCompClassTestCase(unittest.TestCase):
440	def test_missing_notif_iterator_class(self):
441	with self.assertRaises(bt2.IncompleteUserClassError):
442	class MySource(bt2.UserSourceComponent):
443	pass
444
445	def test_invalid_notif_iterator_class(self):
446	class Lel:
447	pass
448
449	with self.assertRaises(bt2.IncompleteUserClassError):
450	class MySource(bt2.UserSourceComponent, notification_iterator_class=Lel):
451	pass
452
453	def test_notif_iterator_class_missing_get(self):
454	class MyIter(bt2.UserNotificationIterator):
455	def _next(self):
456	pass
457
458	with self.assertRaises(bt2.IncompleteUserClassError):
459	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
460	pass
461
462	def test_notif_iterator_class_missing_next(self):
463	class MyIter(bt2.UserNotificationIterator):
464	def _get(self):
465	pass
466
467	with self.assertRaises(bt2.IncompleteUserClassError):
468	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
469	pass
470
471	def test_create_notification_iterator(self):
472	class MyIter(bt2.UserNotificationIterator):
473	def __init__(self):
474	nonlocal created
475	created = True
476
477	def _next(self):
478	pass
479
480	def _get(self):
481	pass
482
483	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
484	pass
485
486	created = False
487	source = MySource()
488	notif_iter = source.create_notification_iterator()
489	self.assertTrue(created)
490
491
492	class UserSourceCompClassTestCase(unittest.TestCase):
493	def test_missing_notif_iterator_class(self):
494	with self.assertRaises(bt2.IncompleteUserClassError):
495	class MyFilter(bt2.UserFilterComponent):
496	pass
497
498	def test_invalid_notif_iterator_class(self):
499	class Lel:
500	pass
501
502	with self.assertRaises(bt2.IncompleteUserClassError):
503	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=Lel):
504	pass
505
506	def test_notif_iterator_class_missing_get(self):
507	class MyIter(bt2.UserNotificationIterator):
508	def _next(self):
509	pass
510
511	with self.assertRaises(bt2.IncompleteUserClassError):
512	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
513	pass
514
515	def test_notif_iterator_class_missing_next(self):
516	class MyIter(bt2.UserNotificationIterator):
517	def _get(self):
518	pass
519
520	with self.assertRaises(bt2.IncompleteUserClassError):
521	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
522	pass
523
524	def test_create_notification_iterator(self):
525	class MyIter(bt2.UserNotificationIterator):
526	def __init__(self):
527	nonlocal created
528	created = True
529
530	def _next(self):
531	pass
532
533	def _get(self):
534	pass
535
536	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
537	pass
538
539	created = False
540	filter = MyFilter()
541	notif_iter = filter.create_notification_iterator()
542	self.assertTrue(created)
543
544	def test_set_min_input(self):
545	class MyIter(bt2.UserNotificationIterator):
546	def _next(self):
547	pass
548
549	def _get(self):
550	pass
551
552	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
553	def __init__(self):
554	self._maximum_input_notification_iterator_count = 10
555	self._minimum_input_notification_iterator_count = 5
556
557	filter = MyFilter()
558
559	def test_set_max_input(self):
560	class MyIter(bt2.UserNotificationIterator):
561	def _next(self):
562	pass
563
564	def _get(self):
565	pass
566
567	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
568	def __init__(self):
569	self._maximum_input_notification_iterator_count = 5
570
571	filter = MyFilter()
572
573	def test_add_notification_iterator(self):
574	class MyIter(bt2.UserNotificationIterator):
575	def _next(self):
576	pass
577
578	def _get(self):
579	pass
580
581	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
582	def __init__(self):
583	pass
584
585	def _add_notification_iterator(self, notif_iter):
586	nonlocal added
587	added = True
588
589	filter = MyFilter()
590	added = False
591	source = _create_source()
592	notif_iter = source.create_notification_iterator()
593	filter.add_notification_iterator(notif_iter)
594	self.assertTrue(added)
595
596	def test_input_notif_iterators(self):
597	class MyIter(bt2.UserNotificationIterator):
598	def _next(self):
599	pass
600
601	def _get(self):
602	pass
603
604	class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
605	def __init__(self):
606	self._maximum_input_notification_iterator_count = 5
607
608	def _test(self):
609	nonlocal count
610	count = 0
611
612	for notif_iter in self._input_notification_iterators:
613	count += 1
614
615	filter = MyFilter()
616	count = 0
617	source = _create_source()
618	notif_iter1 = source.create_notification_iterator()
619	notif_iter2 = source.create_notification_iterator()
620	filter.add_notification_iterator(notif_iter1)
621	filter.add_notification_iterator(notif_iter2)
622	filter._test()
623	self.assertEqual(count, 2)
624
625
626	class UserNotificationIteratorTestCase(unittest.TestCase):
627	def test_init(self):
628	class MyIter(bt2.UserNotificationIterator):
629	def __init__(self):
630	nonlocal inited
631	inited = True
632
633	def _next(self):
634	pass
635
636	def _get(self):
637	pass
638
639	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
640	pass
641
642	inited = False
643	source = MySource()
644	notif_iter = source.create_notification_iterator()
645	self.assertTrue(inited)
646
647	def test_destroy(self):
648	class MyIter(bt2.UserNotificationIterator):
649	def _next(self):
650	pass
651
652	def _get(self):
653	pass
654
655	def _destroy(self):
656	nonlocal destroyed
657	destroyed = True
658
659	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
660	pass
661
662	source = MySource()
663	notif_iter = source.create_notification_iterator()
664	destroyed = False
665	del notif_iter
666	self.assertTrue(destroyed)
667
668	def test_attr_component(self):
669	class MyIter(bt2.UserNotificationIterator):
670	def __init__(self):
671	nonlocal source, is_same
672	is_same = self.component is source
673
674	def _next(self):
675	pass
676
677	def _get(self):
678	pass
679
680	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
681	pass
682
683	source = MySource()
684	is_same = False
685	notif_iter = source.create_notification_iterator()
686	self.assertTrue(is_same)
687
688	def test_next_raises_stop(self):
689	class MyIter(bt2.UserNotificationIterator):
690	def _next(self):
691	raise bt2.Stop
692
693	def _get(self):
694	pass
695
696	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
697	pass
698
699	source = MySource()
700	is_same = False
701	notif_iter = source.create_notification_iterator()
702
703	with self.assertRaises(bt2.Stop):
704	notif_iter.next()
705
706	def test_next_raises_unsupported_feature(self):
707	class MyIter(bt2.UserNotificationIterator):
708	def _next(self):
709	raise bt2.UnsupportedFeature
710
711	def _get(self):
712	pass
713
714	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
715	pass
716
717	source = MySource()
718	is_same = False
719	notif_iter = source.create_notification_iterator()
720
721	with self.assertRaises(bt2.UnsupportedFeature):
722	notif_iter.next()
723
724	def test_next_raises_unknown(self):
725	class MyIter(bt2.UserNotificationIterator):
726	def _next(self):
727	raise TypeError('lel')
728
729	def _get(self):
730	pass
731
732	class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
733	pass
734
735	source = MySource()
736	is_same = False
737	notif_iter = source.create_notification_iterator()
738
739	with self.assertRaises(bt2.Error):
740	notif_iter.next()
741
742	def test_iteration(self):
743	source = _create_source()
744	notif_iter = source.create_notification_iterator()
745	pkt_addr = None
746
747	for index, notif in enumerate(notif_iter):
748	if index == 0:
749	self.assertIsInstance(notif, bt2.BeginningOfPacketNotification)
750	self.assertEqual(notif.packet.header_field['magic'], 0xc1fc1fc1)
751	self.assertEqual(notif.packet.context_field['something'], 194)
752	pkt_addr = notif.packet.addr
753	elif index < 5:
754	self.assertIsInstance(notif, bt2.TraceEventNotification)
755	self.assertEqual(notif.event.header_field['ts'], 19487)
756	self.assertEqual(notif.event.payload_field['mosquito'], 'at {}'.format(index))
757	elif index == 5:
758	self.assertIsInstance(notif, bt2.EndOfPacketNotification)
759	self.assertEqual(notif.packet.header_field['magic'], 0xc1fc1fc1)
760	self.assertEqual(notif.packet.context_field['something'], 194)
761	self.assertEqual(notif.packet.addr, pkt_addr)
762	elif index == 6:
763	self.assertIsInstance(notif, bt2.EndOfStreamNotification)
764	self.assertEqual(notif.stream.name, 'abcdef')
765	else:
766	raise RuntimeError('FAIL')
767
768
769	class GenNotificationIteratorTestCase(unittest.TestCase):
770	def test_attr_component(self):
771	source = _create_source()
772	notif_iter = source.create_notification_iterator()
773	self.assertIsInstance(notif_iter, bt2.notification_iterator._GenericNotificationIterator)
774	self.assertEqual(notif_iter.component.addr, source.addr)