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

from bt2 import value
import collections
import unittest
import copy
import bt2


class UserMessageIteratorTestCase(unittest.TestCase):
    @staticmethod
    def _create_graph(src_comp_cls):
        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                next(self._msg_iter)

            def _graph_is_configured(self):
                self._msg_iter = self._input_ports['in'].create_message_iterator()

        graph = bt2.Graph()
        src_comp = graph.add_component(src_comp_cls, 'src')
        sink_comp = graph.add_component(MySink, 'sink')
        graph.connect_ports(src_comp.output_ports['out'],
                            sink_comp.input_ports['in'])
        return graph

    def test_init(self):
        the_output_port_from_source = None
        the_output_port_from_iter = None

        class MyIter(bt2._UserMessageIterator):
            def __init__(self, self_port_output):
                nonlocal initialized
                nonlocal the_output_port_from_iter
                initialized = True
                the_output_port_from_iter = self_port_output

        class MySource(bt2._UserSourceComponent,
                       message_iterator_class=MyIter):
            def __init__(self, params):
                nonlocal the_output_port_from_source
                the_output_port_from_source = self._add_output_port('out', 'user data')

        initialized = False
        graph = self._create_graph(MySource)
        graph.run()
        self.assertTrue(initialized)
        self.assertEqual(the_output_port_from_source.addr, the_output_port_from_iter.addr)
        self.assertEqual(the_output_port_from_iter.user_data, 'user data')

    def test_finalize(self):
        class MyIter(bt2._UserMessageIterator):
            def _finalize(self):
                nonlocal finalized
                finalized = True

        class MySource(bt2._UserSourceComponent,
                       message_iterator_class=MyIter):
            def __init__(self, params):
                self._add_output_port('out')

        finalized = False
        graph = self._create_graph(MySource)
        graph.run()
        del graph
        self.assertTrue(finalized)

    def test_component(self):
        class MyIter(bt2._UserMessageIterator):
            def __init__(self, self_port_output):
                nonlocal salut
                salut = self._component._salut

        class MySource(bt2._UserSourceComponent,
                       message_iterator_class=MyIter):
            def __init__(self, params):
                self._add_output_port('out')
                self._salut = 23

        salut = None
        graph = self._create_graph(MySource)
        graph.run()
        self.assertEqual(salut, 23)

    def test_addr(self):
        class MyIter(bt2._UserMessageIterator):
            def __init__(self, self_port_output):
                nonlocal addr
                addr = self.addr

        class MySource(bt2._UserSourceComponent,
                       message_iterator_class=MyIter):
            def __init__(self, params):
                self._add_output_port('out')

        addr = None
        graph = self._create_graph(MySource)
        graph.run()
        self.assertIsNotNone(addr)
        self.assertNotEqual(addr, 0)


class OutputPortMessageIteratorTestCase(unittest.TestCase):
    def test_component(self):
        class MyIter(bt2._UserMessageIterator):
            def __init__(self, self_port_output):
                self._at = 0

            def __next__(self):
                if self._at == 7:
                    raise bt2.Stop

                if self._at == 0:
                    msg = self._create_stream_beginning_message(test_obj._stream)
                elif self._at == 1:
                    msg = self._create_packet_beginning_message(test_obj._packet)
                elif self._at == 5:
                    msg = self._create_packet_end_message(test_obj._packet)
                elif self._at == 6:
                    msg = self._create_stream_end_message(test_obj._stream)
                else:
                    msg = self._create_event_message(test_obj._event_class, test_obj._packet)
                    msg.event.payload_field['my_int'] = self._at * 3

                self._at += 1
                return msg

        class MySource(bt2._UserSourceComponent,
                       message_iterator_class=MyIter):
            def __init__(self, params):
                self._add_output_port('out')

                trace_class = self._create_trace_class()
                stream_class = trace_class.create_stream_class()

                # Create payload field class
                my_int_ft = trace_class.create_signed_integer_field_class(32)
                payload_ft = trace_class.create_structure_field_class()
                payload_ft += collections.OrderedDict([
                    ('my_int', my_int_ft),
                ])

                event_class = stream_class.create_event_class(name='salut', payload_field_class=payload_ft)

                trace = trace_class()
                stream = trace.create_stream(stream_class)
                packet = stream.create_packet()

                test_obj._event_class = event_class
                test_obj._stream = stream
                test_obj._packet = packet

        test_obj = self
        graph = bt2.Graph()
        src = graph.add_component(MySource, 'src')
        msg_iter = graph.create_output_port_message_iterator(src.output_ports['out'])

        for at, msg in enumerate(msg_iter):
            if at == 0:
                self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
            elif at == 1:
                self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
            elif at == 5:
                self.assertIsInstance(msg, bt2.message._PacketEndMessage)
            elif at == 6:
                self.assertIsInstance(msg, bt2.message._StreamEndMessage)
            else:
                self.assertIsInstance(msg, bt2.message._EventMessage)
                self.assertEqual(msg.event.cls.name, 'salut')
                field = msg.event.payload_field['my_int']
                self.assertEqual(field, at * 3)
