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
.clock_values
.add(self
._clock
_class
(1772))
33 self
._event
.payload_field
['my_int'] = 23
34 self
._event
.packet
= self
._packet
48 @unittest.skip("this is broken")
49 class EventNotificationTestCase(_NotificationTestCase
):
50 def test_create_no_cc_prio_map(self
):
51 notif
= bt2
.EventNotification(self
._event
)
52 self
.assertEqual(notif
.event
.addr
, self
._event
.addr
)
53 self
.assertEqual(len(notif
.clock_class_priority_map
), 0)
55 def test_create_with_cc_prio_map(self
):
56 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
57 self
.assertEqual(notif
.event
.addr
, self
._event
.addr
)
58 self
.assertEqual(len(notif
.clock_class_priority_map
), 1)
59 self
.assertEqual(notif
.clock_class_priority_map
.highest_priority_clock_class
.addr
,
60 self
._clock
_class
.addr
)
61 self
.assertEqual(notif
.clock_class_priority_map
[self
._clock
_class
], 231)
64 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
65 event_copy
= copy
.copy(self
._event
)
66 event_copy
.packet
= self
._packet
67 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
68 notif2
= bt2
.EventNotification(event_copy
, cc_prio_map_copy
)
69 self
.assertEqual(notif
, notif2
)
71 def test_ne_event(self
):
72 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
73 event_copy
= copy
.copy(self
._event
)
74 event_copy
.payload_field
['my_int'] = 17
75 event_copy
.packet
= self
._packet
76 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
77 notif2
= bt2
.EventNotification(event_copy
, cc_prio_map_copy
)
78 self
.assertNotEqual(notif
, notif2
)
80 def test_ne_cc_prio_map(self
):
81 notif
= bt2
.EventNotification(self
._event
)
82 event_copy
= copy
.copy(self
._event
)
83 event_copy
.packet
= self
._packet
84 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
85 notif2
= bt2
.EventNotification(event_copy
, cc_prio_map_copy
)
86 self
.assertNotEqual(notif
, notif2
)
88 def test_eq_invalid(self
):
89 notif
= bt2
.EventNotification(self
._event
)
90 self
.assertNotEqual(notif
, 23)
93 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
94 notif2
= copy
.copy(notif
)
95 self
.assertEqual(notif
, notif2
)
97 def test_deepcopy(self
):
98 notif
= bt2
.EventNotification(self
._event
, self
._cc
_prio
_map
)
99 notif2
= copy
.deepcopy(notif
)
100 self
.assertEqual(notif
, notif2
)
103 @unittest.skip("this is broken")
104 class PacketBeginningNotificationTestCase(_NotificationTestCase
):
105 def test_create(self
):
106 notif
= bt2
.PacketBeginningNotification(self
._packet
)
107 self
.assertEqual(notif
.packet
.addr
, self
._packet
.addr
)
110 notif
= bt2
.PacketBeginningNotification(self
._packet
)
111 packet_copy
= copy
.copy(self
._packet
)
112 notif2
= bt2
.PacketBeginningNotification(packet_copy
)
113 self
.assertEqual(notif
, notif2
)
115 def test_ne_packet(self
):
116 notif
= bt2
.PacketBeginningNotification(self
._packet
)
117 packet_copy
= copy
.copy(self
._packet
)
118 packet_copy
.header_field
['hello'] = 1847
119 notif2
= bt2
.PacketBeginningNotification(packet_copy
)
120 self
.assertNotEqual(notif
, notif2
)
122 def test_eq_invalid(self
):
123 notif
= bt2
.PacketBeginningNotification(self
._packet
)
124 self
.assertNotEqual(notif
, 23)
127 notif
= bt2
.PacketBeginningNotification(self
._packet
)
128 notif2
= copy
.copy(notif
)
129 self
.assertEqual(notif
, notif2
)
131 def test_deepcopy(self
):
132 notif
= bt2
.PacketBeginningNotification(self
._packet
)
133 notif2
= copy
.deepcopy(notif
)
134 self
.assertEqual(notif
, notif2
)
137 @unittest.skip("this is broken")
138 class PacketEndNotificationTestCase(_NotificationTestCase
):
139 def test_create(self
):
140 notif
= bt2
.PacketEndNotification(self
._packet
)
141 self
.assertEqual(notif
.packet
.addr
, self
._packet
.addr
)
144 notif
= bt2
.PacketEndNotification(self
._packet
)
145 packet_copy
= copy
.copy(self
._packet
)
146 notif2
= bt2
.PacketEndNotification(packet_copy
)
147 self
.assertEqual(notif
, notif2
)
149 def test_ne_packet(self
):
150 notif
= bt2
.PacketEndNotification(self
._packet
)
151 packet_copy
= copy
.copy(self
._packet
)
152 packet_copy
.header_field
['hello'] = 1847
153 notif2
= bt2
.PacketEndNotification(packet_copy
)
154 self
.assertNotEqual(notif
, notif2
)
156 def test_eq_invalid(self
):
157 notif
= bt2
.PacketEndNotification(self
._packet
)
158 self
.assertNotEqual(notif
, 23)
161 notif
= bt2
.PacketEndNotification(self
._packet
)
162 notif2
= copy
.copy(notif
)
163 self
.assertEqual(notif
, notif2
)
165 def test_deepcopy(self
):
166 notif
= bt2
.PacketEndNotification(self
._packet
)
167 notif2
= copy
.deepcopy(notif
)
168 self
.assertEqual(notif
, notif2
)
171 @unittest.skip("this is broken")
172 class StreamBeginningNotificationTestCase(_NotificationTestCase
):
173 def test_create(self
):
174 notif
= bt2
.StreamBeginningNotification(self
._stream
)
175 self
.assertEqual(notif
.stream
.addr
, self
._stream
.addr
)
178 notif
= bt2
.StreamBeginningNotification(self
._stream
)
179 stream_copy
= copy
.copy(self
._stream
)
180 notif2
= bt2
.StreamBeginningNotification(stream_copy
)
181 self
.assertEqual(notif
, notif2
)
183 def test_ne_stream(self
):
184 notif
= bt2
.StreamBeginningNotification(self
._stream
)
185 stream_copy
= self
._sc
(name
='salut')
186 notif2
= bt2
.StreamBeginningNotification(stream_copy
)
187 self
.assertNotEqual(notif
, notif2
)
189 def test_eq_invalid(self
):
190 notif
= bt2
.StreamBeginningNotification(self
._stream
)
191 self
.assertNotEqual(notif
, 23)
194 notif
= bt2
.StreamBeginningNotification(self
._stream
)
195 notif2
= copy
.copy(notif
)
196 self
.assertEqual(notif
, notif2
)
198 def test_deepcopy(self
):
199 notif
= bt2
.StreamBeginningNotification(self
._stream
)
200 notif2
= copy
.deepcopy(notif
)
201 self
.assertEqual(notif
, notif2
)
204 @unittest.skip("this is broken")
205 class StreamEndNotificationTestCase(_NotificationTestCase
):
206 def test_create(self
):
207 notif
= bt2
.StreamEndNotification(self
._stream
)
208 self
.assertEqual(notif
.stream
.addr
, self
._stream
.addr
)
211 notif
= bt2
.StreamEndNotification(self
._stream
)
212 stream_copy
= copy
.copy(self
._stream
)
213 notif2
= bt2
.StreamEndNotification(stream_copy
)
214 self
.assertEqual(notif
, notif2
)
216 def test_ne_stream(self
):
217 notif
= bt2
.StreamEndNotification(self
._stream
)
218 stream_copy
= self
._sc
(name
='salut')
219 notif2
= bt2
.StreamEndNotification(stream_copy
)
220 self
.assertNotEqual(notif
, notif2
)
222 def test_eq_invalid(self
):
223 notif
= bt2
.StreamEndNotification(self
._stream
)
224 self
.assertNotEqual(notif
, 23)
227 notif
= bt2
.StreamEndNotification(self
._stream
)
228 notif2
= copy
.copy(notif
)
229 self
.assertEqual(notif
, notif2
)
231 def test_deepcopy(self
):
232 notif
= bt2
.StreamEndNotification(self
._stream
)
233 notif2
= copy
.deepcopy(notif
)
234 self
.assertEqual(notif
, notif2
)
237 @unittest.skip("this is broken")
238 class InactivityNotificationTestCase(unittest
.TestCase
):
240 self
._cc
1 = bt2
.ClockClass('cc1', 1000)
241 self
._cc
2 = bt2
.ClockClass('cc2', 2000)
242 self
._cc
_prio
_map
= bt2
.ClockClassPriorityMap()
243 self
._cc
_prio
_map
[self
._cc
1] = 25
244 self
._cc
_prio
_map
[self
._cc
2] = 50
249 del self
._cc
_prio
_map
251 def test_create_no_cc_prio_map(self
):
252 notif
= bt2
.InactivityNotification()
253 self
.assertEqual(len(notif
.clock_class_priority_map
), 0)
255 def test_create_with_cc_prio_map(self
):
256 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
257 notif
.clock_values
.add(self
._cc
1(123))
258 notif
.clock_values
.add(self
._cc
2(19487))
259 self
.assertEqual(len(notif
.clock_class_priority_map
), 2)
260 self
.assertEqual(notif
.clock_class_priority_map
, self
._cc
_prio
_map
)
261 self
.assertEqual(notif
.clock_values
[self
._cc
1], 123)
262 self
.assertEqual(notif
.clock_values
[self
._cc
2], 19487)
265 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
266 notif
.clock_values
.add(self
._cc
1(123))
267 notif
.clock_values
.add(self
._cc
2(19487))
268 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
269 notif2
= bt2
.InactivityNotification(cc_prio_map_copy
)
270 notif2
.clock_values
.add(self
._cc
1(123))
271 notif2
.clock_values
.add(self
._cc
2(19487))
272 self
.assertEqual(notif
, notif2
)
274 def test_ne_cc_prio_map(self
):
275 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
276 notif
.clock_values
.add(self
._cc
1(123))
277 notif
.clock_values
.add(self
._cc
2(19487))
278 cc_prio_map_copy
= copy
.copy(self
._cc
_prio
_map
)
279 cc_prio_map_copy
[self
._cc
2] = 23
280 notif2
= bt2
.InactivityNotification(cc_prio_map_copy
)
281 self
.assertNotEqual(notif
, notif2
)
283 def test_ne_clock_value(self
):
284 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
285 notif
.clock_values
.add(self
._cc
1(123))
286 notif
.clock_values
.add(self
._cc
2(19487))
287 notif2
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
288 notif
.clock_values
.add(self
._cc
1(123))
289 notif
.clock_values
.add(self
._cc
2(1847))
290 self
.assertNotEqual(notif
, notif2
)
292 def test_eq_invalid(self
):
293 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
294 self
.assertNotEqual(notif
, 23)
297 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
298 notif
.clock_values
.add(self
._cc
1(123))
299 notif
.clock_values
.add(self
._cc
2(19487))
300 notif_copy
= copy
.copy(notif
)
301 self
.assertEqual(notif
, notif_copy
)
302 self
.assertNotEqual(notif
.addr
, notif_copy
.addr
)
303 self
.assertEqual(notif
.clock_class_priority_map
.addr
,
304 notif_copy
.clock_class_priority_map
.addr
)
305 self
.assertEqual(notif_copy
.clock_values
[self
._cc
1], 123)
306 self
.assertEqual(notif_copy
.clock_values
[self
._cc
2], 19487)
308 def test_deepcopy(self
):
309 notif
= bt2
.InactivityNotification(self
._cc
_prio
_map
)
310 notif
.clock_values
.add(self
._cc
1(123))
311 notif
.clock_values
.add(self
._cc
2(19487))
312 notif_copy
= copy
.deepcopy(notif
)
313 self
.assertEqual(notif
, notif_copy
)
314 self
.assertNotEqual(notif
.addr
, notif_copy
.addr
)
315 self
.assertNotEqual(notif
.clock_class_priority_map
.addr
,
316 notif_copy
.clock_class_priority_map
.addr
)
317 self
.assertEqual(notif
.clock_class_priority_map
,
318 notif_copy
.clock_class_priority_map
)
319 self
.assertNotEqual(list(notif
.clock_class_priority_map
)[0].addr
,
320 list(notif_copy
.clock_class_priority_map
)[0].addr
)
321 self
.assertIsNone(notif_copy
.clock_values
[self
._cc
1])
322 self
.assertIsNone(notif_copy
.clock_values
[self
._cc
2])
323 self
.assertEqual(notif_copy
.clock_values
[list(notif_copy
.clock_class_priority_map
)[0]], 123)
324 self
.assertEqual(notif_copy
.clock_values
[list(notif_copy
.clock_class_priority_map
)[1]], 19487)
327 @unittest.skip("this is broken")
328 class DiscardedPacketsNotificationTestCase(unittest
.TestCase
):
330 self
._trace
= bt2
.Trace()
331 self
._sc
= bt2
.StreamClass()
332 self
._ec
= bt2
.EventClass('salut')
333 self
._clock
_class
= bt2
.ClockClass('yo', 1000)
334 self
._uint
64_int
_ft
= bt2
.IntegerFieldType(64, mapped_clock_class
=self
._clock
_class
)
335 self
._my
_int
_ft
= bt2
.IntegerFieldType(32)
336 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
337 self
._ec
.payload_field_type
+= collections
.OrderedDict([
338 ('my_int', self
._my
_int
_ft
),
340 self
._sc
.add_event_class(self
._ec
)
341 self
._sc
.packet_context_field_type
= bt2
.StructureFieldType()
342 self
._sc
.packet_context_field_type
+= collections
.OrderedDict([
343 ('packet_seq_num', self
._my
_int
_ft
),
344 ('timestamp_begin', self
._uint
64_int
_ft
),
345 ('timestamp_end', self
._uint
64_int
_ft
),
347 self
._trace
.add_clock_class(self
._clock
_class
)
348 self
._trace
.add_stream_class(self
._sc
)
349 self
._stream
= self
._sc
()
355 del self
._clock
_class
356 del self
._uint
64_int
_ft
360 def _create_event(self
, packet
):
362 event
.payload_field
['my_int'] = 23
363 event
.packet
= packet
366 def _get_notif(self
):
367 class MyIter(bt2
._UserNotificationIterator
):
368 def __init__(iter_self
):
369 packet1
= self
._stream
.create_packet()
370 packet1
.context_field
['packet_seq_num'] = 0
371 packet1
.context_field
['timestamp_begin'] = 3
372 packet1
.context_field
['timestamp_end'] = 6
373 packet2
= self
._stream
.create_packet()
374 packet2
.context_field
['packet_seq_num'] = 5
375 packet2
.context_field
['timestamp_begin'] = 7
376 packet2
.context_field
['timestamp_end'] = 10
377 iter_self
._ev
1 = self
._create
_event
(packet1
)
378 iter_self
._ev
2 = self
._create
_event
(packet2
)
383 notif
= bt2
.EventNotification(self
._ev
1)
385 notif
= bt2
.EventNotification(self
._ev
2)
392 class MySource(bt2
._UserSourceComponent
,
393 notification_iterator_class
=MyIter
):
394 def __init__(self
, params
):
395 self
._add
_output
_port
('out')
397 class MySink(bt2
._UserSinkComponent
):
398 def __init__(self
, params
):
399 self
._add
_input
_port
('in')
401 def _consume(comp_self
):
403 notif
= next(comp_self
._notif
_iter
)
405 if type(notif
) is bt2
._DiscardedPacketsNotification
:
409 def _port_connected(self
, port
, other_port
):
410 self
._notif
_iter
= port
.connection
.create_notification_iterator()
414 src
= graph
.add_component(MySource
, 'src')
415 sink
= graph
.add_component(MySink
, 'sink')
416 conn
= graph
.connect_ports(src
.output_ports
['out'],
417 sink
.input_ports
['in'])
421 def test_create(self
):
422 self
.assertIsInstance(self
._get
_notif
(), bt2
._DiscardedPacketsNotification
)
424 def test_count(self
):
425 self
.assertEqual(self
._get
_notif
().count
, 4)
427 def test_stream(self
):
428 self
.assertEqual(self
._get
_notif
().stream
.addr
, self
._stream
.addr
)
430 def test_beginning_clock_value(self
):
431 notif
= self
._get
_notif
()
432 beginning_clock_value
= notif
.beginning_clock_value
433 self
.assertEqual(beginning_clock_value
.clock_class
, self
._clock
_class
)
434 self
.assertEqual(beginning_clock_value
, 6)
436 def test_end_clock_value(self
):
437 notif
= self
._get
_notif
()
438 end_clock_value
= notif
.end_clock_value
439 self
.assertEqual(end_clock_value
.clock_class
, self
._clock
_class
)
440 self
.assertEqual(end_clock_value
, 7)
443 notif1
= self
._get
_notif
()
444 notif2
= self
._get
_notif
()
445 self
.assertEqual(notif1
, notif2
)
447 def test_eq_invalid(self
):
448 notif1
= self
._get
_notif
()
449 self
.assertNotEqual(notif1
, 23)
452 @unittest.skip("this is broken")
453 class DiscardedEventsNotificationTestCase(unittest
.TestCase
):
455 self
._trace
= bt2
.Trace()
456 self
._sc
= bt2
.StreamClass()
457 self
._ec
= bt2
.EventClass('salut')
458 self
._clock
_class
= bt2
.ClockClass('yo', 1000)
459 self
._uint
64_int
_ft
= bt2
.IntegerFieldType(64, mapped_clock_class
=self
._clock
_class
)
460 self
._my
_int
_ft
= bt2
.IntegerFieldType(32)
461 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
462 self
._ec
.payload_field_type
+= collections
.OrderedDict([
463 ('my_int', self
._my
_int
_ft
),
465 self
._sc
.add_event_class(self
._ec
)
466 self
._sc
.packet_context_field_type
= bt2
.StructureFieldType()
467 self
._sc
.packet_context_field_type
+= collections
.OrderedDict([
468 ('events_discarded', self
._my
_int
_ft
),
469 ('timestamp_begin', self
._uint
64_int
_ft
),
470 ('timestamp_end', self
._uint
64_int
_ft
),
472 self
._trace
.add_clock_class(self
._clock
_class
)
473 self
._trace
.add_stream_class(self
._sc
)
474 self
._stream
= self
._sc
()
480 del self
._clock
_class
481 del self
._uint
64_int
_ft
485 def _create_event(self
, packet
):
487 event
.payload_field
['my_int'] = 23
488 event
.packet
= packet
491 def _get_notif(self
):
492 class MyIter(bt2
._UserNotificationIterator
):
493 def __init__(iter_self
):
494 packet1
= self
._stream
.create_packet()
495 packet1
.context_field
['events_discarded'] = 0
496 packet1
.context_field
['timestamp_begin'] = 3
497 packet1
.context_field
['timestamp_end'] = 6
498 packet2
= self
._stream
.create_packet()
499 packet2
.context_field
['events_discarded'] = 10
500 packet2
.context_field
['timestamp_begin'] = 7
501 packet2
.context_field
['timestamp_end'] = 10
502 iter_self
._ev
1 = self
._create
_event
(packet1
)
503 iter_self
._ev
2 = self
._create
_event
(packet2
)
508 notif
= bt2
.EventNotification(self
._ev
1)
510 notif
= bt2
.EventNotification(self
._ev
2)
517 class MySource(bt2
._UserSourceComponent
,
518 notification_iterator_class
=MyIter
):
519 def __init__(self
, params
):
520 self
._add
_output
_port
('out')
522 class MySink(bt2
._UserSinkComponent
):
523 def __init__(self
, params
):
524 self
._add
_input
_port
('in')
526 def _consume(comp_self
):
528 notif
= next(comp_self
._notif
_iter
)
530 if type(notif
) is bt2
._DiscardedEventsNotification
:
534 def _port_connected(self
, port
, other_port
):
535 self
._notif
_iter
= port
.connection
.create_notification_iterator()
539 src
= graph
.add_component(MySource
, 'src')
540 sink
= graph
.add_component(MySink
, 'sink')
541 conn
= graph
.connect_ports(src
.output_ports
['out'],
542 sink
.input_ports
['in'])
546 def test_create(self
):
547 self
.assertIsInstance(self
._get
_notif
(), bt2
._DiscardedEventsNotification
)
549 def test_count(self
):
550 self
.assertEqual(self
._get
_notif
().count
, 10)
552 def test_stream(self
):
553 self
.assertEqual(self
._get
_notif
().stream
.addr
, self
._stream
.addr
)
555 def test_beginning_clock_value(self
):
556 notif
= self
._get
_notif
()
557 beginning_clock_value
= notif
.beginning_clock_value
558 self
.assertEqual(beginning_clock_value
.clock_class
, self
._clock
_class
)
559 self
.assertEqual(beginning_clock_value
, 6)
561 def test_end_clock_value(self
):
562 notif
= self
._get
_notif
()
563 end_clock_value
= notif
.end_clock_value
564 self
.assertEqual(end_clock_value
.clock_class
, self
._clock
_class
)
565 self
.assertEqual(end_clock_value
, 10)
568 notif1
= self
._get
_notif
()
569 notif2
= self
._get
_notif
()
570 self
.assertEqual(notif1
, notif2
)
572 def test_eq_invalid(self
):
573 notif1
= self
._get
_notif
()
574 self
.assertNotEqual(notif1
, 23)