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

#
# Copyright (C) 2019 EfficiOS Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; only version 2
# of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
#

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.cls, '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_add_component_invalid_logging_level_type(self):
        class MySink(bt2._UserSinkComponent):
            def _consume(self):
                pass

        with self.assertRaises(TypeError):
            self._graph.add_component(MySink, 'salut', logging_level='yo')

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

        with self.assertRaises(ValueError):
            self._graph.add_component(MySink, 'salut', logging_level=12345)

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

        comp = self._graph.add_component(MySink, 'salut',
                                         logging_level=bt2.LoggingLevel.DEBUG)
        self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)

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