Commit	Line	Data
ae0bfae8	1	from bt2 import value
0d0edb5e	2	import collections
f6a5e476 PP	3	import unittest
	4	import copy
	5	import bt2
	6
	7
fa4c33e3	8	class UserMessageIteratorTestCase(unittest.TestCase):
f6a5e476 PP	9	@staticmethod
	10	def _create_graph(src_comp_cls):
	11	class MySink(bt2._UserSinkComponent):
	12	def __init__(self, params):
	13	self._add_input_port('in')
	14
	15	def _consume(self):
fa4c33e3	16	next(self._msg_iter)
f6a5e476	17
a4dcfa96 SM	18	def _graph_is_configured(self):
a4dcfa96 SM	19	self._msg_iter = self._input_ports['in'].create_message_iterator()
f6a5e476 PP	20
	21	graph = bt2.Graph()
	22	src_comp = graph.add_component(src_comp_cls, 'src')
	23	sink_comp = graph.add_component(MySink, 'sink')
	24	graph.connect_ports(src_comp.output_ports['out'],
	25	sink_comp.input_ports['in'])
	26	return graph
	27
	28	def test_init(self):
a4dcfa96 SM	29	the_output_port_from_source = None
	30	the_output_port_from_iter = None
	31
fa4c33e3	32	class MyIter(bt2._UserMessageIterator):
a4dcfa96	33	def __init__(self, self_port_output):
f6a5e476	34	nonlocal initialized
a4dcfa96	35	nonlocal the_output_port_from_iter
f6a5e476	36	initialized = True
a4dcfa96	37	the_output_port_from_iter = self_port_output
f6a5e476 PP	38
f6a5e476 PP	39	class MySource(bt2._UserSourceComponent,
fa4c33e3	40	message_iterator_class=MyIter):
f6a5e476	41	def __init__(self, params):
a4dcfa96	42	nonlocal the_output_port_from_source
03ec9ebd	43	the_output_port_from_source = self._add_output_port('out', 'user data')
f6a5e476 PP	44
	45	initialized = False
	46	graph = self._create_graph(MySource)
a4dcfa96	47	graph.run()
f6a5e476	48	self.assertTrue(initialized)
a4dcfa96	49	self.assertEqual(the_output_port_from_source.addr, the_output_port_from_iter.addr)
03ec9ebd	50	self.assertEqual(the_output_port_from_iter.user_data, 'user data')
f6a5e476 PP	51
f6a5e476 PP	52	def test_finalize(self):
fa4c33e3	53	class MyIter(bt2._UserMessageIterator):
f6a5e476 PP	54	def _finalize(self):
	55	nonlocal finalized
	56	finalized = True
	57
	58	class MySource(bt2._UserSourceComponent,
fa4c33e3	59	message_iterator_class=MyIter):
f6a5e476 PP	60	def __init__(self, params):
	61	self._add_output_port('out')
	62
	63	finalized = False
	64	graph = self._create_graph(MySource)
a4dcfa96	65	graph.run()
f6a5e476 PP	66	del graph
	67	self.assertTrue(finalized)
	68
	69	def test_component(self):
fa4c33e3	70	class MyIter(bt2._UserMessageIterator):
a4dcfa96	71	def __init__(self, self_port_output):
f6a5e476 PP	72	nonlocal salut
	73	salut = self._component._salut
	74
	75	class MySource(bt2._UserSourceComponent,
fa4c33e3	76	message_iterator_class=MyIter):
f6a5e476 PP	77	def __init__(self, params):
	78	self._add_output_port('out')
	79	self._salut = 23
	80
	81	salut = None
	82	graph = self._create_graph(MySource)
