8 class UserMessageIteratorTestCase(unittest
.TestCase
):
10 def _create_graph(src_comp_cls
):
11 class MySink(bt2
._UserSinkComponent
):
12 def __init__(self
, params
):
13 self
._add
_input
_port
('in')
18 def _graph_is_configured(self
):
19 self
._msg
_iter
= self
._input
_ports
['in'].create_message_iterator()
22 src_comp
= graph
.add_component(src_comp_cls
, 'src')
23 sink_comp
= graph
.add_component(MySink
, 'sink')
24 graph
.connect_ports(src_comp
.output_ports
['out'],
25 sink_comp
.input_ports
['in'])
29 the_output_port_from_source
= None
30 the_output_port_from_iter
= None
32 class MyIter(bt2
._UserMessageIterator
):
33 def __init__(self
, self_port_output
):
35 nonlocal the_output_port_from_iter
37 the_output_port_from_iter
= self_port_output
39 class MySource(bt2
._UserSourceComponent
,
40 message_iterator_class
=MyIter
):
41 def __init__(self
, params
):
42 nonlocal the_output_port_from_source
43 the_output_port_from_source
= self
._add
_output
_port
('out')
46 graph
= self
._create
_graph
(MySource
)
48 self
.assertTrue(initialized
)
49 self
.assertEqual(the_output_port_from_source
.addr
, the_output_port_from_iter
.addr
)
51 def test_finalize(self
):
52 class MyIter(bt2
._UserMessageIterator
):
57 class MySource(bt2
._UserSourceComponent
,
58 message_iterator_class
=MyIter
):
59 def __init__(self
, params
):
60 self
._add
_output
_port
('out')
63 graph
= self
._create
_graph
(MySource
)
66 self
.assertTrue(finalized
)
68 def test_component(self
):
69 class MyIter(bt2
._UserMessageIterator
):
70 def __init__(self
, self_port_output
):
72 salut
= self
._component
._salut
74 class MySource(bt2
._UserSourceComponent
,
75 message_iterator_class
=MyIter
):
76 def __init__(self
, params
):
77 self
._add
_output
_port
('out')
81 graph
= self
._create
_graph
(MySource
)
83 self
.assertEqual(salut
, 23)
86 class MyIter(bt2
._UserMessageIterator
):
87 def __init__(self
, self_port_output
):
91 class MySource(bt2
._UserSourceComponent
,
92 message_iterator_class
=MyIter
):
93 def __init__(self
, params
):
94 self
._add
_output
_port
('out')
97 graph
= self
._create
_graph
(MySource
)
99 self
.assertIsNotNone(addr
)
100 self
.assertNotEqual(addr
, 0)
103 class OutputPortMessageIteratorTestCase(unittest
.TestCase
):
104 def test_component(self
):
105 class MyIter(bt2
._UserMessageIterator
):
106 def __init__(self
, self_port_output
):
114 msg
= self
._create
_stream
_beginning
_message
(test_obj
._stream
)
116 msg
= self
._create
_packet
_beginning
_message
(test_obj
._packet
)
118 msg
= self
._create
_packet
_end
_message
(test_obj
._packet
)
120 msg
= self
._create
_stream
_end
_message
(test_obj
._stream
)
122 msg
= self
._create
_event
_message
(test_obj
._event
_class
, test_obj
._packet
)
123 msg
.event
.payload_field
['my_int'] = self
._at
* 3
128 class MySource(bt2
._UserSourceComponent
,
129 message_iterator_class
=MyIter
):
130 def __init__(self
, params
):
131 self
._add
_output
_port
('out')
133 trace_class
= self
._create
_trace
_class
()
134 stream_class
= trace_class
.create_stream_class()
136 # Create payload field class
137 my_int_ft
= trace_class
.create_signed_integer_field_class(32)
138 payload_ft
= trace_class
.create_structure_field_class()
139 payload_ft
+= collections
.OrderedDict([
140 ('my_int', my_int_ft
),
143 event_class
= stream_class
.create_event_class(name
='salut', payload_field_class
=payload_ft
)
145 trace
= trace_class()
146 stream
= trace
.create_stream(stream_class
)
147 packet
= stream
.create_packet()
149 test_obj
._event
_class
= event_class
150 test_obj
._stream
= stream
151 test_obj
._packet
= packet
155 src
= graph
.add_component(MySource
, 'src')
156 msg_iter
= graph
.create_output_port_message_iterator(src
.output_ports
['out'])
158 for at
, msg
in enumerate(msg_iter
):
160 self
.assertIsInstance(msg
, bt2
.message
._StreamBeginningMessage
)
162 self
.assertIsInstance(msg
, bt2
.message
._PacketBeginningMessage
)
164 self
.assertIsInstance(msg
, bt2
.message
._PacketEndMessage
)
166 self
.assertIsInstance(msg
, bt2
.message
._StreamEndMessage
)
168 self
.assertIsInstance(msg
, bt2
.message
._EventMessage
)
169 self
.assertEqual(msg
.event
.event_class
.name
, 'salut')
170 field
= msg
.event
.payload_field
['my_int']
171 self
.assertEqual(field
, at
* 3)