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

from bt2 import values
import unittest
import copy
import bt2


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

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

            def _port_connected(self, port, other_port):
                self._notif_iter = port.connection.create_notification_iterator()

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

    def test_init(self):
        class MyIter(bt2._UserNotificationIterator):
            def __init__(self):
                nonlocal initialized
                initialized = True

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

        initialized = False
        graph = self._create_graph(MySource)
        self.assertTrue(initialized)

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

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

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

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

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

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

    def test_addr(self):
        class MyIter(bt2._UserNotificationIterator):
            def __init__(self):
                nonlocal addr
                addr = self.addr

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

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


class GenericNotificationIteratorTestCase(unittest.TestCase):
    def test_component(self):
        class MyIter(bt2._UserNotificationIterator):
            pass

        class MySource(bt2._UserSourceComponent,
                       notification_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._notif_iter)

            def _port_connected(self, port, other_port):
                nonlocal upstream_comp
                self._notif_iter = port.connection.create_notification_iterator()
                upstream_comp = self._notif_iter.component

        upstream_comp = None
        graph = bt2.Graph()
        src_comp = graph.add_component(MySource, 'src')
        sink_comp = graph.add_component(MySink, 'sink')
        graph.connect_ports(src_comp.output_ports['out'],
                            sink_comp.input_ports['in'])
        self.assertEqual(src_comp, upstream_comp)
        del upstream_comp
Commit	Line	Data
811644b8 PP	1	from bt2 import values
	2	import unittest
	3	import copy
	4	import bt2
	5
	6
	7	class UserNotificationIteratorTestCase(unittest.TestCase):
	8	@staticmethod
	9	def _create_graph(src_comp_cls):
	10	class MySink(bt2._UserSinkComponent):
	11	def __init__(self, params):
	12	self._add_input_port('in')
	13
	14	def _consume(self):
	15	next(self._notif_iter)
	16
	17	def _port_connected(self, port, other_port):
	18	self._notif_iter = port.connection.create_notification_iterator()
	19
	20	graph = bt2.Graph()
	21	src_comp = graph.add_component(src_comp_cls, 'src')
	22	sink_comp = graph.add_component(MySink, 'sink')
	23	graph.connect_ports(src_comp.output_ports['out'],
	24	sink_comp.input_ports['in'])
	25	return graph
	26
	27	def test_init(self):
	28	class MyIter(bt2._UserNotificationIterator):
	29	def __init__(self):
	30	nonlocal initialized
	31	initialized = True
	32
	33	class MySource(bt2._UserSourceComponent,
	34	notification_iterator_class=MyIter):
	35	def __init__(self, params):
	36	self._add_output_port('out')
	37
	38	initialized = False
	39	graph = self._create_graph(MySource)
	40	self.assertTrue(initialized)
	41
	42	def test_finalize(self):
	43	class MyIter(bt2._UserNotificationIterator):
	44	def _finalize(self):
	45	nonlocal finalized
	46	finalized = True
	47
	48	class MySource(bt2._UserSourceComponent,
	49	notification_iterator_class=MyIter):
	50	def __init__(self, params):
	51	self._add_output_port('out')
	52
	53	finalized = False
	54	graph = self._create_graph(MySource)
	55	del graph
	56	self.assertTrue(finalized)
	57
	58	def test_component(self):
	59	class MyIter(bt2._UserNotificationIterator):
	60	def __init__(self):
	61	nonlocal salut
	62	salut = self._component._salut
	63
	64	class MySource(bt2._UserSourceComponent,
65	notification_iterator_class=MyIter):
66	def __init__(self, params):
67	self._add_output_port('out')
68	self._salut = 23
69
70	salut = None
71	graph = self._create_graph(MySource)
72	self.assertEqual(salut, 23)
73
74	def test_addr(self):
75	class MyIter(bt2._UserNotificationIterator):
76	def __init__(self):
77	nonlocal addr
78	addr = self.addr
79
80	class MySource(bt2._UserSourceComponent,
81	notification_iterator_class=MyIter):
82	def __init__(self, params):
83	self._add_output_port('out')
84
85	addr = None
86	graph = self._create_graph(MySource)
87	self.assertIsNotNone(addr)
88	self.assertNotEqual(addr, 0)
89
90
91	class GenericNotificationIteratorTestCase(unittest.TestCase):
92	def test_component(self):
93	class MyIter(bt2._UserNotificationIterator):
94	pass
95
96	class MySource(bt2._UserSourceComponent,
97	notification_iterator_class=MyIter):
98	def __init__(self, params):
99	self._add_output_port('out')
100
101	class MySink(bt2._UserSinkComponent):
102	def __init__(self, params):
103	self._add_input_port('in')
104
105	def _consume(self):
106	next(self._notif_iter)
107
108	def _port_connected(self, port, other_port):
109	nonlocal upstream_comp
110	self._notif_iter = port.connection.create_notification_iterator()
111	upstream_comp = self._notif_iter.component
112
113	upstream_comp = None
114	graph = bt2.Graph()
115	src_comp = graph.add_component(MySource, 'src')
116	sink_comp = graph.add_component(MySink, 'sink')
117	graph.connect_ports(src_comp.output_ports['out'],
118	sink_comp.input_ports['in'])
119	self.assertEqual(src_comp, upstream_comp)
120	del upstream_comp