8 class _NotificationTestCase(unittest
.TestCase
):
10 self
._trace
= bt2
.Trace()
11 self
._sc
= bt2
.StreamClass()
12 self
._ec
= bt2
.EventClass('salut')
13 self
._my
_int
_ft
= bt2
.IntegerFieldType(32)
14 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
15 self
._ec
.payload_field_type
+= collections
.OrderedDict([
16 ('my_int', self
._my
_int
_ft
),
18 self
._sc
.add_event_class(self
._ec
)
19 self
._clock
_class
= bt2
.ClockClass('allo', 1000)
20 self
._trace
.add_clock_class(self
._clock
_class
)
21 self
._trace
.packet_header_field_type
= bt2
.StructureFieldType()
22 self
._trace
.packet_header_field_type
+= collections
.OrderedDict([
23 ('hello', self
._my
_int
_ft
),
25 self
._trace
.add_stream_class(self
._sc
)
26 self
._cc
_prio
_map
= bt2
.ClockClassPriorityMap()
27 self
._cc
_prio
_map
[self
._clock
_class
] = 231
28 self
._stream
= self
._sc
()
29 self
._packet
= self
._stream
.create_packet()
30 self
._packet
.header_field
['hello'] = 19487
31 self
._event
= self
._ec
()
32 self
._event
.add_clock_value(self
._clock
_class
(1772))
33 self
._event
.payload_field
['my_int'] = 23
34 self
._event
.packet
= self
._packet
48 class EventNotificationTestCase(_NotificationTestCase
):
49 def test_create_no_cc_prio_map(self
):
50 notif
= bt2
.EventNotification(self
._event
)
51 self
.assertEqual(notif
.event
.addr
, self
._event
.addr
)
52 self
.assertEqual(len(notif
.clock_class_priority_map
), 0)
54 def test_create_with_cc_prio_map(self
):
55 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
56 self
.assertEqual(notif
.event
.addr
, self
._event
.addr
)
57 self
.assertEqual(len(notif
.clock_class_priority_map
), 1)
58 self
.assertEqual(notif
.clock_class_priority_map
.highest_priority_clock_class
.addr
,
59 self
._clock
_class
.addr
)
60 self
.assertEqual(notif
.clock_class_priority_map
[self
._clock
_class
], 231)
63 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
64 event_copy
= copy
.copy(self
._event
)
65 event_copy
.packet
= self
._packet
66 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
67 notif2
= bt2
.EventNotification(event_copy
, cc_prio_map_copy
)
68 self
.assertEqual(notif
, notif2
)
70 def test_ne_event(self
):
71 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
72 event_copy
= copy
.copy(self
._event
)
73 event_copy
.payload_field
['my_int'] = 17
74 event_copy
.packet
= self
._packet
75 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
76 notif2
= bt2
.EventNotification(event_copy
, cc_prio_map_copy
)
77 self
.assertNotEqual(notif
, notif2
)
79 def test_ne_cc_prio_map(self
):
80 notif
= bt2
.EventNotification(self
._event
)
81 event_copy
= copy
.copy(self
._event
)
82 event_copy
.packet
= self
._packet
83 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
84 notif2
= bt2
.EventNotification(event_copy
, cc_prio_map_copy
)
85 self
.assertNotEqual(notif
, notif2
)
87 def test_eq_invalid(self
):
88 notif
= bt2
.EventNotification(self
._event
)
89 self
.assertNotEqual(notif
, 23)
92 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
93 notif2
= copy
.copy(notif
)
94 self
.assertEqual(notif
, notif2
)
96 def test_deepcopy(self
):
97 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
98 notif2
= copy
.deepcopy(notif
)
99 self
.assertEqual(notif
, notif2
)
102 class PacketBeginningNotificationTestCase(_NotificationTestCase
):
103 def test_create(self
):
104 notif
= bt2
.PacketBeginningNotification(self
._packet
)
105 self
.assertEqual(notif
.packet
.addr
, self
._packet
.addr
)
108 notif
= bt2
.PacketBeginningNotification(self
._packet
)
109 packet_copy
= copy
.copy(self
._packet
)
110 notif2
= bt2
.PacketBeginningNotification(packet_copy
)
111 self
.assertEqual(notif
, notif2
)
113 def test_ne_packet(self
):
114 notif
= bt2
.PacketBeginningNotification(self
._packet
)
115 packet_copy
= copy
.copy(self
._packet
)
116 packet_copy
.header_field
['hello'] = 1847
117 notif2
= bt2
.PacketBeginningNotification(packet_copy
)
118 self
.assertNotEqual(notif
, notif2
)
120 def test_eq_invalid(self
):
121 notif
= bt2
.PacketBeginningNotification(self
._packet
)
122 self
.assertNotEqual(notif
, 23)
125 notif
= bt2
.PacketBeginningNotification(self
._packet
)
126 notif2
= copy
.copy(notif
)
127 self
.assertEqual(notif
, notif2
)
129 def test_deepcopy(self
):
130 notif
= bt2
.PacketBeginningNotification(self
._packet
)
131 notif2
= copy
.deepcopy(notif
)
132 self
.assertEqual(notif
, notif2
)
135 class PacketEndNotificationTestCase(_NotificationTestCase
):
136 def test_create(self
):
137 notif
= bt2
.PacketEndNotification(self
._packet
)
138 self
.assertEqual(notif
.packet
.addr
, self
._packet
.addr
)
141 notif
= bt2
.PacketEndNotification(self
._packet
)
142 packet_copy
= copy
.copy(self
._packet
)
143 notif2
= bt2
.PacketEndNotification(packet_copy
)
144 self
.assertEqual(notif
, notif2
)
146 def test_ne_packet(self
):
147 notif
= bt2
.PacketEndNotification(self
._packet
)
148 packet_copy
= copy
.copy(self
._packet
)
149 packet_copy
.header_field
['hello'] = 1847
150 notif2
= bt2
.PacketEndNotification(packet_copy
)
151 self
.assertNotEqual(notif
, notif2
)
153 def test_eq_invalid(self
):
154 notif
= bt2
.PacketEndNotification(self
._packet
)
155 self
.assertNotEqual(notif
, 23)
158 notif
= bt2
.PacketEndNotification(self
._packet
)
159 notif2
= copy
.copy(notif
)
160 self
.assertEqual(notif
, notif2
)
162 def test_deepcopy(self
):
163 notif
= bt2
.PacketEndNotification(self
._packet
)
164 notif2
= copy
.deepcopy(notif
)
165 self
.assertEqual(notif
, notif2
)
168 class StreamBeginningNotificationTestCase(_NotificationTestCase
):
169 def test_create(self
):
170 notif
= bt2
.StreamBeginningNotification(self
._stream
)
171 self
.assertEqual(notif
.stream
.addr
, self
._stream
.addr
)
174 notif
= bt2
.StreamBeginningNotification(self
._stream
)
175 stream_copy
= copy
.copy(self
._stream
)
176 notif2
= bt2
.StreamBeginningNotification(stream_copy
)
177 self
.assertEqual(notif
, notif2
)
179 def test_ne_stream(self
):
180 notif
= bt2
.StreamBeginningNotification(self
._stream
)
181 stream_copy
= self
._sc
(name
='salut')
182 notif2
= bt2
.StreamBeginningNotification(stream_copy
)
183 self
.assertNotEqual(notif
, notif2
)
185 def test_eq_invalid(self
):
186 notif
= bt2
.StreamBeginningNotification(self
._stream
)
187 self
.assertNotEqual(notif
, 23)
190 notif
= bt2
.StreamBeginningNotification(self
._stream
)
191 notif2
= copy
.copy(notif
)
192 self
.assertEqual(notif
, notif2
)
194 def test_deepcopy(self
):
195 notif
= bt2
.StreamBeginningNotification(self
._stream
)
196 notif2
= copy
.deepcopy(notif
)
197 self
.assertEqual(notif
, notif2
)
200 class StreamEndNotificationTestCase(_NotificationTestCase
):
201 def test_create(self
):
202 notif
= bt2
.StreamEndNotification(self
._stream
)
203 self
.assertEqual(notif
.stream
.addr
, self
._stream
.addr
)
206 notif
= bt2
.StreamEndNotification(self
._stream
)
207 stream_copy
= copy
.copy(self
._stream
)
208 notif2
= bt2
.StreamEndNotification(stream_copy
)
209 self
.assertEqual(notif
, notif2
)
211 def test_ne_stream(self
):
212 notif
= bt2
.StreamEndNotification(self
._stream
)
213 stream_copy
= self
._sc
(name
='salut')
214 notif2
= bt2
.StreamEndNotification(stream_copy
)
215 self
.assertNotEqual(notif
, notif2
)
217 def test_eq_invalid(self
):
218 notif
= bt2
.StreamEndNotification(self
._stream
)
219 self
.assertNotEqual(notif
, 23)
222 notif
= bt2
.StreamEndNotification(self
._stream
)
223 notif2
= copy
.copy(notif
)
224 self
.assertEqual(notif
, notif2
)
226 def test_deepcopy(self
):
227 notif
= bt2
.StreamEndNotification(self
._stream
)
228 notif2
= copy
.deepcopy(notif
)
229 self
.assertEqual(notif
, notif2
)
232 class InactivityNotificationTestCase(unittest
.TestCase
):
234 self
._cc
1 = bt2
.ClockClass('cc1', 1000)
235 self
._cc
2 = bt2
.ClockClass('cc2', 2000)
236 self
._cc
_prio
_map
= bt2
.ClockClassPriorityMap()
237 self
._cc
_prio
_map
[self
._cc
1] = 25
238 self
._cc
_prio
_map
[self
._cc
2] = 50
243 del self
._cc
_prio
_map
245 def test_create_no_cc_prio_map(self
):
246 notif
= bt2
.InactivityNotification()
247 self
.assertEqual(len(notif
.clock_class_priority_map
), 0)
249 def test_create_with_cc_prio_map(self
):
250 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
251 notif
.add_clock_value(self
._cc
1(123))
252 notif
.add_clock_value(self
._cc
2(19487))
253 self
.assertEqual(len(notif
.clock_class_priority_map
), 2)
254 self
.assertEqual(notif
.clock_class_priority_map
, self
._cc
_prio
_map
)
255 self
.assertEqual(notif
.clock_value(self
._cc
1), 123)
256 self
.assertEqual(notif
.clock_value(self
._cc
2), 19487)
259 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
260 notif
.add_clock_value(self
._cc
1(123))
261 notif
.add_clock_value(self
._cc
2(19487))
262 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
263 notif2
= bt2
.InactivityNotification(cc_prio_map_copy
)
264 notif2
.add_clock_value(self
._cc
1(123))
265 notif2
.add_clock_value(self
._cc
2(19487))
266 self
.assertEqual(notif
, notif2
)
268 def test_ne_cc_prio_map(self
):
269 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
270 notif
.add_clock_value(self
._cc
1(123))
271 notif
.add_clock_value(self
._cc
2(19487))
272 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
273 cc_prio_map_copy
[self
._cc
2] = 23
274 notif2
= bt2
.InactivityNotification(cc_prio_map_copy
)
275 self
.assertNotEqual(notif
, notif2
)
277 def test_ne_clock_value(self
):
278 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
279 notif
.add_clock_value(self
._cc
1(123))
280 notif
.add_clock_value(self
._cc
2(19487))
281 notif2
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
282 notif
.add_clock_value(self
._cc
1(123))
283 notif
.add_clock_value(self
._cc
2(1847))
284 self
.assertNotEqual(notif
, notif2
)
286 def test_eq_invalid(self
):
287 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
288 self
.assertNotEqual(notif
, 23)
291 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
292 notif
.add_clock_value(self
._cc
1(123))
293 notif
.add_clock_value(self
._cc
2(19487))
294 notif_copy
= copy
.copy(notif
)
295 self
.assertEqual(notif
, notif_copy
)
296 self
.assertNotEqual(notif
.addr
, notif_copy
.addr
)
297 self
.assertEqual(notif
.clock_class_priority_map
.addr
,
298 notif_copy
.clock_class_priority_map
.addr
)
299 self
.assertEqual(notif_copy
.clock_value(self
._cc
1), 123)
300 self
.assertEqual(notif_copy
.clock_value(self
._cc
2), 19487)
302 def test_deepcopy(self
):
303 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
304 notif
.add_clock_value(self
._cc
1(123))
305 notif
.add_clock_value(self
._cc
2(19487))
306 notif_copy
= copy
.deepcopy(notif
)
307 self
.assertEqual(notif
, notif_copy
)
308 self
.assertNotEqual(notif
.addr
, notif_copy
.addr
)
309 self
.assertNotEqual(notif
.clock_class_priority_map
.addr
,
310 notif_copy
.clock_class_priority_map
.addr
)
311 self
.assertEqual(notif
.clock_class_priority_map
,
312 notif_copy
.clock_class_priority_map
)
313 self
.assertNotEqual(list(notif
.clock_class_priority_map
)[0].addr
,
314 list(notif_copy
.clock_class_priority_map
)[0].addr
)
315 self
.assertIsNone(notif_copy
.clock_value(self
._cc
1))
316 self
.assertIsNone(notif_copy
.clock_value(self
._cc
2))
317 self
.assertEqual(notif_copy
.clock_value(list(notif_copy
.clock_class_priority_map
)[0]), 123)
318 self
.assertEqual(notif_copy
.clock_value(list(notif_copy
.clock_class_priority_map
)[1]), 19487)
321 class DiscardedPacketsNotificationTestCase(unittest
.TestCase
):
323 self
._trace
= bt2
.Trace()
324 self
._sc
= bt2
.StreamClass()
325 self
._ec
= bt2
.EventClass('salut')
326 self
._clock
_class
= bt2
.ClockClass('yo', 1000)
327 self
._uint
64_int
_ft
= bt2
.IntegerFieldType(64, mapped_clock_class
=self
._clock
_class
)
328 self
._my
_int
_ft
= bt2
.IntegerFieldType(32)
329 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
330 self
._ec
.payload_field_type
+= collections
.OrderedDict([
331 ('my_int', self
._my
_int
_ft
),
333 self
._sc
.add_event_class(self
._ec
)
334 self
._sc
.packet_context_field_type
= bt2
.StructureFieldType()
335 self
._sc
.packet_context_field_type
+= collections
.OrderedDict([
336 ('packet_seq_num', self
._my
_int
_ft
),
337 ('timestamp_begin', self
._uint
64_int
_ft
),
338 ('timestamp_end', self
._uint
64_int
_ft
),
340 self
._trace
.add_clock_class(self
._clock
_class
)
341 self
._trace
.add_stream_class(self
._sc
)
342 self
._stream
= self
._sc
()
348 del self
._clock
_class
349 del self
._uint
64_int
_ft
353 def _create_event(self
, packet
):
355 event
.payload_field
['my_int'] = 23
356 event
.packet
= packet
359 def _get_notif(self
):
360 class MyIter(bt2
._UserNotificationIterator
):
361 def __init__(iter_self
):
362 packet1
= self
._stream
.create_packet()
363 packet1
.context_field
['packet_seq_num'] = 0
364 packet1
.context_field
['timestamp_begin'] = 3
365 packet1
.context_field
['timestamp_end'] = 6
366 packet2
= self
._stream
.create_packet()
367 packet2
.context_field
['packet_seq_num'] = 5
368 packet2
.context_field
['timestamp_begin'] = 7
369 packet2
.context_field
['timestamp_end'] = 10
370 iter_self
._ev
1 = self
._create
_event
(packet1
)
371 iter_self
._ev
2 = self
._create
_event
(packet2
)
376 notif
= bt2
.EventNotification(self
._ev
1)
378 notif
= bt2
.EventNotification(self
._ev
2)
385 class MySource(bt2
._UserSourceComponent
,
386 notification_iterator_class
=MyIter
):
387 def __init__(self
, params
):
388 self
._add
_output
_port
('out')
390 class MySink(bt2
._UserSinkComponent
):
391 def __init__(self
, params
):
392 self
._add
_input
_port
('in')
394 def _consume(comp_self
):
396 notif
= next(comp_self
._notif
_iter
)
398 if type(notif
) is bt2
._DiscardedPacketsNotification
:
402 def _port_connected(self
, port
, other_port
):
403 self
._notif
_iter
= port
.connection
.create_notification_iterator()
407 src
= graph
.add_component(MySource
, 'src')
408 sink
= graph
.add_component(MySink
, 'sink')
409 conn
= graph
.connect_ports(src
.output_ports
['out'],
410 sink
.input_ports
['in'])
414 def test_create(self
):
415 self
.assertIsInstance(self
._get
_notif
(), bt2
._DiscardedPacketsNotification
)
417 def test_count(self
):
418 self
.assertEqual(self
._get
_notif
().count
, 4)
420 def test_stream(self
):
421 self
.assertEqual(self
._get
_notif
().stream
.addr
, self
._stream
.addr
)
423 def test_beginning_clock_value(self
):
424 notif
= self
._get
_notif
()
425 beginning_clock_value
= notif
.beginning_clock_value
426 self
.assertEqual(beginning_clock_value
.clock_class
, self
._clock
_class
)
427 self
.assertEqual(beginning_clock_value
, 6)
429 def test_end_clock_value(self
):
430 notif
= self
._get
_notif
()
431 end_clock_value
= notif
.end_clock_value
432 self
.assertEqual(end_clock_value
.clock_class
, self
._clock
_class
)
433 self
.assertEqual(end_clock_value
, 7)
436 notif1
= self
._get
_notif
()
437 notif2
= self
._get
_notif
()
438 self
.assertEqual(notif1
, notif2
)
440 def test_eq_invalid(self
):
441 notif1
= self
._get
_notif
()
442 self
.assertNotEqual(notif1
, 23)
445 class DiscardedEventsNotificationTestCase(unittest
.TestCase
):
447 self
._trace
= bt2
.Trace()
448 self
._sc
= bt2
.StreamClass()
449 self
._ec
= bt2
.EventClass('salut')
450 self
._clock
_class
= bt2
.ClockClass('yo', 1000)
451 self
._uint
64_int
_ft
= bt2
.IntegerFieldType(64, mapped_clock_class
=self
._clock
_class
)
452 self
._my
_int
_ft
= bt2
.IntegerFieldType(32)
453 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
454 self
._ec
.payload_field_type
+= collections
.OrderedDict([
455 ('my_int', self
._my
_int
_ft
),
457 self
._sc
.add_event_class(self
._ec
)
458 self
._sc
.packet_context_field_type
= bt2
.StructureFieldType()
459 self
._sc
.packet_context_field_type
+= collections
.OrderedDict([
460 ('events_discarded', self
._my
_int
_ft
),
461 ('timestamp_begin', self
._uint
64_int
_ft
),
462 ('timestamp_end', self
._uint
64_int
_ft
),
464 self
._trace
.add_clock_class(self
._clock
_class
)
465 self
._trace
.add_stream_class(self
._sc
)
466 self
._stream
= self
._sc
()
472 del self
._clock
_class
473 del self
._uint
64_int
_ft
477 def _create_event(self
, packet
):
479 event
.payload_field
['my_int'] = 23
480 event
.packet
= packet
483 def _get_notif(self
):
484 class MyIter(bt2
._UserNotificationIterator
):
485 def __init__(iter_self
):
486 packet1
= self
._stream
.create_packet()
487 packet1
.context_field
['events_discarded'] = 0
488 packet1
.context_field
['timestamp_begin'] = 3
489 packet1
.context_field
['timestamp_end'] = 6
490 packet2
= self
._stream
.create_packet()
491 packet2
.context_field
['events_discarded'] = 10
492 packet2
.context_field
['timestamp_begin'] = 7
493 packet2
.context_field
['timestamp_end'] = 10
494 iter_self
._ev
1 = self
._create
_event
(packet1
)
495 iter_self
._ev
2 = self
._create
_event
(packet2
)
500 notif
= bt2
.EventNotification(self
._ev
1)
502 notif
= bt2
.EventNotification(self
._ev
2)
509 class MySource(bt2
._UserSourceComponent
,
510 notification_iterator_class
=MyIter
):
511 def __init__(self
, params
):
512 self
._add
_output
_port
('out')
514 class MySink(bt2
._UserSinkComponent
):
515 def __init__(self
, params
):
516 self
._add
_input
_port
('in')
518 def _consume(comp_self
):
520 notif
= next(comp_self
._notif
_iter
)
522 if type(notif
) is bt2
._DiscardedEventsNotification
:
526 def _port_connected(self
, port
, other_port
):
527 self
._notif
_iter
= port
.connection
.create_notification_iterator()
531 src
= graph
.add_component(MySource
, 'src')
532 sink
= graph
.add_component(MySink
, 'sink')
533 conn
= graph
.connect_ports(src
.output_ports
['out'],
534 sink
.input_ports
['in'])
538 def test_create(self
):
539 self
.assertIsInstance(self
._get
_notif
(), bt2
._DiscardedEventsNotification
)
541 def test_count(self
):
542 self
.assertEqual(self
._get
_notif
().count
, 10)
544 def test_stream(self
):
545 self
.assertEqual(self
._get
_notif
().stream
.addr
, self
._stream
.addr
)
547 def test_beginning_clock_value(self
):
548 notif
= self
._get
_notif
()
549 beginning_clock_value
= notif
.beginning_clock_value
550 self
.assertEqual(beginning_clock_value
.clock_class
, self
._clock
_class
)
551 self
.assertEqual(beginning_clock_value
, 6)
553 def test_end_clock_value(self
):
554 notif
= self
._get
_notif
()
555 end_clock_value
= notif
.end_clock_value
556 self
.assertEqual(end_clock_value
.clock_class
, self
._clock
_class
)
557 self
.assertEqual(end_clock_value
, 10)
560 notif1
= self
._get
_notif
()
561 notif2
= self
._get
_notif
()
562 self
.assertEqual(notif1
, notif2
)
564 def test_eq_invalid(self
):
565 notif1
= self
._get
_notif
()
566 self
.assertNotEqual(notif1
, 23)