2 # Copyright (C) 2019 EfficiOS Inc.
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
19 from collections
import OrderedDict
24 class EventTestCase(unittest
.TestCase
):
25 def _create_test_event_message(self
, packet_fields_config
=None,
26 event_fields_config
=None,
27 with_clockclass
=False,
28 with_cc
=False, with_sc
=False,
31 class MyIter(bt2
._UserMessageIterator
):
32 def __init__(self
, self_output_port
):
37 msg
= self
._create
_stream
_beginning
_message
(test_obj
.stream
)
39 assert test_obj
.packet
40 msg
= self
._create
_packet
_beginning
_message
(test_obj
.packet
)
42 default_clock_snapshot
= 789 if with_clockclass
else None
43 assert test_obj
.packet
44 msg
= self
._create
_event
_message
(test_obj
.event_class
, test_obj
.packet
, default_clock_snapshot
)
45 if event_fields_config
is not None:
46 event_fields_config(msg
.event
)
48 msg
= self
._create
_packet
_end
_message
(test_obj
.packet
)
50 msg
= self
._create
_stream
_end
_message
(test_obj
.stream
)
57 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
58 def __init__(self
, params
):
59 self
._add
_output
_port
('out')
60 tc
= self
._create
_trace
_class
()
64 clock_class
= self
._create
_clock
_class
(frequency
=1000)
66 # event common context (stream-class-defined)
69 cc
= tc
.create_structure_field_class()
71 ('cpu_id', tc
.create_signed_integer_field_class(8)),
72 ('stuff', tc
.create_real_field_class()),
75 # packet context (stream-class-defined)
76 pc
= tc
.create_structure_field_class()
78 ('something', tc
.create_unsigned_integer_field_class(8)),
79 ('something_else', tc
.create_real_field_class()),
82 stream_class
= tc
.create_stream_class(default_clock_class
=clock_class
,
83 event_common_context_field_class
=cc
,
84 packet_context_field_class
=pc
)
86 # specific context (event-class-defined)
89 sc
= tc
.create_structure_field_class()
91 ('ant', tc
.create_signed_integer_field_class(16)),
92 ('msg', tc
.create_string_field_class()),
98 ep
= tc
.create_structure_field_class()
100 ('giraffe', tc
.create_signed_integer_field_class(32)),
101 ('gnu', tc
.create_signed_integer_field_class(8)),
102 ('mosquito', tc
.create_signed_integer_field_class(8)),
105 event_class
= stream_class
.create_event_class(name
='garou',
106 specific_context_field_class
=sc
,
107 payload_field_class
=ep
)
110 stream
= trace
.create_stream(stream_class
)
111 packet
= stream
.create_packet()
113 if packet_fields_config
is not None:
114 packet_fields_config(packet
)
116 test_obj
.packet
= packet
117 test_obj
.stream
= stream
118 test_obj
.event_class
= event_class
121 self
._graph
= bt2
.Graph()
122 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
123 self
._msg
_iter
= self
._graph
.create_output_port_message_iterator(self
._src
_comp
.output_ports
['out'])
125 for i
, msg
in enumerate(self
._msg
_iter
):
129 def test_attr_event_class(self
):
130 msg
= self
._create
_test
_event
_message
()
131 self
.assertEqual(msg
.event
.cls
.addr
, self
.event_class
.addr
)
133 def test_attr_name(self
):
134 msg
= self
._create
_test
_event
_message
()
135 self
.assertEqual(msg
.event
.name
, self
.event_class
.name
)
137 def test_attr_id(self
):
138 msg
= self
._create
_test
_event
_message
()
139 self
.assertEqual(msg
.event
.id, self
.event_class
.id)
141 def test_get_common_context_field(self
):
142 def event_fields_config(event
):
143 event
.common_context_field
['cpu_id'] = 1
144 event
.common_context_field
['stuff'] = 13.194
146 msg
= self
._create
_test
_event
_message
(event_fields_config
=event_fields_config
, with_cc
=True)
148 self
.assertEqual(msg
.event
.common_context_field
['cpu_id'], 1)
149 self
.assertEqual(msg
.event
.common_context_field
['stuff'], 13.194)
151 def test_no_common_context_field(self
):
152 msg
= self
._create
_test
_event
_message
(with_cc
=False)
153 self
.assertIsNone(msg
.event
.common_context_field
)
155 def test_get_specific_context_field(self
):
156 def event_fields_config(event
):
157 event
.specific_context_field
['ant'] = -1
158 event
.specific_context_field
['msg'] = 'hellooo'
160 msg
= self
._create
_test
_event
_message
(event_fields_config
=event_fields_config
, with_sc
=True)
162 self
.assertEqual(msg
.event
.specific_context_field
['ant'], -1)
163 self
.assertEqual(msg
.event
.specific_context_field
['msg'], 'hellooo')
165 def test_no_specific_context_field(self
):
166 msg
= self
._create
_test
_event
_message
(with_sc
=False)
167 self
.assertIsNone(msg
.event
.specific_context_field
)
169 def test_get_event_payload_field(self
):
170 def event_fields_config(event
):
171 event
.payload_field
['giraffe'] = 1
172 event
.payload_field
['gnu'] = 23
173 event
.payload_field
['mosquito'] = 42
175 msg
= self
._create
_test
_event
_message
(event_fields_config
=event_fields_config
, with_ep
=True)
177 self
.assertEqual(msg
.event
.payload_field
['giraffe'], 1)
178 self
.assertEqual(msg
.event
.payload_field
['gnu'], 23)
179 self
.assertEqual(msg
.event
.payload_field
['mosquito'], 42)
181 def test_no_payload_field(self
):
182 msg
= self
._create
_test
_event
_message
(with_ep
=False)
183 self
.assertIsNone(msg
.event
.payload_field
)
185 def test_clock_value(self
):
186 msg
= self
._create
_test
_event
_message
(with_clockclass
=True)
187 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
189 def test_no_clock_value(self
):
190 msg
= self
._create
_test
_event
_message
(with_clockclass
=False)
191 with self
.assertRaises(bt2
.NonexistentClockSnapshot
):
192 msg
.default_clock_snapshot
194 def test_stream(self
):
195 msg
= self
._create
_test
_event
_message
()
196 self
.assertEqual(msg
.event
.stream
.addr
, self
.stream
.addr
)
198 def test_getitem(self
):
199 def event_fields_config(event
):
200 event
.payload_field
['giraffe'] = 1
201 event
.payload_field
['gnu'] = 23
202 event
.payload_field
['mosquito'] = 42
203 event
.specific_context_field
['ant'] = -1
204 event
.specific_context_field
['msg'] = 'hellooo'
205 event
.common_context_field
['cpu_id'] = 1
206 event
.common_context_field
['stuff'] = 13.194
208 def packet_fields_config(packet
):
209 packet
.context_field
['something'] = 154
210 packet
.context_field
['something_else'] = 17.2
212 msg
= self
._create
_test
_event
_message
(packet_fields_config
=packet_fields_config
,
213 event_fields_config
=event_fields_config
,
214 with_cc
=True, with_sc
=True, with_ep
=True)
218 self
.assertEqual(ev
['giraffe'], 1)
219 self
.assertEqual(ev
['gnu'], 23)
220 self
.assertEqual(ev
['mosquito'], 42)
221 self
.assertEqual(ev
['ant'], -1)
222 self
.assertEqual(ev
['msg'], 'hellooo')
223 self
.assertEqual(ev
['cpu_id'], 1)
224 self
.assertEqual(ev
['stuff'], 13.194)
227 self
.assertEqual(ev
['something'], 154)
228 self
.assertEqual(ev
['something_else'], 17.2)
230 with self
.assertRaises(KeyError):
234 if __name__
== "__main__":