lib: fix compilation for glib < 2.40
[babeltrace.git] / tests / bindings / python / bt2 / test_message_iterator.py
... / ...
CommitLineData
1
2#
3# Copyright (C) 2019 EfficiOS Inc.
4#
5# This program is free software; you can redistribute it and/or
6# modify it under the terms of the GNU General Public License
7# as published by the Free Software Foundation; only version 2
8# of the License.
9#
10# This program is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with this program; if not, write to the Free Software
17# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
18#
19
20from bt2 import value
21import collections
22import unittest
23import copy
24import bt2
25
26
27class UserMessageIteratorTestCase(unittest.TestCase):
28 @staticmethod
29 def _create_graph(src_comp_cls):
30 class MySink(bt2._UserSinkComponent):
31 def __init__(self, params):
32 self._add_input_port('in')
33
34 def _consume(self):
35 next(self._msg_iter)
36
37 def _graph_is_configured(self):
38 self._msg_iter = self._input_ports['in'].create_message_iterator()
39
40 graph = bt2.Graph()
41 src_comp = graph.add_component(src_comp_cls, 'src')
42 sink_comp = graph.add_component(MySink, 'sink')
43 graph.connect_ports(src_comp.output_ports['out'],
44 sink_comp.input_ports['in'])
45 return graph
46
47 def test_init(self):
48 the_output_port_from_source = None
49 the_output_port_from_iter = None
50
51 class MyIter(bt2._UserMessageIterator):
52 def __init__(self, self_port_output):
53 nonlocal initialized
54 nonlocal the_output_port_from_iter
55 initialized = True
56 the_output_port_from_iter = self_port_output
57
58 class MySource(bt2._UserSourceComponent,
59 message_iterator_class=MyIter):
60 def __init__(self, params):
61 nonlocal the_output_port_from_source
62 the_output_port_from_source = self._add_output_port('out', 'user data')
63
64 initialized = False
65 graph = self._create_graph(MySource)
66 graph.run()
67 self.assertTrue(initialized)
68 self.assertEqual(the_output_port_from_source.addr, the_output_port_from_iter.addr)
69 self.assertEqual(the_output_port_from_iter.user_data, 'user data')
70
71 def test_finalize(self):
72 class MyIter(bt2._UserMessageIterator):
73 def _finalize(self):
74 nonlocal finalized
75 finalized = True
76
77 class MySource(bt2._UserSourceComponent,
78 message_iterator_class=MyIter):
79 def __init__(self, params):
80 self._add_output_port('out')
81
82 finalized = False
83 graph = self._create_graph(MySource)
84 graph.run()
85 del graph
86 self.assertTrue(finalized)
87
88 def test_component(self):
89 class MyIter(bt2._UserMessageIterator):
90 def __init__(self, self_port_output):
91 nonlocal salut
92 salut = self._component._salut
93
94 class MySource(bt2._UserSourceComponent,
95 message_iterator_class=MyIter):
96 def __init__(self, params):
97 self._add_output_port('out')
98 self._salut = 23
99
100 salut = None
101 graph = self._create_graph(MySource)
102 graph.run()
103 self.assertEqual(salut, 23)
104
105 def test_addr(self):
106 class MyIter(bt2._UserMessageIterator):
107 def __init__(self, self_port_output):
108 nonlocal addr
109 addr = self.addr
110
111 class MySource(bt2._UserSourceComponent,
112 message_iterator_class=MyIter):
113 def __init__(self, params):
114 self._add_output_port('out')
115
116 addr = None
117 graph = self._create_graph(MySource)
118 graph.run()
119 self.assertIsNotNone(addr)
120 self.assertNotEqual(addr, 0)
121
122
123class OutputPortMessageIteratorTestCase(unittest.TestCase):
124 def test_component(self):
125 class MyIter(bt2._UserMessageIterator):
126 def __init__(self, self_port_output):
127 self._at = 0
128
129 def __next__(self):
130 if self._at == 7:
131 raise bt2.Stop
132
133 if self._at == 0:
134 msg = self._create_stream_beginning_message(test_obj._stream)
135 elif self._at == 1:
136 msg = self._create_packet_beginning_message(test_obj._packet)
137 elif self._at == 5:
138 msg = self._create_packet_end_message(test_obj._packet)
139 elif self._at == 6:
140 msg = self._create_stream_end_message(test_obj._stream)
141 else:
142 msg = self._create_event_message(test_obj._event_class, test_obj._packet)
143 msg.event.payload_field['my_int'] = self._at * 3
144
145 self._at += 1
146 return msg
147
148 class MySource(bt2._UserSourceComponent,
149 message_iterator_class=MyIter):
150 def __init__(self, params):
151 self._add_output_port('out')
152
153 trace_class = self._create_trace_class()
154 stream_class = trace_class.create_stream_class()
155
156 # Create payload field class
157 my_int_ft = trace_class.create_signed_integer_field_class(32)
158 payload_ft = trace_class.create_structure_field_class()
159 payload_ft += collections.OrderedDict([
160 ('my_int', my_int_ft),
161 ])
162
163 event_class = stream_class.create_event_class(name='salut', payload_field_class=payload_ft)
164
165 trace = trace_class()
166 stream = trace.create_stream(stream_class)
167 packet = stream.create_packet()
168
169 test_obj._event_class = event_class
170 test_obj._stream = stream
171 test_obj._packet = packet
172
173 test_obj = self
174 graph = bt2.Graph()
175 src = graph.add_component(MySource, 'src')
176 msg_iter = graph.create_output_port_message_iterator(src.output_ports['out'])
177
178 for at, msg in enumerate(msg_iter):
179 if at == 0:
180 self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
181 elif at == 1:
182 self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
183 elif at == 5:
184 self.assertIsInstance(msg, bt2.message._PacketEndMessage)
185 elif at == 6:
186 self.assertIsInstance(msg, bt2.message._StreamEndMessage)
187 else:
188 self.assertIsInstance(msg, bt2.message._EventMessage)
189 self.assertEqual(msg.event.cls.name, 'salut')
190 field = msg.event.payload_field['my_int']
191 self.assertEqual(field, at * 3)
This page took 0.024724 seconds and 4 git commands to generate.