a4dcfa96	83	graph.run()
f6a5e476 PP	84	self.assertEqual(salut, 23)
	85
	86	def test_addr(self):
fa4c33e3	87	class MyIter(bt2._UserMessageIterator):
a4dcfa96	88	def __init__(self, self_port_output):
f6a5e476 PP	89	nonlocal addr
	90	addr = self.addr
	91
	92	class MySource(bt2._UserSourceComponent,
fa4c33e3	93	message_iterator_class=MyIter):
f6a5e476 PP	94	def __init__(self, params):
	95	self._add_output_port('out')
	96
	97	addr = None
	98	graph = self._create_graph(MySource)
a4dcfa96	99	graph.run()
f6a5e476 PP	100	self.assertIsNotNone(addr)
	101	self.assertNotEqual(addr, 0)
	102
	103
fa4c33e3	104	class OutputPortMessageIteratorTestCase(unittest.TestCase):
0d0edb5e	105	def test_component(self):
fa4c33e3	106	class MyIter(bt2._UserMessageIterator):
a4dcfa96	107	def __init__(self, self_port_output):
0d0edb5e PP	108	self._at = 0
0d0edb5e PP	109
0d0edb5e	110	def __next__(self):
a4dcfa96	111	if self._at == 7:
0d0edb5e PP	112	raise bt2.Stop
0d0edb5e PP	113
a4dcfa96 SM	114	if self._at == 0:
	115	msg = self._create_stream_beginning_message(test_obj._stream)
	116	elif self._at == 1:
	117	msg = self._create_packet_beginning_message(test_obj._packet)
	118	elif self._at == 5:
	119	msg = self._create_packet_end_message(test_obj._packet)
	120	elif self._at == 6:
	121	msg = self._create_stream_end_message(test_obj._stream)
	122	else:
	123	msg = self._create_event_message(test_obj._event_class, test_obj._packet)
	124	msg.event.payload_field['my_int'] = self._at * 3
	125
0d0edb5e	126	self._at += 1
fa4c33e3	127	return msg
0d0edb5e PP	128
0d0edb5e PP	129	class MySource(bt2._UserSourceComponent,
fa4c33e3	130	message_iterator_class=MyIter):
0d0edb5e PP	131	def __init__(self, params):
	132	self._add_output_port('out')
	133
a4dcfa96 SM	134	trace_class = self._create_trace_class()
	135	stream_class = trace_class.create_stream_class()
	136
	137	# Create payload field class
	138	my_int_ft = trace_class.create_signed_integer_field_class(32)
	139	payload_ft = trace_class.create_structure_field_class()
	140	payload_ft += collections.OrderedDict([
	141	('my_int', my_int_ft),
	142	])
	143
	144	event_class = stream_class.create_event_class(name='salut', payload_field_class=payload_ft)
	145
	146	trace = trace_class()
	147	stream = trace.create_stream(stream_class)
	148	packet = stream.create_packet()
	149
	150	test_obj._event_class = event_class
	151	test_obj._stream = stream
	152	test_obj._packet = packet
	153
	154	test_obj = self
0d0edb5e PP	155	graph = bt2.Graph()
0d0edb5e PP	156	src = graph.add_component(MySource, 'src')
a4dcfa96	157	msg_iter = graph.create_output_port_message_iterator(src.output_ports['out'])
0d0edb5e	158
fa4c33e3	159	for at, msg in enumerate(msg_iter):
a4dcfa96 SM	160	if at == 0:
	161	self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
	162	elif at == 1:
	163	self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
	164	elif at == 5:
	165	self.assertIsInstance(msg, bt2.message._PacketEndMessage)
	166	elif at == 6:
	167	self.assertIsInstance(msg, bt2.message._StreamEndMessage)
	168	else:
	169	self.assertIsInstance(msg, bt2.message._EventMessage)
c88be1c8	170	self.assertEqual(msg.event.cls.name, 'salut')
a4dcfa96 SM	171	field = msg.event.payload_field['my_int']
a4dcfa96 SM	172	self.assertEqual(field, at * 3)