Add configuration for python code formatter black
[babeltrace.git] / tests / bindings / python / bt2 / test_message_iterator.py
CommitLineData
c4239792 1from bt2 import value
548bb510 2import collections
811644b8
PP
3import unittest
4import copy
5import bt2
6
7
976c241d 8@unittest.skip("this is broken")
5602ef81 9class UserMessageIteratorTestCase(unittest.TestCase):
811644b8
PP
10 @staticmethod
11 def _create_graph(src_comp_cls):
12 class MySink(bt2._UserSinkComponent):
13 def __init__(self, params):
14 self._add_input_port('in')
15
16 def _consume(self):
5602ef81 17 next(self._msg_iter)
811644b8
PP
18
19 def _port_connected(self, port, other_port):
5602ef81 20 self._msg_iter = port.connection.create_message_iterator()
811644b8
PP
21
22 graph = bt2.Graph()
23 src_comp = graph.add_component(src_comp_cls, 'src')
24 sink_comp = graph.add_component(MySink, 'sink')
25 graph.connect_ports(src_comp.output_ports['out'],
26 sink_comp.input_ports['in'])
27 return graph
28
29 def test_init(self):
5602ef81 30 class MyIter(bt2._UserMessageIterator):
811644b8
PP
31 def __init__(self):
32 nonlocal initialized
33 initialized = True
34
35 class MySource(bt2._UserSourceComponent,
5602ef81 36 message_iterator_class=MyIter):
811644b8
PP
37 def __init__(self, params):
38 self._add_output_port('out')
39
40 initialized = False
41 graph = self._create_graph(MySource)
42 self.assertTrue(initialized)
43
44 def test_finalize(self):
5602ef81 45 class MyIter(bt2._UserMessageIterator):
811644b8
PP
46 def _finalize(self):
47 nonlocal finalized
48 finalized = True
49
50 class MySource(bt2._UserSourceComponent,
5602ef81 51 message_iterator_class=MyIter):
811644b8
PP
52 def __init__(self, params):
53 self._add_output_port('out')
54
55 finalized = False
56 graph = self._create_graph(MySource)
57 del graph
58 self.assertTrue(finalized)
59
60 def test_component(self):
5602ef81 61 class MyIter(bt2._UserMessageIterator):
811644b8
PP
62 def __init__(self):
63 nonlocal salut
64 salut = self._component._salut
65
66 class MySource(bt2._UserSourceComponent,
5602ef81 67 message_iterator_class=MyIter):
811644b8
PP
68 def __init__(self, params):
69 self._add_output_port('out')
70 self._salut = 23
71
72 salut = None
73 graph = self._create_graph(MySource)
74 self.assertEqual(salut, 23)
75
76 def test_addr(self):
5602ef81 77 class MyIter(bt2._UserMessageIterator):
811644b8
PP
78 def __init__(self):
79 nonlocal addr
80 addr = self.addr
81
82 class MySource(bt2._UserSourceComponent,
5602ef81 83 message_iterator_class=MyIter):
811644b8
PP
84 def __init__(self, params):
85 self._add_output_port('out')
86
87 addr = None
88 graph = self._create_graph(MySource)
89 self.assertIsNotNone(addr)
90 self.assertNotEqual(addr, 0)
91
92
976c241d 93@unittest.skip("this is broken")
5602ef81 94class PrivateConnectionMessageIteratorTestCase(unittest.TestCase):
811644b8 95 def test_component(self):
5602ef81 96 class MyIter(bt2._UserMessageIterator):
811644b8
PP
97 pass
98
99 class MySource(bt2._UserSourceComponent,
5602ef81 100 message_iterator_class=MyIter):
811644b8
PP
101 def __init__(self, params):
102 self._add_output_port('out')
103
104 class MySink(bt2._UserSinkComponent):
105 def __init__(self, params):
106 self._add_input_port('in')
107
108 def _consume(self):
5602ef81 109 next(self._msg_iter)
811644b8
PP
110
111 def _port_connected(self, port, other_port):
112 nonlocal upstream_comp
5602ef81
SM
113 self._msg_iter = port.connection.create_message_iterator()
114 upstream_comp = self._msg_iter.component
811644b8
PP
115
116 upstream_comp = None
117 graph = bt2.Graph()
118 src_comp = graph.add_component(MySource, 'src')
119 sink_comp = graph.add_component(MySink, 'sink')
120 graph.connect_ports(src_comp.output_ports['out'],
121 sink_comp.input_ports['in'])
122 self.assertEqual(src_comp, upstream_comp)
123 del upstream_comp
548bb510
PP
124
125
976c241d 126@unittest.skip("this is broken")
5602ef81 127class OutputPortMessageIteratorTestCase(unittest.TestCase):
548bb510 128 def test_component(self):
5602ef81 129 class MyIter(bt2._UserMessageIterator):
548bb510
PP
130 def __init__(self):
131 self._build_meta()
132 self._at = 0
133
134 def _build_meta(self):
135 self._trace = bt2.Trace()
136 self._sc = bt2.StreamClass()
137 self._ec = bt2.EventClass('salut')
b4f45851
SM
138 self._my_int_fc = bt2.IntegerFieldClass(32)
139 self._ec.payload_field_class = bt2.StructureFieldClass()
140 self._ec.payload_field_class += collections.OrderedDict([
141 ('my_int', self._my_int_fc),
548bb510
PP
142 ])
143 self._sc.add_event_class(self._ec)
144 self._trace.add_stream_class(self._sc)
145 self._stream = self._sc()
146 self._packet = self._stream.create_packet()
147
148 def _create_event(self, value):
149 ev = self._ec()
150 ev.payload_field['my_int'] = value
151 ev.packet = self._packet
152 return ev
153
154 def __next__(self):
155 if self._at == 5:
156 raise bt2.Stop
157
5602ef81 158 msg = bt2.EventMessage(self._create_event(self._at * 3))
548bb510 159 self._at += 1
5602ef81 160 return msg
548bb510
PP
161
162 class MySource(bt2._UserSourceComponent,
5602ef81 163 message_iterator_class=MyIter):
548bb510
PP
164 def __init__(self, params):
165 self._add_output_port('out')
166
167 graph = bt2.Graph()
168 src = graph.add_component(MySource, 'src')
5602ef81
SM
169 types = [bt2.EventMessage]
170 msg_iter = src.output_ports['out'].create_message_iterator(types)
548bb510 171
5602ef81
SM
172 for at, msg in enumerate(msg_iter):
173 self.assertIsInstance(msg, bt2.EventMessage)
174 self.assertEqual(msg.event.event_class.name, 'salut')
175 field = msg.event.payload_field['my_int']
548bb510 176 self.assertEqual(field, at * 3)
This page took 0.038408 seconds and 4 git commands to generate.