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
):
125 '''This is the description.
141 self
.assertEqual(sink
.component_class
.description
,
142 'This is the description.')
144 def test_attr_help(self
):
145 class MySink(bt2
.UserSinkComponent
):
146 '''This is the description.
162 expected_help
= '''This is the help.
170 self
.assertEqual(sink
.component_class
.help, expected_help
)
172 def test_instantiate(self
):
173 class MySink(bt2
.UserSinkComponent
):
176 def __init__(self
, params
=None, name
=None):
177 if params
is None or name
is None:
180 nonlocal nl_params_a
, nl_name
181 nl_params_a
= params
['a']
190 self
.assertIsNone(nl_params_a
)
191 self
.assertIsNone(nl_name
)
192 gen_comp_class
= sink
.component_class
193 sink2
= gen_comp_class(params
={'a': 23}, name
='salut')
194 self
.assertEqual(nl_params_a
, 23)
195 self
.assertEqual(nl_name
, 'salut')
198 class UserCompClassTestCase(unittest
.TestCase
):
199 def test_attr_name(self
):
200 class MySink(bt2
.UserSinkComponent
):
204 self
.assertEqual(MySink
.name
, 'MySink')
206 def test_attr_name_param(self
):
207 class MySink(bt2
.UserSinkComponent
, name
='my custom name'):
211 self
.assertEqual(MySink
.name
, 'my custom name')
213 def test_attr_description(self
):
214 class MySink(bt2
.UserSinkComponent
):
215 '''This is the description.
230 self
.assertEqual(MySink
.description
, 'This is the description.')
232 def test_attr_help(self
):
233 class MySink(bt2
.UserSinkComponent
):
234 '''This is the description.
249 expected_help
= '''This is the help.
257 self
.assertEqual(MySink
.help, expected_help
)
260 class MySink(bt2
.UserSinkComponent
):
270 self
.assertTrue(inited
)
272 def test_init_raises(self
):
273 class MySink(bt2
.UserSinkComponent
):
275 raise RuntimeError('oops')
280 with self
.assertRaises(RuntimeError):
283 def test_destroy(self
):
284 class MySink(bt2
.UserSinkComponent
):
295 self
.assertTrue(destroyed
)
297 def test_destroy_raises(self
):
298 class MySink(bt2
.UserSinkComponent
):
300 raise RuntimeError('oh oh')
310 self
.assertFalse(destroyed
)
312 def test_comp_attr_name(self
):
313 class MySink(bt2
.UserSinkComponent
):
317 sink
= MySink(name
='salut')
318 self
.assertEqual(sink
.name
, 'salut')
320 def test_comp_attr_no_name(self
):
321 class MySink(bt2
.UserSinkComponent
):
326 self
.assertIsNone(sink
.name
)
328 def test_comp_attr_class(self
):
329 class MySink(bt2
.UserSinkComponent
):
334 self
.assertEqual(sink
.component_class
.name
, 'MySink')
337 class UserSinkCompClassTestCase(unittest
.TestCase
):
338 def test_missing_consume(self
):
339 with self
.assertRaises(bt2
.IncompleteUserClassError
):
340 class MySink(bt2
.UserSinkComponent
):
343 def test_set_min_input(self
):
344 class MySink(bt2
.UserSinkComponent
):
346 self
._maximum
_input
_notification
_iterator
_count
= 10
347 self
._minimum
_input
_notification
_iterator
_count
= 5
354 def test_set_max_input(self
):
355 class MySink(bt2
.UserSinkComponent
):
357 self
._maximum
_input
_notification
_iterator
_count
= 5
364 def test_consume(self
):
365 class MySink(bt2
.UserSinkComponent
):
375 source
= _create_source()
376 notif_iter
= source
.create_notification_iterator()
377 sink
.add_notification_iterator(notif_iter
)
379 self
.assertTrue(consumed
)
381 def test_consume_raises(self
):
382 class MySink(bt2
.UserSinkComponent
):
387 raise RuntimeError('hello')
390 source
= _create_source()
391 notif_iter
= source
.create_notification_iterator()
392 sink
.add_notification_iterator(notif_iter
)
394 with self
.assertRaises(bt2
.Error
):
397 def test_add_notification_iterator(self
):
398 class MySink(bt2
.UserSinkComponent
):
405 def _add_notification_iterator(self
, notif_iter
):
411 source
= _create_source()
412 notif_iter
= source
.create_notification_iterator()
413 sink
.add_notification_iterator(notif_iter
)
414 self
.assertTrue(added
)
416 def test_input_notif_iterators(self
):
417 class MySink(bt2
.UserSinkComponent
):
419 self
._maximum
_input
_notification
_iterator
_count
= 5
425 for notif_iter
in self
._input
_notification
_iterators
:
430 source
= _create_source()
431 notif_iter1
= source
.create_notification_iterator()
432 notif_iter2
= source
.create_notification_iterator()
433 sink
.add_notification_iterator(notif_iter1
)
434 sink
.add_notification_iterator(notif_iter2
)
436 self
.assertEqual(count
, 2)
439 class UserSourceCompClassTestCase(unittest
.TestCase
):
440 def test_missing_notif_iterator_class(self
):
441 with self
.assertRaises(bt2
.IncompleteUserClassError
):
442 class MySource(bt2
.UserSourceComponent
):
445 def test_invalid_notif_iterator_class(self
):
449 with self
.assertRaises(bt2
.IncompleteUserClassError
):
450 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=Lel
):
453 def test_notif_iterator_class_missing_get(self
):
454 class MyIter(bt2
.UserNotificationIterator
):
458 with self
.assertRaises(bt2
.IncompleteUserClassError
):
459 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
462 def test_notif_iterator_class_missing_next(self
):
463 class MyIter(bt2
.UserNotificationIterator
):
467 with self
.assertRaises(bt2
.IncompleteUserClassError
):
468 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
471 def test_create_notification_iterator(self
):
472 class MyIter(bt2
.UserNotificationIterator
):
483 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
488 notif_iter
= source
.create_notification_iterator()
489 self
.assertTrue(created
)
492 class UserSourceCompClassTestCase(unittest
.TestCase
):
493 def test_missing_notif_iterator_class(self
):
494 with self
.assertRaises(bt2
.IncompleteUserClassError
):
495 class MyFilter(bt2
.UserFilterComponent
):
498 def test_invalid_notif_iterator_class(self
):
502 with self
.assertRaises(bt2
.IncompleteUserClassError
):
503 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=Lel
):
506 def test_notif_iterator_class_missing_get(self
):
507 class MyIter(bt2
.UserNotificationIterator
):
511 with self
.assertRaises(bt2
.IncompleteUserClassError
):
512 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
515 def test_notif_iterator_class_missing_next(self
):
516 class MyIter(bt2
.UserNotificationIterator
):
520 with self
.assertRaises(bt2
.IncompleteUserClassError
):
521 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
524 def test_create_notification_iterator(self
):
525 class MyIter(bt2
.UserNotificationIterator
):
536 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
541 notif_iter
= filter.create_notification_iterator()
542 self
.assertTrue(created
)
544 def test_set_min_input(self
):
545 class MyIter(bt2
.UserNotificationIterator
):
552 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
554 self
._maximum
_input
_notification
_iterator
_count
= 10
555 self
._minimum
_input
_notification
_iterator
_count
= 5
559 def test_set_max_input(self
):
560 class MyIter(bt2
.UserNotificationIterator
):
567 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
569 self
._maximum
_input
_notification
_iterator
_count
= 5
573 def test_add_notification_iterator(self
):
574 class MyIter(bt2
.UserNotificationIterator
):
581 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
585 def _add_notification_iterator(self
, notif_iter
):
591 source
= _create_source()
592 notif_iter
= source
.create_notification_iterator()
593 filter.add_notification_iterator(notif_iter
)
594 self
.assertTrue(added
)
596 def test_input_notif_iterators(self
):
597 class MyIter(bt2
.UserNotificationIterator
):
604 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
606 self
._maximum
_input
_notification
_iterator
_count
= 5
612 for notif_iter
in self
._input
_notification
_iterators
:
617 source
= _create_source()
618 notif_iter1
= source
.create_notification_iterator()
619 notif_iter2
= source
.create_notification_iterator()
620 filter.add_notification_iterator(notif_iter1
)
621 filter.add_notification_iterator(notif_iter2
)
623 self
.assertEqual(count
, 2)
626 class UserNotificationIteratorTestCase(unittest
.TestCase
):
628 class MyIter(bt2
.UserNotificationIterator
):
639 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
644 notif_iter
= source
.create_notification_iterator()
645 self
.assertTrue(inited
)
647 def test_destroy(self
):
648 class MyIter(bt2
.UserNotificationIterator
):
659 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
663 notif_iter
= source
.create_notification_iterator()
666 self
.assertTrue(destroyed
)
668 def test_attr_component(self
):
669 class MyIter(bt2
.UserNotificationIterator
):
671 nonlocal source
, is_same
672 is_same
= self
.component
is source
680 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
685 notif_iter
= source
.create_notification_iterator()
686 self
.assertTrue(is_same
)
688 def test_next_raises_stop(self
):
689 class MyIter(bt2
.UserNotificationIterator
):
696 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
701 notif_iter
= source
.create_notification_iterator()
703 with self
.assertRaises(bt2
.Stop
):
706 def test_next_raises_unsupported_feature(self
):
707 class MyIter(bt2
.UserNotificationIterator
):
709 raise bt2
.UnsupportedFeature
714 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
719 notif_iter
= source
.create_notification_iterator()
721 with self
.assertRaises(bt2
.UnsupportedFeature
):
724 def test_next_raises_unknown(self
):
725 class MyIter(bt2
.UserNotificationIterator
):
727 raise TypeError('lel')
732 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
737 notif_iter
= source
.create_notification_iterator()
739 with self
.assertRaises(bt2
.Error
):
742 def test_iteration(self
):
743 source
= _create_source()
744 notif_iter
= source
.create_notification_iterator()
747 for index
, notif
in enumerate(notif_iter
):
749 self
.assertIsInstance(notif
, bt2
.BeginningOfPacketNotification
)
750 self
.assertEqual(notif
.packet
.header_field
['magic'], 0xc1fc1fc1)
751 self
.assertEqual(notif
.packet
.context_field
['something'], 194)
752 pkt_addr
= notif
.packet
.addr
754 self
.assertIsInstance(notif
, bt2
.TraceEventNotification
)
755 self
.assertEqual(notif
.event
.header_field
['ts'], 19487)
756 self
.assertEqual(notif
.event
.payload_field
['mosquito'], 'at {}'.format(index
))
758 self
.assertIsInstance(notif
, bt2
.EndOfPacketNotification
)
759 self
.assertEqual(notif
.packet
.header_field
['magic'], 0xc1fc1fc1)
760 self
.assertEqual(notif
.packet
.context_field
['something'], 194)
761 self
.assertEqual(notif
.packet
.addr
, pkt_addr
)
763 self
.assertIsInstance(notif
, bt2
.EndOfStreamNotification
)
764 self
.assertEqual(notif
.stream
.name
, 'abcdef')
766 raise RuntimeError('FAIL')
769 class GenNotificationIteratorTestCase(unittest
.TestCase
):
770 def test_attr_component(self
):
771 source
= _create_source()
772 notif_iter
= source
.create_notification_iterator()
773 self
.assertIsInstance(notif_iter
, bt2
.notification_iterator
._GenericNotificationIterator
)
774 self
.assertEqual(notif_iter
.component
.addr
, source
.addr
)