1 from collections
import OrderedDict
6 class EventTestCase(unittest
.TestCase
):
7 def _create_test_event_message(self
, packet_fields_config
=None,
8 event_fields_config
=None,
10 with_cc
=False, with_sc
=False,
13 class MyIter(bt2
._UserMessageIterator
):
19 msg
= self
._create
_stream
_beginning
_message
(test_obj
.stream
)
21 assert test_obj
.packet
22 msg
= self
._create
_packet
_beginning
_message
(test_obj
.packet
)
24 default_clock_snapshot
= 789 if with_clockclass
else None
25 assert test_obj
.packet
26 msg
= self
._create
_event
_message
(test_obj
.event_class
, test_obj
.packet
, default_clock_snapshot
)
27 if event_fields_config
is not None:
28 event_fields_config(msg
.event
)
30 msg
= self
._create
_packet
_end
_message
(test_obj
.packet
)
32 msg
= self
._create
_stream
_end
_message
(test_obj
.stream
)
39 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
40 def __init__(self
, params
):
41 self
._add
_output
_port
('out')
42 tc
= self
._create
_trace
_class
()
46 clock_class
= self
._create
_clock
_class
(frequency
=1000)
48 # event common context (stream-class-defined)
51 cc
= tc
.create_structure_field_class()
53 ('cpu_id', tc
.create_signed_integer_field_class(8)),
54 ('stuff', tc
.create_real_field_class()),
57 # packet context (stream-class-defined)
58 pc
= tc
.create_structure_field_class()
60 ('something', tc
.create_unsigned_integer_field_class(8)),
61 ('something_else', tc
.create_real_field_class()),
64 stream_class
= tc
.create_stream_class(default_clock_class
=clock_class
,
65 event_common_context_field_class
=cc
,
66 packet_context_field_class
=pc
)
68 # specific context (event-class-defined)
71 sc
= tc
.create_structure_field_class()
73 ('ant', tc
.create_signed_integer_field_class(16)),
74 ('msg', tc
.create_string_field_class()),
80 ep
= tc
.create_structure_field_class()
82 ('giraffe', tc
.create_signed_integer_field_class(32)),
83 ('gnu', tc
.create_signed_integer_field_class(8)),
84 ('mosquito', tc
.create_signed_integer_field_class(8)),
87 event_class
= stream_class
.create_event_class(name
='garou',
88 specific_context_field_class
=sc
,
89 payload_field_class
=ep
)
92 stream
= trace
.create_stream(stream_class
)
93 packet
= stream
.create_packet()
95 if packet_fields_config
is not None:
96 packet_fields_config(packet
)
98 test_obj
.packet
= packet
99 test_obj
.stream
= stream
100 test_obj
.event_class
= event_class
103 self
._graph
= bt2
.Graph()
104 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
105 self
._msg
_iter
= self
._graph
.create_output_port_message_iterator(self
._src
_comp
.output_ports
['out'])
107 for i
, msg
in enumerate(self
._msg
_iter
):
111 def test_attr_event_class(self
):
112 msg
= self
._create
_test
_event
_message
()
113 self
.assertEqual(msg
.event
.event_class
.addr
, self
.event_class
.addr
)
115 def test_attr_name(self
):
116 msg
= self
._create
_test
_event
_message
()
117 self
.assertEqual(msg
.event
.name
, self
.event_class
.name
)
119 def test_attr_id(self
):
120 msg
= self
._create
_test
_event
_message
()
121 self
.assertEqual(msg
.event
.id, self
.event_class
.id)
123 def test_get_common_context_field(self
):
124 def event_fields_config(event
):
125 event
.common_context_field
['cpu_id'] = 1
126 event
.common_context_field
['stuff'] = 13.194
128 msg
= self
._create
_test
_event
_message
(event_fields_config
=event_fields_config
, with_cc
=True)
130 self
.assertEqual(msg
.event
.common_context_field
['cpu_id'], 1)
131 self
.assertEqual(msg
.event
.common_context_field
['stuff'], 13.194)
133 def test_no_common_context_field(self
):
134 msg
= self
._create
_test
_event
_message
(with_cc
=False)
135 self
.assertIsNone(msg
.event
.common_context_field
)
137 def test_get_specific_context_field(self
):
138 def event_fields_config(event
):
139 event
.specific_context_field
['ant'] = -1
140 event
.specific_context_field
['msg'] = 'hellooo'
142 msg
= self
._create
_test
_event
_message
(event_fields_config
=event_fields_config
, with_sc
=True)
144 self
.assertEqual(msg
.event
.specific_context_field
['ant'], -1)
145 self
.assertEqual(msg
.event
.specific_context_field
['msg'], 'hellooo')
147 def test_no_specific_context_field(self
):
148 msg
= self
._create
_test
_event
_message
(with_sc
=False)
149 self
.assertIsNone(msg
.event
.specific_context_field
)
151 def test_get_event_payload_field(self
):
152 def event_fields_config(event
):
153 event
.payload_field
['giraffe'] = 1
154 event
.payload_field
['gnu'] = 23
155 event
.payload_field
['mosquito'] = 42
157 msg
= self
._create
_test
_event
_message
(event_fields_config
=event_fields_config
, with_ep
=True)
159 self
.assertEqual(msg
.event
.payload_field
['giraffe'], 1)
160 self
.assertEqual(msg
.event
.payload_field
['gnu'], 23)
161 self
.assertEqual(msg
.event
.payload_field
['mosquito'], 42)
163 def test_no_payload_field(self
):
164 msg
= self
._create
_test
_event
_message
(with_ep
=False)
165 self
.assertIsNone(msg
.event
.payload_field
)
167 def test_clock_value(self
):
168 msg
= self
._create
_test
_event
_message
(with_clockclass
=True)
169 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
171 def test_no_clock_value(self
):
172 msg
= self
._create
_test
_event
_message
(with_clockclass
=False)
173 self
.assertIsNone(msg
.default_clock_snapshot
)
175 def test_stream(self
):
176 msg
= self
._create
_test
_event
_message
()
177 self
.assertEqual(msg
.event
.stream
.addr
, self
.stream
.addr
)
179 def test_getitem(self
):
180 def event_fields_config(event
):
181 event
.payload_field
['giraffe'] = 1
182 event
.payload_field
['gnu'] = 23
183 event
.payload_field
['mosquito'] = 42
184 event
.specific_context_field
['ant'] = -1
185 event
.specific_context_field
['msg'] = 'hellooo'
186 event
.common_context_field
['cpu_id'] = 1
187 event
.common_context_field
['stuff'] = 13.194
189 def packet_fields_config(packet
):
190 packet
.context_field
['something'] = 154
191 packet
.context_field
['something_else'] = 17.2
193 msg
= self
._create
_test
_event
_message
(packet_fields_config
=packet_fields_config
,
194 event_fields_config
=event_fields_config
,
195 with_cc
=True, with_sc
=True, with_ep
=True)
199 self
.assertEqual(ev
['giraffe'], 1)
200 self
.assertEqual(ev
['gnu'], 23)
201 self
.assertEqual(ev
['mosquito'], 42)
202 self
.assertEqual(ev
['ant'], -1)
203 self
.assertEqual(ev
['msg'], 'hellooo')
204 self
.assertEqual(ev
['cpu_id'], 1)
205 self
.assertEqual(ev
['stuff'], 13.194)
208 self
.assertEqual(ev
['something'], 154)
209 self
.assertEqual(ev
['something_else'], 17.2)
211 with self
.assertRaises(KeyError):
215 if __name__
== "__main__":