1 from collections
import OrderedDict
4 import bt2
.notification_iterator
10 cc
= bt2
.ClockClass('salut_clock')
13 eh
= bt2
.StructureFieldType()
15 ('id', bt2
.IntegerFieldType(8)),
16 ('ts', bt2
.IntegerFieldType(32, mapped_clock_class
=cc
)),
20 pc
= bt2
.StructureFieldType()
22 ('something', bt2
.IntegerFieldType(8)),
26 sc
= bt2
.StreamClass()
27 sc
.packet_context_field_type
= pc
28 sc
.event_header_field_type
= eh
29 sc
.event_context_field_type
= None
32 ep
= bt2
.StructureFieldType()
34 ('mosquito', bt2
.StringFieldType()),
38 event_class
= bt2
.EventClass('ec', id=0)
39 event_class
.context_field_type
= None
40 event_class
.payload_field_type
= ep
41 sc
.add_event_class(event_class
)
44 ph
= bt2
.StructureFieldType()
46 ('magic', bt2
.IntegerFieldType(32)),
47 ('stream_id', bt2
.IntegerFieldType(16)),
52 trace
.packet_header_field_type
= ph
53 trace
.add_clock_class(cc
)
54 trace
.add_stream_class(sc
)
58 def _create_event(event_class
, msg
):
60 ev
.header_field
['id'] = 0
61 ev
.header_field
['ts'] = 19487
62 ev
.payload_field
['mosquito'] = msg
66 def _create_packet(stream
):
67 packet
= stream
.create_packet()
68 packet
.header_field
['magic'] = 0xc1fc1fc1
69 packet
.header_field
['stream_id'] = 0
70 packet
.context_field
['something'] = 194
75 class MyIter(bt2
.UserNotificationIterator
):
77 self
._event
_class
= self
.component
._event
_class
78 self
._stream
= self
.component
._stream
79 self
._packet
= _create_packet(self
._stream
)
81 self
._cur
_notif
= None
84 if self
._cur
_notif
is None:
85 raise bt2
.Error('nothing here!')
87 return self
._cur
_notif
91 notif
= bt2
.BeginningOfPacketNotification(self
._packet
)
93 ev
= _create_event(self
._event
_class
, 'at {}'.format(self
._at
))
94 ev
.packet
= self
._packet
95 notif
= bt2
.TraceEventNotification(ev
)
97 notif
= bt2
.EndOfPacketNotification(self
._packet
)
99 notif
= bt2
.EndOfStreamNotification(self
._stream
)
104 self
._cur
_notif
= notif
106 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
108 self
._event
_class
= _create_ec()
109 self
._stream
= self
._event
_class
.stream_class(name
='abcdef')
114 class GenCompClassTestCase(unittest
.TestCase
):
115 def test_attr_name(self
):
116 class MySink(bt2
.UserSinkComponent
):
121 self
.assertEqual(sink
.component_class
.name
, 'MySink')
123 def test_attr_description(self
):
124 class MySink(bt2
.UserSinkComponent
):
131 self
.assertEqual(sink
.component_class
.description
, 'hello')
133 def test_instantiate(self
):
134 class MySink(bt2
.UserSinkComponent
):
137 def __init__(self
, params
=None, name
=None):
138 if params
is None or name
is None:
141 nonlocal nl_params_a
, nl_name
142 nl_params_a
= params
['a']
151 self
.assertIsNone(nl_params_a
)
152 self
.assertIsNone(nl_name
)
153 gen_comp_class
= sink
.component_class
154 sink2
= gen_comp_class(params
={'a': 23}, name
='salut')
155 self
.assertEqual(nl_params_a
, 23)
156 self
.assertEqual(nl_name
, 'salut')
159 class UserCompClassTestCase(unittest
.TestCase
):
160 def test_attr_name(self
):
161 class MySink(bt2
.UserSinkComponent
):
165 self
.assertEqual(MySink
.name
, 'MySink')
167 def test_attr_name_param(self
):
168 class MySink(bt2
.UserSinkComponent
, name
='my custom name'):
172 self
.assertEqual(MySink
.name
, 'my custom name')
174 def test_attr_description(self
):
175 class MySink(bt2
.UserSinkComponent
):
176 'here is your description'
181 self
.assertEqual(MySink
.description
, 'here is your description')
184 class MySink(bt2
.UserSinkComponent
):
194 self
.assertTrue(inited
)
196 def test_init_raises(self
):
197 class MySink(bt2
.UserSinkComponent
):
199 raise RuntimeError('oops')
204 with self
.assertRaises(RuntimeError):
207 def test_destroy(self
):
208 class MySink(bt2
.UserSinkComponent
):
219 self
.assertTrue(destroyed
)
221 def test_destroy_raises(self
):
222 class MySink(bt2
.UserSinkComponent
):
224 raise RuntimeError('oh oh')
234 self
.assertFalse(destroyed
)
236 def test_comp_attr_name(self
):
237 class MySink(bt2
.UserSinkComponent
):
241 sink
= MySink(name
='salut')
242 self
.assertEqual(sink
.name
, 'salut')
244 def test_comp_attr_no_name(self
):
245 class MySink(bt2
.UserSinkComponent
):
250 self
.assertIsNone(sink
.name
)
252 def test_comp_attr_class(self
):
253 class MySink(bt2
.UserSinkComponent
):
258 self
.assertEqual(sink
.component_class
.name
, 'MySink')
261 class UserSinkCompClassTestCase(unittest
.TestCase
):
262 def test_missing_consume(self
):
263 with self
.assertRaises(bt2
.IncompleteUserClassError
):
264 class MySink(bt2
.UserSinkComponent
):
267 def test_set_min_input(self
):
268 class MySink(bt2
.UserSinkComponent
):
270 self
._maximum
_input
_notification
_iterator
_count
= 10
271 self
._minimum
_input
_notification
_iterator
_count
= 5
278 def test_set_max_input(self
):
279 class MySink(bt2
.UserSinkComponent
):
281 self
._maximum
_input
_notification
_iterator
_count
= 5
288 def test_consume(self
):
289 class MySink(bt2
.UserSinkComponent
):
299 source
= _create_source()
300 notif_iter
= source
.create_notification_iterator()
301 sink
.add_notification_iterator(notif_iter
)
303 self
.assertTrue(consumed
)
305 def test_consume_raises(self
):
306 class MySink(bt2
.UserSinkComponent
):
311 raise RuntimeError('hello')
314 source
= _create_source()
315 notif_iter
= source
.create_notification_iterator()
316 sink
.add_notification_iterator(notif_iter
)
318 with self
.assertRaises(bt2
.Error
):
321 def test_add_notification_iterator(self
):
322 class MySink(bt2
.UserSinkComponent
):
329 def _add_notification_iterator(self
, notif_iter
):
335 source
= _create_source()
336 notif_iter
= source
.create_notification_iterator()
337 sink
.add_notification_iterator(notif_iter
)
338 self
.assertTrue(added
)
340 def test_input_notif_iterators(self
):
341 class MySink(bt2
.UserSinkComponent
):
343 self
._maximum
_input
_notification
_iterator
_count
= 5
349 for notif_iter
in self
._input
_notification
_iterators
:
354 source
= _create_source()
355 notif_iter1
= source
.create_notification_iterator()
356 notif_iter2
= source
.create_notification_iterator()
357 sink
.add_notification_iterator(notif_iter1
)
358 sink
.add_notification_iterator(notif_iter2
)
360 self
.assertEqual(count
, 2)
363 class UserSourceCompClassTestCase(unittest
.TestCase
):
364 def test_missing_notif_iterator_class(self
):
365 with self
.assertRaises(bt2
.IncompleteUserClassError
):
366 class MySource(bt2
.UserSourceComponent
):
369 def test_invalid_notif_iterator_class(self
):
373 with self
.assertRaises(bt2
.IncompleteUserClassError
):
374 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=Lel
):
377 def test_notif_iterator_class_missing_get(self
):
378 class MyIter(bt2
.UserNotificationIterator
):
382 with self
.assertRaises(bt2
.IncompleteUserClassError
):
383 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
386 def test_notif_iterator_class_missing_next(self
):
387 class MyIter(bt2
.UserNotificationIterator
):
391 with self
.assertRaises(bt2
.IncompleteUserClassError
):
392 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
395 def test_create_notification_iterator(self
):
396 class MyIter(bt2
.UserNotificationIterator
):
407 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
412 notif_iter
= source
.create_notification_iterator()
413 self
.assertTrue(created
)
416 class UserSourceCompClassTestCase(unittest
.TestCase
):
417 def test_missing_notif_iterator_class(self
):
418 with self
.assertRaises(bt2
.IncompleteUserClassError
):
419 class MyFilter(bt2
.UserFilterComponent
):
422 def test_invalid_notif_iterator_class(self
):
426 with self
.assertRaises(bt2
.IncompleteUserClassError
):
427 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=Lel
):
430 def test_notif_iterator_class_missing_get(self
):
431 class MyIter(bt2
.UserNotificationIterator
):
435 with self
.assertRaises(bt2
.IncompleteUserClassError
):
436 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
439 def test_notif_iterator_class_missing_next(self
):
440 class MyIter(bt2
.UserNotificationIterator
):
444 with self
.assertRaises(bt2
.IncompleteUserClassError
):
445 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
448 def test_create_notification_iterator(self
):
449 class MyIter(bt2
.UserNotificationIterator
):
460 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
465 notif_iter
= filter.create_notification_iterator()
466 self
.assertTrue(created
)
468 def test_set_min_input(self
):
469 class MyIter(bt2
.UserNotificationIterator
):
476 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
478 self
._maximum
_input
_notification
_iterator
_count
= 10
479 self
._minimum
_input
_notification
_iterator
_count
= 5
483 def test_set_max_input(self
):
484 class MyIter(bt2
.UserNotificationIterator
):
491 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
493 self
._maximum
_input
_notification
_iterator
_count
= 5
497 def test_add_notification_iterator(self
):
498 class MyIter(bt2
.UserNotificationIterator
):
505 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
509 def _add_notification_iterator(self
, notif_iter
):
515 source
= _create_source()
516 notif_iter
= source
.create_notification_iterator()
517 filter.add_notification_iterator(notif_iter
)
518 self
.assertTrue(added
)
520 def test_input_notif_iterators(self
):
521 class MyIter(bt2
.UserNotificationIterator
):
528 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
530 self
._maximum
_input
_notification
_iterator
_count
= 5
536 for notif_iter
in self
._input
_notification
_iterators
:
541 source
= _create_source()
542 notif_iter1
= source
.create_notification_iterator()
543 notif_iter2
= source
.create_notification_iterator()
544 filter.add_notification_iterator(notif_iter1
)
545 filter.add_notification_iterator(notif_iter2
)
547 self
.assertEqual(count
, 2)
550 class UserNotificationIteratorTestCase(unittest
.TestCase
):
552 class MyIter(bt2
.UserNotificationIterator
):
563 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
568 notif_iter
= source
.create_notification_iterator()
569 self
.assertTrue(inited
)
571 def test_destroy(self
):
572 class MyIter(bt2
.UserNotificationIterator
):
583 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
587 notif_iter
= source
.create_notification_iterator()
590 self
.assertTrue(destroyed
)
592 def test_attr_component(self
):
593 class MyIter(bt2
.UserNotificationIterator
):
595 nonlocal source
, is_same
596 is_same
= self
.component
is source
604 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
609 notif_iter
= source
.create_notification_iterator()
610 self
.assertTrue(is_same
)
612 def test_next_raises_stop(self
):
613 class MyIter(bt2
.UserNotificationIterator
):
620 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
625 notif_iter
= source
.create_notification_iterator()
627 with self
.assertRaises(bt2
.Stop
):
630 def test_next_raises_unsupported_feature(self
):
631 class MyIter(bt2
.UserNotificationIterator
):
633 raise bt2
.UnsupportedFeature
638 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
643 notif_iter
= source
.create_notification_iterator()
645 with self
.assertRaises(bt2
.UnsupportedFeature
):
648 def test_next_raises_unknown(self
):
649 class MyIter(bt2
.UserNotificationIterator
):
651 raise TypeError('lel')
656 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
661 notif_iter
= source
.create_notification_iterator()
663 with self
.assertRaises(bt2
.Error
):
666 def test_iteration(self
):
667 source
= _create_source()
668 notif_iter
= source
.create_notification_iterator()
671 for index
, notif
in enumerate(notif_iter
):
673 self
.assertIsInstance(notif
, bt2
.BeginningOfPacketNotification
)
674 self
.assertEqual(notif
.packet
.header_field
['magic'], 0xc1fc1fc1)
675 self
.assertEqual(notif
.packet
.context_field
['something'], 194)
676 pkt_addr
= notif
.packet
.addr
678 self
.assertIsInstance(notif
, bt2
.TraceEventNotification
)
679 self
.assertEqual(notif
.event
.header_field
['ts'], 19487)
680 self
.assertEqual(notif
.event
.payload_field
['mosquito'], 'at {}'.format(index
))
682 self
.assertIsInstance(notif
, bt2
.EndOfPacketNotification
)
683 self
.assertEqual(notif
.packet
.header_field
['magic'], 0xc1fc1fc1)
684 self
.assertEqual(notif
.packet
.context_field
['something'], 194)
685 self
.assertEqual(notif
.packet
.addr
, pkt_addr
)
687 self
.assertIsInstance(notif
, bt2
.EndOfStreamNotification
)
688 self
.assertEqual(notif
.stream
.name
, 'abcdef')
690 raise RuntimeError('FAIL')
693 class GenNotificationIteratorTestCase(unittest
.TestCase
):
694 def test_attr_component(self
):
695 source
= _create_source()
696 notif_iter
= source
.create_notification_iterator()
697 self
.assertIsInstance(notif_iter
, bt2
.notification_iterator
._GenericNotificationIterator
)
698 self
.assertEqual(notif_iter
.component
.addr
, source
.addr
)