[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(supports_packets=True)
        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 += [('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

            def _graph_is_configured(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

            def _graph_is_configured(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

            def _graph_is_configured(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

            def _graph_is_configured(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

            def _graph_is_configured(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

            def _graph_is_configured(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

            def _graph_is_configured(self):
                pass

        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.addr, conn.addr)
        self.assertEqual(sink.input_ports['in'].connection.addr, conn.addr)

    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

            def _graph_is_configured(self):
                pass

        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_add_interrupter(self):
        class MyIter(bt2._UserMessageIterator):
            def __next__(self):
                raise TypeError

        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):
                next(self._msg_iter)

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

        # add two interrupters, set one of them
        interrupter1 = bt2.Interrupter()
        interrupter2 = bt2.Interrupter()
        self._graph.add_interrupter(interrupter1)
        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._graph.add_interrupter(interrupter2)

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

        interrupter2.set()

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

        interrupter2.reset()

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

    # Test that Graph.run() raises bt2.Interrupted if the graph gets
    # interrupted during execution.
    def test_interrupt_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 interrupted the graph.
                nonlocal graph
                graph.interrupt()
                return next(self._msg_iter)

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

        graph = self._graph
        up = self._graph.add_component(MySource, 'down')
        down = self._graph.add_component(MySink, 'up')
        self._graph.connect_ports(up.output_ports['out'], down.input_ports['in'])

        with self.assertRaises(bt2.TryAgain):
            self._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._StreamBeginningMessage)
                elif comp_self._at == 1:
                    self.assertIsInstance(msg, bt2._PacketBeginningMessage)
                elif comp_self._at >= 2 and comp_self._at <= 6:
                    self.assertIsInstance(msg, bt2._EventMessage)
                    self.assertEqual(msg.event.cls.name, 'salut')
                elif comp_self._at == 7:
                    self.assertIsInstance(msg, bt2._PacketEndMessage)
                elif comp_self._at == 8:
                    self.assertIsInstance(msg, bt2._StreamEndMessage)

                comp_self._at += 1

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

        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._StreamBeginningMessage)
                elif comp_self._at == 1:
                    self.assertIsInstance(msg, bt2._PacketBeginningMessage)
                elif comp_self._at == 2:
                    self.assertIsInstance(msg, bt2._EventMessage)
                    raise bt2.TryAgain
                else:
                    pass

                comp_self._at += 1

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

        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._StreamBeginningMessage)
                elif comp_self._at == 1:
                    self.assertIsInstance(msg, bt2._PacketBeginningMessage)
                elif comp_self._at == 2:
                    self.assertIsInstance(msg, bt2._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._create_input_port_message_iterator(
                    self._input_port
                )

        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 _graph_is_configured(self):
                pass

            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 _graph_is_configured(self):
                pass

            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

            def _graph_is_configured(self):
                pass

        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 _graph_is_configured(self):
                pass

        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 _graph_is_configured(self):
                pass

        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
32d2d479 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
ae0bfae8	19	from bt2 import value
f6a5e476 PP	20	import collections
	21	import unittest
	22	import copy
	23	import bt2
	24
	25
871a292a	26	class _MyIter(bt2._UserMessageIterator):
a4dcfa96	27	def __init__(self, self_output_port):
871a292a 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()
37a93d41	34	self._sc = self._tc.create_stream_class(supports_packets=True)
871a292a SM	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()
61d96b89	38	payload_ft += [('my_int', self._my_int_ft)]
871a292a SM	39	self._ec.payload_field_type = payload_ft
	40	self._stream = self._t.create_stream(self._sc)
	41	self._packet = self._stream.create_packet()
	42
	43	def _create_event(self, value):
	44	ev = self._ec()
	45	ev.payload_field['my_int'] = value
	46	ev.packet = self._packet
	47	return ev
	48
	49
f6a5e476 PP	50	class GraphTestCase(unittest.TestCase):
	51	def setUp(self):
	52	self._graph = bt2.Graph()
	53
	54	def tearDown(self):
	55	del self._graph
	56
	57	def test_create_empty(self):
	58	graph = bt2.Graph()
	59
	60	def test_add_component_user_cls(self):
	61	class MySink(bt2._UserSinkComponent):
	62	def _consume(self):
	63	pass
61d96b89	64
8a08af82 SM	65	def _graph_is_configured(self):
8a08af82 SM	66	pass
f6a5e476 PP	67
	68	comp = self._graph.add_component(MySink, 'salut')
	69	self.assertEqual(comp.name, 'salut')
	70
	71	def test_add_component_gen_cls(self):
	72	class MySink(bt2._UserSinkComponent):
	73	def _consume(self):
	74	pass
61d96b89	75
8a08af82 SM	76	def _graph_is_configured(self):
8a08af82 SM	77	pass
f6a5e476 PP	78
f6a5e476 PP	79	comp = self._graph.add_component(MySink, 'salut')
871a292a	80	assert comp
c88be1c8	81	comp2 = self._graph.add_component(comp.cls, 'salut2')
f6a5e476 PP	82	self.assertEqual(comp2.name, 'salut2')
	83
	84	def test_add_component_params(self):
	85	comp_params = None
	86
	87	class MySink(bt2._UserSinkComponent):
	88	def __init__(self, params):
	89	nonlocal comp_params
	90	comp_params = params
	91
	92	def _consume(self):
	93	pass
61d96b89	94
8a08af82 SM	95	def _graph_is_configured(self):
8a08af82 SM	96	pass
f6a5e476 PP	97
	98	params = {'hello': 23, 'path': '/path/to/stuff'}
	99	comp = self._graph.add_component(MySink, 'salut', params)
	100	self.assertEqual(params, comp_params)
	101	del comp_params
	102
	103	def test_add_component_invalid_cls_type(self):
	104	with self.assertRaises(TypeError):
	105	self._graph.add_component(int, 'salut')
	106
78c432bb PP	107	def test_add_component_invalid_logging_level_type(self):
	108	class MySink(bt2._UserSinkComponent):
	109	def _consume(self):
	110	pass
61d96b89	111
8a08af82 SM	112	def _graph_is_configured(self):
8a08af82 SM	113	pass
78c432bb PP	114
	115	with self.assertRaises(TypeError):
	116	self._graph.add_component(MySink, 'salut', logging_level='yo')
	117
	118	def test_add_component_invalid_logging_level_value(self):
	119	class MySink(bt2._UserSinkComponent):
	120	def _consume(self):
	121	pass
61d96b89	122
8a08af82 SM	123	def _graph_is_configured(self):
8a08af82 SM	124	pass
78c432bb PP	125
	126	with self.assertRaises(ValueError):
	127	self._graph.add_component(MySink, 'salut', logging_level=12345)
	128
	129	def test_add_component_logging_level(self):
	130	class MySink(bt2._UserSinkComponent):
	131	def _consume(self):
	132	pass
61d96b89	133
8a08af82 SM	134	def _graph_is_configured(self):
8a08af82 SM	135	pass
78c432bb	136
61d96b89 FD	137	comp = self._graph.add_component(
	138	MySink, 'salut', logging_level=bt2.LoggingLevel.DEBUG
	139	)
78c432bb PP	140	self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)
78c432bb PP	141
f6a5e476	142	def test_connect_ports(self):
fa4c33e3	143	class MyIter(bt2._UserMessageIterator):
f6a5e476 PP	144	def __next__(self):
	145	raise bt2.Stop
	146
61d96b89	147	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
f6a5e476 PP	148	def __init__(self, params):
	149	self._add_output_port('out')
	150
	151	class MySink(bt2._UserSinkComponent):
	152	def __init__(self, params):
	153	self._add_input_port('in')
	154
	155	def _consume(self):
	156	raise bt2.Stop
	157
8a08af82 SM	158	def _graph_is_configured(self):
	159	pass
	160
f6a5e476 PP	161	src = self._graph.add_component(MySource, 'src')
f6a5e476 PP	162	sink = self._graph.add_component(MySink, 'sink')
871a292a	163
61d96b89 FD	164	conn = self._graph.connect_ports(
	165	src.output_ports['out'], sink.input_ports['in']
	166	)
f6a5e476 PP	167	self.assertTrue(src.output_ports['out'].is_connected)
f6a5e476 PP	168	self.assertTrue(sink.input_ports['in'].is_connected)
3e2cc5dc FD	169	self.assertEqual(src.output_ports['out'].connection.addr, conn.addr)
3e2cc5dc FD	170	self.assertEqual(sink.input_ports['in'].connection.addr, conn.addr)
f6a5e476 PP	171
f6a5e476 PP	172	def test_connect_ports_invalid_direction(self):
fa4c33e3	173	class MyIter(bt2._UserMessageIterator):
f6a5e476 PP	174	def __next__(self):
	175	raise bt2.Stop
	176
61d96b89	177	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
f6a5e476 PP	178	def __init__(self, params):
	179	self._add_output_port('out')
	180
	181	class MySink(bt2._UserSinkComponent):
	182	def __init__(self, params):
	183	self._add_input_port('in')
	184
	185	def _consume(self):
	186	raise bt2.Stop
	187
8a08af82 SM	188	def _graph_is_configured(self):
	189	pass
	190
f6a5e476 PP	191	src = self._graph.add_component(MySource, 'src')
	192	sink = self._graph.add_component(MySink, 'sink')
	193
	194	with self.assertRaises(TypeError):
61d96b89 FD	195	conn = self._graph.connect_ports(
	196	sink.input_ports['in'], src.output_ports['out']
	197	)
f6a5e476	198
d73bb381 PP	199	def test_add_interrupter(self):
	200	class MyIter(bt2._UserMessageIterator):
	201	def __next__(self):
	202	raise TypeError
	203
	204	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
	205	def __init__(self, params):
	206	self._add_output_port('out')
	207
	208	class MySink(bt2._UserSinkComponent):
	209	def __init__(self, params):
	210	self._add_input_port('in')
	211
	212	def _consume(self):
	213	next(self._msg_iter)
	214
	215	def _graph_is_configured(self):
	216	self._msg_iter = self._create_input_port_message_iterator(
	217	self._input_ports['in']
	218	)
	219
	220	# add two interrupters, set one of them
	221	interrupter1 = bt2.Interrupter()
	222	interrupter2 = bt2.Interrupter()
	223	self._graph.add_interrupter(interrupter1)
	224	src = self._graph.add_component(MySource, 'src')
	225	sink = self._graph.add_component(MySink, 'sink')
	226	self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
	227	self._graph.add_interrupter(interrupter2)
	228
	229	with self.assertRaises(bt2._Error):
	230	self._graph.run()
	231
	232	interrupter2.set()
	233
	234	with self.assertRaises(bt2.TryAgain):
	235	self._graph.run()
	236
	237	interrupter2.reset()
	238
	239	with self.assertRaises(bt2._Error):
	240	self._graph.run()
	241
	242	# Test that Graph.run() raises bt2.Interrupted if the graph gets
	243	# interrupted during execution.
	244	def test_interrupt_while_running(self):
871a292a	245	class MyIter(_MyIter):
9ef22b36	246	def __next__(self):
871a292a	247	return self._create_stream_beginning_message(self._stream)
9ef22b36	248
61d96b89	249	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
9ef22b36 PP	250	def __init__(self, params):
	251	self._add_output_port('out')
	252
	253	class MySink(bt2._UserSinkComponent):
	254	def __init__(self, params):
	255	self._add_input_port('in')
	256
	257	def _consume(self):
d73bb381	258	# Pretend that somebody asynchronously interrupted the graph.
871a292a	259	nonlocal graph
d73bb381	260	graph.interrupt()
871a292a	261	return next(self._msg_iter)
9ef22b36	262
871a292a	263	def _graph_is_configured(self):
692f1a01 PP	264	self._msg_iter = self._create_input_port_message_iterator(
	265	self._input_ports['in']
	266	)
9ef22b36	267
d73bb381 PP	268	graph = self._graph
	269	up = self._graph.add_component(MySource, 'down')
	270	down = self._graph.add_component(MySink, 'up')
	271	self._graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
	272
	273	with self.assertRaises(bt2.TryAgain):
	274	self._graph.run()
f6a5e476 PP	275
f6a5e476 PP	276	def test_run(self):
871a292a	277	class MyIter(_MyIter):
f6a5e476	278	def __next__(self):
871a292a SM	279	if self._at == 9:
	280	raise StopIteration
	281
	282	if self._at == 0:
	283	msg = self._create_stream_beginning_message(self._stream)
	284	elif self._at == 1:
	285	msg = self._create_packet_beginning_message(self._packet)
	286	elif self._at == 7:
	287	msg = self._create_packet_end_message(self._packet)
	288	elif self._at == 8:
	289	msg = self._create_stream_end_message(self._stream)
	290	else:
	291	msg = self._create_event_message(self._ec, self._packet)
f6a5e476	292
f6a5e476	293	self._at += 1
fa4c33e3	294	return msg
f6a5e476	295
61d96b89	296	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
f6a5e476 PP	297	def __init__(self, params):
	298	self._add_output_port('out')
	299
	300	class MySink(bt2._UserSinkComponent):
	301	def __init__(self, params):
871a292a	302	self._input_port = self._add_input_port('in')
f6a5e476 PP	303	self._at = 0
	304
	305	def _consume(comp_self):
fa4c33e3	306	msg = next(comp_self._msg_iter)
f6a5e476 PP	307
f6a5e476 PP	308	if comp_self._at == 0:
c946c9de	309	self.assertIsInstance(msg, bt2._StreamBeginningMessage)
f6a5e476	310	elif comp_self._at == 1:
c946c9de	311	self.assertIsInstance(msg, bt2._PacketBeginningMessage)
f6a5e476	312	elif comp_self._at >= 2 and comp_self._at <= 6:
c946c9de	313	self.assertIsInstance(msg, bt2._EventMessage)
c88be1c8	314	self.assertEqual(msg.event.cls.name, 'salut')
f6a5e476	315	elif comp_self._at == 7:
c946c9de	316	self.assertIsInstance(msg, bt2._PacketEndMessage)
f6a5e476	317	elif comp_self._at == 8:
c946c9de	318	self.assertIsInstance(msg, bt2._StreamEndMessage)
f6a5e476 PP	319
	320	comp_self._at += 1
	321
871a292a	322	def _graph_is_configured(self):
692f1a01 PP	323	self._msg_iter = self._create_input_port_message_iterator(
	324	self._input_port
	325	)
f6a5e476 PP	326
	327	src = self._graph.add_component(MySource, 'src')
	328	sink = self._graph.add_component(MySink, 'sink')
61d96b89 FD	329	conn = self._graph.connect_ports(
	330	src.output_ports['out'], sink.input_ports['in']
	331	)
f6a5e476 PP	332	self._graph.run()
	333
	334	def test_run_again(self):
871a292a	335	class MyIter(_MyIter):
f6a5e476	336	def __next__(self):
871a292a	337	if self._at == 3:
f6a5e476 PP	338	raise bt2.TryAgain
f6a5e476 PP	339
871a292a SM	340	if self._at == 0:
	341	msg = self._create_stream_beginning_message(self._stream)
	342	elif self._at == 1:
	343	msg = self._create_packet_beginning_message(self._packet)
	344	elif self._at == 2:
	345	msg = self._create_event_message(self._ec, self._packet)
	346
f6a5e476	347	self._at += 1
fa4c33e3	348	return msg
f6a5e476	349
61d96b89	350	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
f6a5e476 PP	351	def __init__(self, params):
	352	self._add_output_port('out')
	353
	354	class MySink(bt2._UserSinkComponent):
	355	def __init__(self, params):
871a292a	356	self._input_port = self._add_input_port('in')
f6a5e476 PP	357	self._at = 0
	358
	359	def _consume(comp_self):
871a292a	360	msg = next(comp_self._msg_iter)
f6a5e476	361	if comp_self._at == 0:
c946c9de	362	self.assertIsInstance(msg, bt2._StreamBeginningMessage)
f6a5e476	363	elif comp_self._at == 1:
c946c9de	364	self.assertIsInstance(msg, bt2._PacketBeginningMessage)
871a292a	365	elif comp_self._at == 2:
c946c9de	366	self.assertIsInstance(msg, bt2._EventMessage)
f6a5e476	367	raise bt2.TryAgain
871a292a SM	368	else:
871a292a SM	369	pass
f6a5e476 PP	370
	371	comp_self._at += 1
	372
871a292a	373	def _graph_is_configured(self):
692f1a01 PP	374	self._msg_iter = self._create_input_port_message_iterator(
	375	self._input_port
	376	)
f6a5e476 PP	377
	378	src = self._graph.add_component(MySource, 'src')
	379	sink = self._graph.add_component(MySink, 'sink')
61d96b89 FD	380	conn = self._graph.connect_ports(
	381	src.output_ports['out'], sink.input_ports['in']
	382	)
f6a5e476 PP	383
	384	with self.assertRaises(bt2.TryAgain):
	385	self._graph.run()
	386
f6a5e476	387	def test_run_error(self):
871a292a	388	raised_in_sink = False
f6a5e476	389
871a292a	390	class MyIter(_MyIter):
f6a5e476	391	def __next__(self):
871a292a SM	392	# If this gets called after the sink raised an exception, it is
	393	# an error.
	394	nonlocal raised_in_sink
	395	assert raised_in_sink is False
	396
	397	if self._at == 0:
	398	msg = self._create_stream_beginning_message(self._stream)
	399	elif self._at == 1:
	400	msg = self._create_packet_beginning_message(self._packet)
	401	elif self._at == 2 or self._at == 3:
	402	msg = self._create_event_message(self._ec, self._packet)
	403	else:
f6a5e476	404	raise bt2.TryAgain
f6a5e476	405	self._at += 1
fa4c33e3	406	return msg
f6a5e476	407
61d96b89	408	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
f6a5e476 PP	409	def __init__(self, params):
	410	self._add_output_port('out')
	411
	412	class MySink(bt2._UserSinkComponent):
	413	def __init__(self, params):
871a292a	414	self._input_port = self._add_input_port('in')
f6a5e476 PP	415	self._at = 0
	416
	417	def _consume(comp_self):
871a292a	418	msg = next(comp_self._msg_iter)
f6a5e476	419	if comp_self._at == 0:
c946c9de	420	self.assertIsInstance(msg, bt2._StreamBeginningMessage)
f6a5e476	421	elif comp_self._at == 1:
c946c9de	422	self.assertIsInstance(msg, bt2._PacketBeginningMessage)
871a292a	423	elif comp_self._at == 2:
c946c9de	424	self.assertIsInstance(msg, bt2._EventMessage)
871a292a SM	425	elif comp_self._at == 3:
	426	nonlocal raised_in_sink
	427	raised_in_sink = True
f6a5e476 PP	428	raise RuntimeError('error!')
	429
	430	comp_self._at += 1
	431
871a292a	432	def _graph_is_configured(self):
692f1a01 PP	433	self._msg_iter = self._create_input_port_message_iterator(
	434	self._input_port
	435	)
f6a5e476 PP	436
	437	src = self._graph.add_component(MySource, 'src')
	438	sink = self._graph.add_component(MySink, 'sink')
61d96b89 FD	439	conn = self._graph.connect_ports(
	440	src.output_ports['out'], sink.input_ports['in']
	441	)
f6a5e476	442
614743a5	443	with self.assertRaises(bt2._Error):
f6a5e476 PP	444	self._graph.run()
f6a5e476 PP	445
871a292a	446	def test_listeners(self):
fa4c33e3	447	class MyIter(bt2._UserMessageIterator):
871a292a SM	448	def __next__(self):
871a292a SM	449	raise bt2.Stop
9ef22b36	450
61d96b89	451	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
9ef22b36 PP	452	def __init__(self, params):
9ef22b36 PP	453	self._add_output_port('out')
871a292a	454	self._add_output_port('zero')
9ef22b36 PP	455
	456	class MySink(bt2._UserSinkComponent):
	457	def __init__(self, params):
	458	self._add_input_port('in')
9ef22b36	459
871a292a SM	460	def _consume(self):
871a292a SM	461	raise bt2.Stop
9ef22b36	462
8a08af82 SM	463	def _graph_is_configured(self):
	464	pass
	465
871a292a SM	466	def _port_connected(self, port, other_port):
871a292a SM	467	self._add_input_port('taste')
9ef22b36	468
871a292a SM	469	def port_added_listener(component, port):
	470	nonlocal calls
	471	calls.append((port_added_listener, component, port))
9ef22b36	472
61d96b89 FD	473	def ports_connected_listener(
	474	upstream_component, upstream_port, downstream_component, downstream_port
	475	):
871a292a	476	nonlocal calls
61d96b89 FD	477	calls.append(
	478	(
	479	ports_connected_listener,
	480	upstream_component,
	481	upstream_port,
	482	downstream_component,
	483	downstream_port,
	484	)
	485	)
871a292a SM	486
	487	calls = []
	488	self._graph.add_port_added_listener(port_added_listener)
	489	self._graph.add_ports_connected_listener(ports_connected_listener)
9ef22b36 PP	490	src = self._graph.add_component(MySource, 'src')
9ef22b36 PP	491	sink = self._graph.add_component(MySink, 'sink')
61d96b89	492	self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
9ef22b36	493
871a292a SM	494	self.assertEqual(len(calls), 5)
	495
	496	self.assertIs(calls[0][0], port_added_listener)
	497	self.assertEqual(calls[0][1].name, 'src')
	498	self.assertEqual(calls[0][2].name, 'out')
	499
	500	self.assertIs(calls[1][0], port_added_listener)
	501	self.assertEqual(calls[1][1].name, 'src')
	502	self.assertEqual(calls[1][2].name, 'zero')
	503
	504	self.assertIs(calls[2][0], port_added_listener)
	505	self.assertEqual(calls[2][1].name, 'sink')
	506	self.assertEqual(calls[2][2].name, 'in')
	507
	508	self.assertIs(calls[3][0], port_added_listener)
	509	self.assertEqual(calls[3][1].name, 'sink')
	510	self.assertEqual(calls[3][2].name, 'taste')
	511
	512	self.assertIs(calls[4][0], ports_connected_listener)
	513	self.assertEqual(calls[4][1].name, 'src')
	514	self.assertEqual(calls[4][2].name, 'out')
	515	self.assertEqual(calls[4][3].name, 'sink')
	516	self.assertEqual(calls[4][4].name, 'in')
	517
	518	def test_invalid_listeners(self):
fa4c33e3	519	class MyIter(bt2._UserMessageIterator):
f6a5e476 PP	520	def __next__(self):
	521	raise bt2.Stop
	522
61d96b89	523	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
f6a5e476 PP	524	def __init__(self, params):
	525	self._add_output_port('out')
	526	self._add_output_port('zero')
	527
f6a5e476 PP	528	class MySink(bt2._UserSinkComponent):
	529	def __init__(self, params):
	530	self._add_input_port('in')
	531
	532	def _consume(self):
	533	raise bt2.Stop
	534
8a08af82 SM	535	def _graph_is_configured(self):
	536	pass
	537
f6a5e476 PP	538	def _port_connected(self, port, other_port):
	539	self._add_input_port('taste')
	540
871a292a SM	541	with self.assertRaises(TypeError):
	542	self._graph.add_port_added_listener(1234)
	543	with self.assertRaises(TypeError):
	544	self._graph.add_ports_connected_listener(1234)
f6a5e476	545
871a292a SM	546	def test_raise_in_component_init(self):
	547	class MySink(bt2._UserSinkComponent):
	548	def __init__(self, params):
	549	raise ValueError('oops!')
f6a5e476	550
871a292a SM	551	def _consume(self):
	552	raise bt2.Stop
	553
8a08af82 SM	554	def _graph_is_configured(self):
	555	pass
	556
871a292a SM	557	graph = bt2.Graph()
871a292a SM	558
614743a5	559	with self.assertRaises(bt2._Error):
871a292a SM	560	graph.add_component(MySink, 'comp')
	561
	562	def test_raise_in_port_added_listener(self):
	563	class MySink(bt2._UserSinkComponent):
	564	def __init__(self, params):
	565	self._add_input_port('in')
	566
	567	def _consume(self):
	568	raise bt2.Stop
	569
8a08af82 SM	570	def _graph_is_configured(self):
	571	pass
	572
871a292a SM	573	def port_added_listener(component, port):
	574	raise ValueError('oh noes!')
	575
	576	graph = bt2.Graph()
	577	graph.add_port_added_listener(port_added_listener)
	578
614743a5	579	with self.assertRaises(bt2._Error):
871a292a SM	580	graph.add_component(MySink, 'comp')
	581
	582	def test_raise_in_ports_connected_listener(self):
	583	class MyIter(bt2._UserMessageIterator):
	584	def __next__(self):
	585	raise bt2.Stop
	586
61d96b89	587	class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
871a292a SM	588	def __init__(self, params):
	589	self._add_output_port('out')
	590
	591	class MySink(bt2._UserSinkComponent):
	592	def __init__(self, params):
	593	self._add_input_port('in')
	594
	595	def _consume(self):
	596	raise bt2.Stop
f6a5e476	597
8a08af82 SM	598	def _graph_is_configured(self):
	599	pass
	600
61d96b89 FD	601	def ports_connected_listener(
	602	upstream_component, upstream_port, downstream_component, downstream_port
	603	):
871a292a	604	raise ValueError('oh noes!')
f6a5e476	605
871a292a SM	606	graph = bt2.Graph()
	607	graph.add_ports_connected_listener(ports_connected_listener)
	608	up = graph.add_component(MySource, 'down')
	609	down = graph.add_component(MySink, 'up')
f6a5e476	610
614743a5	611	with self.assertRaises(bt2._Error):
871a292a	612	graph.connect_ports(up.output_ports['out'], down.input_ports['in'])