3 # Copyright (C) 2019 EfficiOS Inc.
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
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.
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.
27 class UserMessageIteratorTestCase(unittest
.TestCase
):
29 def _create_graph(src_comp_cls
):
30 class MySink(bt2
._UserSinkComponent
):
31 def __init__(self
, params
):
32 self
._add
_input
_port
('in')
37 def _graph_is_configured(self
):
38 self
._msg
_iter
= self
._input
_ports
['in'].create_message_iterator()
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'])
48 the_output_port_from_source
= None
49 the_output_port_from_iter
= None
51 class MyIter(bt2
._UserMessageIterator
):
52 def __init__(self
, self_port_output
):
54 nonlocal the_output_port_from_iter
56 the_output_port_from_iter
= self_port_output
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')
65 graph
= self
._create
_graph
(MySource
)
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')
71 def test_finalize(self
):
72 class MyIter(bt2
._UserMessageIterator
):
77 class MySource(bt2
._UserSourceComponent
,
78 message_iterator_class
=MyIter
):
79 def __init__(self
, params
):
80 self
._add
_output
_port
('out')
83 graph
= self
._create
_graph
(MySource
)
86 self
.assertTrue(finalized
)
88 def test_component(self
):
89 class MyIter(bt2
._UserMessageIterator
):
90 def __init__(self
, self_port_output
):
92 salut
= self
._component
._salut
94 class MySource(bt2
._UserSourceComponent
,
95 message_iterator_class
=MyIter
):
96 def __init__(self
, params
):
97 self
._add
_output
_port
('out')
101 graph
= self
._create
_graph
(MySource
)
103 self
.assertEqual(salut
, 23)
106 class MyIter(bt2
._UserMessageIterator
):
107 def __init__(self
, self_port_output
):
111 class MySource(bt2
._UserSourceComponent
,
112 message_iterator_class
=MyIter
):
113 def __init__(self
, params
):
114 self
._add
_output
_port
('out')
117 graph
= self
._create
_graph
(MySource
)
119 self
.assertIsNotNone(addr
)
120 self
.assertNotEqual(addr
, 0)
123 class OutputPortMessageIteratorTestCase(unittest
.TestCase
):
124 def test_component(self
):
125 class MyIter(bt2
._UserMessageIterator
):
126 def __init__(self
, self_port_output
):
134 msg
= self
._create
_stream
_beginning
_message
(test_obj
._stream
)
136 msg
= self
._create
_packet
_beginning
_message
(test_obj
._packet
)
138 msg
= self
._create
_packet
_end
_message
(test_obj
._packet
)
140 msg
= self
._create
_stream
_end
_message
(test_obj
._stream
)
142 msg
= self
._create
_event
_message
(test_obj
._event
_class
, test_obj
._packet
)
143 msg
.event
.payload_field
['my_int'] = self
._at
* 3
148 class MySource(bt2
._UserSourceComponent
,
149 message_iterator_class
=MyIter
):
150 def __init__(self
, params
):
151 self
._add
_output
_port
('out')
153 trace_class
= self
._create
_trace
_class
()
154 stream_class
= trace_class
.create_stream_class()
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
),
163 event_class
= stream_class
.create_event_class(name
='salut', payload_field_class
=payload_ft
)
165 trace
= trace_class()
166 stream
= trace
.create_stream(stream_class
)
167 packet
= stream
.create_packet()
169 test_obj
._event
_class
= event_class
170 test_obj
._stream
= stream
171 test_obj
._packet
= packet
175 src
= graph
.add_component(MySource
, 'src')
176 msg_iter
= graph
.create_output_port_message_iterator(src
.output_ports
['out'])
178 for at
, msg
in enumerate(msg_iter
):
180 self
.assertIsInstance(msg
, bt2
.message
._StreamBeginningMessage
)
182 self
.assertIsInstance(msg
, bt2
.message
._PacketBeginningMessage
)
184 self
.assertIsInstance(msg
, bt2
.message
._PacketEndMessage
)
186 self
.assertIsInstance(msg
, bt2
.message
._StreamEndMessage
)
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)