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

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


class _MyIter(bt2._UserMessageIterator):
    def __init__(self, self_output_port):
        self._build_meta()
        self._at = 0

    def _build_meta(self):
        self._tc = self._component._create_trace_class()
        self._t = self._tc()
        self._sc = self._tc.create_stream_class()
        self._ec = self._sc.create_event_class(name='salut')
        self._my_int_ft = self._tc.create_signed_integer_field_class(32)
        payload_ft = self._tc.create_structure_field_class()
        payload_ft += collections.OrderedDict([
            ('my_int', self._my_int_ft),
        ])
        self._ec.payload_field_type = payload_ft
        self._stream = self._t.create_stream(self._sc)
        self._packet = self._stream.create_packet()

    def _create_event(self, value):
        ev = self._ec()
        ev.payload_field['my_int'] = value
        ev.packet = self._packet
        return ev


class GraphTestCase(unittest.TestCase):
    def setUp(self):
        self._graph = bt2.Graph()

    def tearDown(self):
        del self._graph

    def test_create_empty(self):
        graph = bt2.Graph()

    def test_add_component_user_cls(self):
        class MySink(bt2._UserSinkComponent):
            def _consume(self):
                pass

        comp = self._graph.add_component(MySink, 'salut')
        self.assertEqual(comp.name, 'salut')

    def test_add_component_gen_cls(self):
        class MySink(bt2._UserSinkComponent):
            def _consume(self):
                pass

        comp = self._graph.add_component(MySink, 'salut')
        assert comp
        comp2 = self._graph.add_component(comp.component_class, 'salut2')
        self.assertEqual(comp2.name, 'salut2')

    def test_add_component_params(self):
        comp_params = None

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                nonlocal comp_params
                comp_params = params

            def _consume(self):
                pass

        params = {'hello': 23, 'path': '/path/to/stuff'}
        comp = self._graph.add_component(MySink, 'salut', params)
        self.assertEqual(params, comp_params)
        del comp_params

    def test_add_component_invalid_cls_type(self):
        with self.assertRaises(TypeError):
            self._graph.add_component(int, 'salut')

    def test_connect_ports(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise bt2.Stop

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')

        conn = self._graph.connect_ports(src.output_ports['out'],
                                         sink.input_ports['in'])
        self.assertTrue(src.output_ports['out'].is_connected)
        self.assertTrue(sink.input_ports['in'].is_connected)
        self.assertEqual(src.output_ports['out'].connection._ptr, conn._ptr)
        self.assertEqual(sink.input_ports['in'].connection._ptr, conn._ptr)

    def test_connect_ports_invalid_direction(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise bt2.Stop

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')

        with self.assertRaises(TypeError):
            conn = self._graph.connect_ports(sink.input_ports['in'],
                                             src.output_ports['out'])

    def test_connect_ports_refused(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise bt2.Stop

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

            def _accept_port_connection(self, port, other_port):
                return False

        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')

        with self.assertRaises(bt2.PortConnectionRefused):
            conn = self._graph.connect_ports(src.output_ports['out'],
                                             sink.input_ports['in'])

    def test_cancel(self):
        self.assertFalse(self._graph.is_canceled)
        self._graph.cancel()
        self.assertTrue(self._graph.is_canceled)

    # Test that Graph.run() raises bt2.GraphCanceled if the graph gets canceled
    # during execution.
    def test_cancel_while_running(self):
        class MyIter(_MyIter):
            def __next__(self):
                return self._create_stream_beginning_message(self._stream)

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                # Pretend that somebody asynchronously cancelled the graph.
                nonlocal graph
                graph.cancel()

                return next(self._msg_iter)

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

        graph = bt2.Graph()
        up = graph.add_component(MySource, 'down')
        down = graph.add_component(MySink, 'up')
        graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
        with self.assertRaises(bt2.GraphCanceled):
            graph.run()

    def test_run(self):
        class MyIter(_MyIter):
            def __next__(self):
                if self._at == 9:
                    raise StopIteration

                if self._at == 0:
                    msg = self._create_stream_beginning_message(self._stream)
                elif self._at == 1:
                    msg = self._create_packet_beginning_message(self._packet)
                elif self._at == 7:
                    msg = self._create_packet_end_message(self._packet)
                elif self._at == 8:
                    msg = self._create_stream_end_message(self._stream)
                else:
                    msg = self._create_event_message(self._ec, self._packet)

                self._at += 1
                return msg

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._input_port = self._add_input_port('in')
                self._at = 0

            def _consume(comp_self):
                msg = next(comp_self._msg_iter)

                if comp_self._at == 0:
                    self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
                elif comp_self._at == 1:
                    self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
                elif comp_self._at >= 2 and comp_self._at <= 6:
                    self.assertIsInstance(msg, bt2.message._EventMessage)
                    self.assertEqual(msg.event.event_class.name, 'salut')
                elif comp_self._at == 7:
                    self.assertIsInstance(msg, bt2.message._PacketEndMessage)
                elif comp_self._at == 8:
                    self.assertIsInstance(msg, bt2.message._StreamEndMessage)

                comp_self._at += 1

            def _graph_is_configured(self):
                self._msg_iter = self._input_port.create_message_iterator()

        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')
        conn = self._graph.connect_ports(src.output_ports['out'],
                                         sink.input_ports['in'])
        self._graph.run()

    def test_run_again(self):
        class MyIter(_MyIter):
            def __next__(self):
                if self._at == 3:
                    raise bt2.TryAgain

                if self._at == 0:
                    msg = self._create_stream_beginning_message(self._stream)
                elif self._at == 1:
                    msg = self._create_packet_beginning_message(self._packet)
                elif self._at == 2:
                    msg = self._create_event_message(self._ec, self._packet)

                self._at += 1
                return msg

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._input_port = self._add_input_port('in')
                self._at = 0

            def _consume(comp_self):
                msg = next(comp_self._msg_iter)
                if comp_self._at == 0:
                    self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
                elif comp_self._at == 1:
                    self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
                elif comp_self._at == 2:
                    self.assertIsInstance(msg, bt2.message._EventMessage)
                    raise bt2.TryAgain
                else:
                    pass

                comp_self._at += 1

            def _graph_is_configured(self):
                self._msg_iter = self._input_port.create_message_iterator()

        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')
        conn = self._graph.connect_ports(src.output_ports['out'],
                                         sink.input_ports['in'])

        with self.assertRaises(bt2.TryAgain):
            self._graph.run()

    def test_run_error(self):
        raised_in_sink = False

        class MyIter(_MyIter):
            def __next__(self):
                # If this gets called after the sink raised an exception, it is
                # an error.
                nonlocal raised_in_sink
                assert raised_in_sink is False

                if self._at == 0:
                    msg = self._create_stream_beginning_message(self._stream)
                elif self._at == 1:
                    msg = self._create_packet_beginning_message(self._packet)
                elif self._at == 2 or self._at == 3:
                    msg = self._create_event_message(self._ec, self._packet)
                else:
                    raise bt2.TryAgain
                self._at += 1
                return msg

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._input_port = self._add_input_port('in')
                self._at = 0

            def _consume(comp_self):
                msg = next(comp_self._msg_iter)
                if comp_self._at == 0:
                    self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
                elif comp_self._at == 1:
                    self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
                elif comp_self._at == 2:
                    self.assertIsInstance(msg, bt2.message._EventMessage)
                elif comp_self._at == 3:
                    nonlocal raised_in_sink
                    raised_in_sink = True
                    raise RuntimeError('error!')

                comp_self._at += 1

            def _graph_is_configured(self):
                self._msg_iter = self._input_port.create_message_iterator()

        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')
        conn = self._graph.connect_ports(src.output_ports['out'],
                                         sink.input_ports['in'])

        with self.assertRaises(bt2.Error):
            self._graph.run()

    def test_listeners(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise bt2.Stop

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

            def _port_connected(self, port, other_port):
                self._add_input_port('taste')

        def port_added_listener(component, port):
            nonlocal calls
            calls.append((port_added_listener, component, port))

        def ports_connected_listener(upstream_component, upstream_port,
                                     downstream_component, downstream_port):
            nonlocal calls
            calls.append((ports_connected_listener,
                          upstream_component, upstream_port,
                          downstream_component, downstream_port))

        calls = []
        self._graph.add_port_added_listener(port_added_listener)
        self._graph.add_ports_connected_listener(ports_connected_listener)
        src = self._graph.add_component(MySource, 'src')
        sink = self._graph.add_component(MySink, 'sink')
        self._graph.connect_ports(src.output_ports['out'],
                                  sink.input_ports['in'])

        self.assertEqual(len(calls), 5)

        self.assertIs(calls[0][0], port_added_listener)
        self.assertEqual(calls[0][1].name, 'src')
        self.assertEqual(calls[0][2].name, 'out')

        self.assertIs(calls[1][0], port_added_listener)
        self.assertEqual(calls[1][1].name, 'src')
        self.assertEqual(calls[1][2].name, 'zero')

        self.assertIs(calls[2][0], port_added_listener)
        self.assertEqual(calls[2][1].name, 'sink')
        self.assertEqual(calls[2][2].name, 'in')

        self.assertIs(calls[3][0], port_added_listener)
        self.assertEqual(calls[3][1].name, 'sink')
        self.assertEqual(calls[3][2].name, 'taste')

        self.assertIs(calls[4][0], ports_connected_listener)
        self.assertEqual(calls[4][1].name, 'src')
        self.assertEqual(calls[4][2].name, 'out')
        self.assertEqual(calls[4][3].name, 'sink')
        self.assertEqual(calls[4][4].name, 'in')

    def test_invalid_listeners(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise bt2.Stop

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

            def _port_connected(self, port, other_port):
                self._add_input_port('taste')

        with self.assertRaises(TypeError):
            self._graph.add_port_added_listener(1234)
        with self.assertRaises(TypeError):
            self._graph.add_ports_connected_listener(1234)

    def test_raise_in_component_init(self):
        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                raise ValueError('oops!')

            def _consume(self):
                raise bt2.Stop

        graph = bt2.Graph()

        with self.assertRaises(bt2.Error):
            graph.add_component(MySink, 'comp')

    def test_raise_in_port_added_listener(self):
        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

        def port_added_listener(component, port):
            raise ValueError('oh noes!')

        graph = bt2.Graph()
        graph.add_port_added_listener(port_added_listener)

        with self.assertRaises(bt2.Error):
            graph.add_component(MySink, 'comp')

    def test_raise_in_ports_connected_listener(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise bt2.Stop

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

        class MySink(bt2._UserSinkComponent):
            def __init__(self, params):
                self._add_input_port('in')

            def _consume(self):
                raise bt2.Stop

        def ports_connected_listener(upstream_component, upstream_port,
                                     downstream_component, downstream_port):
            raise ValueError('oh noes!')

        graph = bt2.Graph()
        graph.add_ports_connected_listener(ports_connected_listener)
        up = graph.add_component(MySource, 'down')
        down = graph.add_component(MySink, 'up')

        with self.assertRaises(bt2.Error):
            graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
Commit	Line	Data
c4239792	1	from bt2 import value
811644b8 PP	2	import collections
	3	import unittest
	4	import copy
	5	import bt2
	6
	7
5f25509b	8	class _MyIter(bt2._UserMessageIterator):
c5f330cd	9	def __init__(self, self_output_port):
5f25509b SM	10	self._build_meta()
	11	self._at = 0
	12
	13	def _build_meta(self):
	14	self._tc = self._component._create_trace_class()
	15	self._t = self._tc()
	16	self._sc = self._tc.create_stream_class()
	17	self._ec = self._sc.create_event_class(name='salut')
	18	self._my_int_ft = self._tc.create_signed_integer_field_class(32)
	19	payload_ft = self._tc.create_structure_field_class()
	20	payload_ft += collections.OrderedDict([
	21	('my_int', self._my_int_ft),
	22	])
	23	self._ec.payload_field_type = payload_ft
	24	self._stream = self._t.create_stream(self._sc)
	25	self._packet = self._stream.create_packet()
	26
	27	def _create_event(self, value):
	28	ev = self._ec()
	29	ev.payload_field['my_int'] = value
	30	ev.packet = self._packet
	31	return ev
	32
	33
811644b8 PP	34	class GraphTestCase(unittest.TestCase):
	35	def setUp(self):
	36	self._graph = bt2.Graph()
	37
	38	def tearDown(self):
	39	del self._graph
	40
	41	def test_create_empty(self):
	42	graph = bt2.Graph()
	43
	44	def test_add_component_user_cls(self):
	45	class MySink(bt2._UserSinkComponent):
	46	def _consume(self):
	47	pass
	48
	49	comp = self._graph.add_component(MySink, 'salut')
	50	self.assertEqual(comp.name, 'salut')
	51
	52	def test_add_component_gen_cls(self):
	53	class MySink(bt2._UserSinkComponent):
	54	def _consume(self):
	55	pass
	56
	57	comp = self._graph.add_component(MySink, 'salut')
5f25509b	58	assert comp
811644b8 PP	59	comp2 = self._graph.add_component(comp.component_class, 'salut2')
	60	self.assertEqual(comp2.name, 'salut2')
	61
	62	def test_add_component_params(self):
	63	comp_params = None
	64
	65	class MySink(bt2._UserSinkComponent):
	66	def __init__(self, params):
	67	nonlocal comp_params
	68	comp_params = params
	69
	70	def _consume(self):
	71	pass
	72
	73	params = {'hello': 23, 'path': '/path/to/stuff'}
	74	comp = self._graph.add_component(MySink, 'salut', params)
	75	self.assertEqual(params, comp_params)
	76	del comp_params
	77
	78	def test_add_component_invalid_cls_type(self):
	79	with self.assertRaises(TypeError):
	80	self._graph.add_component(int, 'salut')
	81
	82	def test_connect_ports(self):
5602ef81	83	class MyIter(bt2._UserMessageIterator):
811644b8 PP	84	def __next__(self):
	85	raise bt2.Stop
	86
	87	class MySource(bt2._UserSourceComponent,
5602ef81	88	message_iterator_class=MyIter):
811644b8 PP	89	def __init__(self, params):
	90	self._add_output_port('out')
	91
	92	class MySink(bt2._UserSinkComponent):
	93	def __init__(self, params):
	94	self._add_input_port('in')
	95
	96	def _consume(self):
	97	raise bt2.Stop
	98
	99	src = self._graph.add_component(MySource, 'src')
	100	sink = self._graph.add_component(MySink, 'sink')
5f25509b	101
811644b8 PP	102	conn = self._graph.connect_ports(src.output_ports['out'],
	103	sink.input_ports['in'])
	104	self.assertTrue(src.output_ports['out'].is_connected)
	105	self.assertTrue(sink.input_ports['in'].is_connected)
5f25509b SM	106	self.assertEqual(src.output_ports['out'].connection._ptr, conn._ptr)
5f25509b SM	107	self.assertEqual(sink.input_ports['in'].connection._ptr, conn._ptr)
811644b8 PP	108
811644b8 PP	109	def test_connect_ports_invalid_direction(self):
5602ef81	110	class MyIter(bt2._UserMessageIterator):
811644b8 PP	111	def __next__(self):
	112	raise bt2.Stop
	113
	114	class MySource(bt2._UserSourceComponent,
5602ef81	115	message_iterator_class=MyIter):
811644b8 PP	116	def __init__(self, params):
	117	self._add_output_port('out')
	118
	119	class MySink(bt2._UserSinkComponent):
	120	def __init__(self, params):
	121	self._add_input_port('in')
	122
	123	def _consume(self):
	124	raise bt2.Stop
	125
	126	src = self._graph.add_component(MySource, 'src')
	127	sink = self._graph.add_component(MySink, 'sink')
	128
	129	with self.assertRaises(TypeError):
	130	conn = self._graph.connect_ports(sink.input_ports['in'],
	131	src.output_ports['out'])
	132
	133	def test_connect_ports_refused(self):
5602ef81	134	class MyIter(bt2._UserMessageIterator):
811644b8 PP	135	def __next__(self):
	136	raise bt2.Stop
	137
	138	class MySource(bt2._UserSourceComponent,
5602ef81	139	message_iterator_class=MyIter):
811644b8 PP	140	def __init__(self, params):
	141	self._add_output_port('out')
	142
	143	class MySink(bt2._UserSinkComponent):
	144	def __init__(self, params):
	145	self._add_input_port('in')
	146
	147	def _consume(self):
	148	raise bt2.Stop
	149
	150	def _accept_port_connection(self, port, other_port):
	151	return False
	152
	153	src = self._graph.add_component(MySource, 'src')
	154	sink = self._graph.add_component(MySink, 'sink')
	155
	156	with self.assertRaises(bt2.PortConnectionRefused):
	157	conn = self._graph.connect_ports(src.output_ports['out'],
	158	sink.input_ports['in'])
	159
5f25509b SM	160	def test_cancel(self):
5f25509b SM	161	self.assertFalse(self._graph.is_canceled)
811644b8	162	self._graph.cancel()
5f25509b	163	self.assertTrue(self._graph.is_canceled)
811644b8	164
5f25509b SM	165	# Test that Graph.run() raises bt2.GraphCanceled if the graph gets canceled
	166	# during execution.
	167	def test_cancel_while_running(self):
	168	class MyIter(_MyIter):
1d915789	169	def __next__(self):
5f25509b	170	return self._create_stream_beginning_message(self._stream)
1d915789 PP	171
1d915789 PP	172	class MySource(bt2._UserSourceComponent,
5602ef81	173	message_iterator_class=MyIter):
1d915789 PP	174	def __init__(self, params):
	175	self._add_output_port('out')
	176
	177	class MySink(bt2._UserSinkComponent):
	178	def __init__(self, params):
	179	self._add_input_port('in')
	180
	181	def _consume(self):
5f25509b SM	182	# Pretend that somebody asynchronously cancelled the graph.
	183	nonlocal graph
	184	graph.cancel()
1d915789	185
5f25509b	186	return next(self._msg_iter)
1d915789	187
5f25509b SM	188	def _graph_is_configured(self):
5f25509b SM	189	self._msg_iter = self._input_ports['in'].create_message_iterator()
1d915789	190
5f25509b SM	191	graph = bt2.Graph()
	192	up = graph.add_component(MySource, 'down')
	193	down = graph.add_component(MySink, 'up')
	194	graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
	195	with self.assertRaises(bt2.GraphCanceled):
	196	graph.run()
811644b8 PP	197
811644b8 PP	198	def test_run(self):
5f25509b	199	class MyIter(_MyIter):
811644b8	200	def __next__(self):
5f25509b SM	201	if self._at == 9:
	202	raise StopIteration
	203
	204	if self._at == 0:
	205	msg = self._create_stream_beginning_message(self._stream)
	206	elif self._at == 1:
	207	msg = self._create_packet_beginning_message(self._packet)
	208	elif self._at == 7:
	209	msg = self._create_packet_end_message(self._packet)
	210	elif self._at == 8:
	211	msg = self._create_stream_end_message(self._stream)
	212	else:
	213	msg = self._create_event_message(self._ec, self._packet)
811644b8	214
811644b8	215	self._at += 1
5602ef81	216	return msg
811644b8 PP	217
811644b8 PP	218	class MySource(bt2._UserSourceComponent,
5602ef81	219	message_iterator_class=MyIter):
811644b8 PP	220	def __init__(self, params):
	221	self._add_output_port('out')
	222
	223	class MySink(bt2._UserSinkComponent):
	224	def __init__(self, params):
5f25509b	225	self._input_port = self._add_input_port('in')
811644b8 PP	226	self._at = 0
	227
	228	def _consume(comp_self):
5602ef81	229	msg = next(comp_self._msg_iter)
811644b8 PP	230
811644b8 PP	231	if comp_self._at == 0:
5f25509b	232	self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
811644b8	233	elif comp_self._at == 1:
5f25509b	234	self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
811644b8	235	elif comp_self._at >= 2 and comp_self._at <= 6:
5f25509b	236	self.assertIsInstance(msg, bt2.message._EventMessage)
5602ef81	237	self.assertEqual(msg.event.event_class.name, 'salut')
811644b8	238	elif comp_self._at == 7:
5f25509b	239	self.assertIsInstance(msg, bt2.message._PacketEndMessage)
811644b8	240	elif comp_self._at == 8:
5f25509b	241	self.assertIsInstance(msg, bt2.message._StreamEndMessage)
811644b8 PP	242
	243	comp_self._at += 1
	244
5f25509b SM	245	def _graph_is_configured(self):
5f25509b SM	246	self._msg_iter = self._input_port.create_message_iterator()
811644b8 PP	247
	248	src = self._graph.add_component(MySource, 'src')
	249	sink = self._graph.add_component(MySink, 'sink')
	250	conn = self._graph.connect_ports(src.output_ports['out'],
	251	sink.input_ports['in'])
	252	self._graph.run()
	253
	254	def test_run_again(self):
5f25509b	255	class MyIter(_MyIter):
811644b8	256	def __next__(self):
5f25509b	257	if self._at == 3:
811644b8 PP	258	raise bt2.TryAgain
811644b8 PP	259
5f25509b SM	260	if self._at == 0:
	261	msg = self._create_stream_beginning_message(self._stream)
	262	elif self._at == 1:
	263	msg = self._create_packet_beginning_message(self._packet)
	264	elif self._at == 2:
	265	msg = self._create_event_message(self._ec, self._packet)
	266
811644b8	267	self._at += 1
5602ef81	268	return msg
811644b8 PP	269
811644b8 PP	270	class MySource(bt2._UserSourceComponent,
5602ef81	271	message_iterator_class=MyIter):
811644b8 PP	272	def __init__(self, params):
	273	self._add_output_port('out')
	274
	275	class MySink(bt2._UserSinkComponent):
	276	def __init__(self, params):
5f25509b	277	self._input_port = self._add_input_port('in')
811644b8 PP	278	self._at = 0
	279
	280	def _consume(comp_self):
5f25509b	281	msg = next(comp_self._msg_iter)
811644b8	282	if comp_self._at == 0:
5f25509b	283	self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
811644b8	284	elif comp_self._at == 1:
5f25509b SM	285	self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
	286	elif comp_self._at == 2:
	287	self.assertIsInstance(msg, bt2.message._EventMessage)
811644b8	288	raise bt2.TryAgain
5f25509b SM	289	else:
5f25509b SM	290	pass
811644b8 PP	291
	292	comp_self._at += 1
	293
5f25509b SM	294	def _graph_is_configured(self):
5f25509b SM	295	self._msg_iter = self._input_port.create_message_iterator()
811644b8 PP	296
	297	src = self._graph.add_component(MySource, 'src')
	298	sink = self._graph.add_component(MySink, 'sink')
	299	conn = self._graph.connect_ports(src.output_ports['out'],
	300	sink.input_ports['in'])
	301
	302	with self.assertRaises(bt2.TryAgain):
	303	self._graph.run()
	304
811644b8	305	def test_run_error(self):
5f25509b	306	raised_in_sink = False
811644b8	307
5f25509b	308	class MyIter(_MyIter):
811644b8	309	def __next__(self):
5f25509b SM	310	# If this gets called after the sink raised an exception, it is
	311	# an error.
	312	nonlocal raised_in_sink
	313	assert raised_in_sink is False
	314
	315	if self._at == 0:
	316	msg = self._create_stream_beginning_message(self._stream)
	317	elif self._at == 1:
	318	msg = self._create_packet_beginning_message(self._packet)
	319	elif self._at == 2 or self._at == 3:
	320	msg = self._create_event_message(self._ec, self._packet)
	321	else:
811644b8	322	raise bt2.TryAgain
811644b8	323	self._at += 1
5602ef81	324	return msg
811644b8 PP	325
811644b8 PP	326	class MySource(bt2._UserSourceComponent,
5602ef81	327	message_iterator_class=MyIter):
811644b8 PP	328	def __init__(self, params):
	329	self._add_output_port('out')
	330
	331	class MySink(bt2._UserSinkComponent):
	332	def __init__(self, params):
5f25509b	333	self._input_port = self._add_input_port('in')
811644b8 PP	334	self._at = 0
	335
	336	def _consume(comp_self):
5f25509b	337	msg = next(comp_self._msg_iter)
811644b8	338	if comp_self._at == 0:
5f25509b	339	self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
811644b8	340	elif comp_self._at == 1:
5f25509b SM	341	self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
	342	elif comp_self._at == 2:
	343	self.assertIsInstance(msg, bt2.message._EventMessage)
	344	elif comp_self._at == 3:
	345	nonlocal raised_in_sink
	346	raised_in_sink = True
811644b8 PP	347	raise RuntimeError('error!')
	348
	349	comp_self._at += 1
	350
5f25509b SM	351	def _graph_is_configured(self):
5f25509b SM	352	self._msg_iter = self._input_port.create_message_iterator()
811644b8 PP	353
	354	src = self._graph.add_component(MySource, 'src')
	355	sink = self._graph.add_component(MySink, 'sink')
	356	conn = self._graph.connect_ports(src.output_ports['out'],
	357	sink.input_ports['in'])
	358
	359	with self.assertRaises(bt2.Error):
	360	self._graph.run()
	361
5f25509b	362	def test_listeners(self):
5602ef81	363	class MyIter(bt2._UserMessageIterator):
5f25509b SM	364	def __next__(self):
5f25509b SM	365	raise bt2.Stop
1d915789 PP	366
1d915789 PP	367	class MySource(bt2._UserSourceComponent,
5602ef81	368	message_iterator_class=MyIter):
1d915789 PP	369	def __init__(self, params):
1d915789 PP	370	self._add_output_port('out')
5f25509b	371	self._add_output_port('zero')
1d915789 PP	372
	373	class MySink(bt2._UserSinkComponent):
	374	def __init__(self, params):
	375	self._add_input_port('in')
1d915789	376
5f25509b SM	377	def _consume(self):
5f25509b SM	378	raise bt2.Stop
1d915789	379
5f25509b SM	380	def _port_connected(self, port, other_port):
5f25509b SM	381	self._add_input_port('taste')
1d915789	382
5f25509b SM	383	def port_added_listener(component, port):
	384	nonlocal calls
	385	calls.append((port_added_listener, component, port))
1d915789	386
5f25509b SM	387	def ports_connected_listener(upstream_component, upstream_port,
	388	downstream_component, downstream_port):
	389	nonlocal calls
	390	calls.append((ports_connected_listener,
	391	upstream_component, upstream_port,
	392	downstream_component, downstream_port))
	393
	394	calls = []
	395	self._graph.add_port_added_listener(port_added_listener)
	396	self._graph.add_ports_connected_listener(ports_connected_listener)
1d915789 PP	397	src = self._graph.add_component(MySource, 'src')
1d915789 PP	398	sink = self._graph.add_component(MySink, 'sink')
5f25509b SM	399	self._graph.connect_ports(src.output_ports['out'],
5f25509b SM	400	sink.input_ports['in'])
1d915789	401
5f25509b SM	402	self.assertEqual(len(calls), 5)
	403
	404	self.assertIs(calls[0][0], port_added_listener)
	405	self.assertEqual(calls[0][1].name, 'src')
	406	self.assertEqual(calls[0][2].name, 'out')
	407
	408	self.assertIs(calls[1][0], port_added_listener)
	409	self.assertEqual(calls[1][1].name, 'src')
	410	self.assertEqual(calls[1][2].name, 'zero')
	411
	412	self.assertIs(calls[2][0], port_added_listener)
	413	self.assertEqual(calls[2][1].name, 'sink')
	414	self.assertEqual(calls[2][2].name, 'in')
	415
	416	self.assertIs(calls[3][0], port_added_listener)
	417	self.assertEqual(calls[3][1].name, 'sink')
	418	self.assertEqual(calls[3][2].name, 'taste')
	419
	420	self.assertIs(calls[4][0], ports_connected_listener)
	421	self.assertEqual(calls[4][1].name, 'src')
	422	self.assertEqual(calls[4][2].name, 'out')
	423	self.assertEqual(calls[4][3].name, 'sink')
	424	self.assertEqual(calls[4][4].name, 'in')
	425
	426	def test_invalid_listeners(self):
5602ef81	427	class MyIter(bt2._UserMessageIterator):
811644b8 PP	428	def __next__(self):
	429	raise bt2.Stop
	430
	431	class MySource(bt2._UserSourceComponent,
5602ef81	432	message_iterator_class=MyIter):
811644b8 PP	433	def __init__(self, params):
	434	self._add_output_port('out')
	435	self._add_output_port('zero')
	436
811644b8 PP	437	class MySink(bt2._UserSinkComponent):
	438	def __init__(self, params):
	439	self._add_input_port('in')
	440
	441	def _consume(self):
	442	raise bt2.Stop
	443
	444	def _port_connected(self, port, other_port):
	445	self._add_input_port('taste')
	446
5f25509b SM	447	with self.assertRaises(TypeError):
	448	self._graph.add_port_added_listener(1234)
	449	with self.assertRaises(TypeError):
	450	self._graph.add_ports_connected_listener(1234)
811644b8	451
5f25509b SM	452	def test_raise_in_component_init(self):
	453	class MySink(bt2._UserSinkComponent):
	454	def __init__(self, params):
	455	raise ValueError('oops!')
811644b8	456
5f25509b SM	457	def _consume(self):
	458	raise bt2.Stop
	459
	460	graph = bt2.Graph()
	461
	462	with self.assertRaises(bt2.Error):
	463	graph.add_component(MySink, 'comp')
	464
	465	def test_raise_in_port_added_listener(self):
	466	class MySink(bt2._UserSinkComponent):
	467	def __init__(self, params):
	468	self._add_input_port('in')
	469
	470	def _consume(self):
	471	raise bt2.Stop
	472
	473	def port_added_listener(component, port):
	474	raise ValueError('oh noes!')
	475
	476	graph = bt2.Graph()
	477	graph.add_port_added_listener(port_added_listener)
	478
	479	with self.assertRaises(bt2.Error):
	480	graph.add_component(MySink, 'comp')
	481
	482	def test_raise_in_ports_connected_listener(self):
	483	class MyIter(bt2._UserMessageIterator):
	484	def __next__(self):
	485	raise bt2.Stop
	486
	487	class MySource(bt2._UserSourceComponent,
	488	message_iterator_class=MyIter):
	489	def __init__(self, params):
	490	self._add_output_port('out')
	491
	492	class MySink(bt2._UserSinkComponent):
	493	def __init__(self, params):
	494	self._add_input_port('in')
	495
	496	def _consume(self):
	497	raise bt2.Stop
811644b8	498
c5f330cd SM	499	def ports_connected_listener(upstream_component, upstream_port,
c5f330cd SM	500	downstream_component, downstream_port):
5f25509b	501	raise ValueError('oh noes!')
811644b8	502
5f25509b SM	503	graph = bt2.Graph()
	504	graph.add_ports_connected_listener(ports_connected_listener)
	505	up = graph.add_component(MySource, 'down')
	506	down = graph.add_component(MySink, 'up')
811644b8	507
5f25509b SM	508	with self.assertRaises(bt2.Error):
5f25509b SM	509	graph.connect_ports(up.output_ports['out'], down.input_ports['in'])