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
)
259 def test_query_info_missing(self
):
260 class MySink(bt2
.UserSinkComponent
):
264 with self
.assertRaises(bt2
.Error
):
265 MySink
.query_info('salut')
267 def test_query_info_raises(self
):
268 class MySink(bt2
.UserSinkComponent
):
273 def _query_info(action
, params
):
276 with self
.assertRaises(bt2
.Error
):
277 MySink
.query_info('salut')
279 def test_query_info_gets_none_params(self
):
280 class MySink(bt2
.UserSinkComponent
):
285 def _query_info(action
, params
):
289 recv_params
= NotImplemented
290 MySink
.query_info('allo', None)
291 self
.assertIsNone(recv_params
)
293 def test_query_info_gets_same_params(self
):
294 class MySink(bt2
.UserSinkComponent
):
299 def _query_info(action
, params
):
303 recv_params
= NotImplemented
304 params
= bt2
.create_value(23)
305 MySink
.query_info('allo', params
)
306 self
.assertEqual(recv_params
.addr
, params
.addr
)
308 def test_query_info_action(self
):
309 class MySink(bt2
.UserSinkComponent
):
314 def _query_info(action
, params
):
319 MySink
.query_info('salut')
320 self
.assertEqual(recv_action
, 'salut')
322 def test_query_info_returns_none(self
):
323 class MySink(bt2
.UserSinkComponent
):
328 def _query_info(action
, params
):
331 self
.assertIsNone(MySink
.query_info('allo', 177))
333 def test_query_info_returns_params(self
):
334 class MySink(bt2
.UserSinkComponent
):
339 def _query_info(action
, params
):
340 return {'action': action
, 'params': params
}
342 results
= MySink
.query_info('hello', (45, 'lol'))
343 self
.assertEqual(results
['action'], 'hello')
344 self
.assertEqual(results
['params'], (45, 'lol'))
347 class MySink(bt2
.UserSinkComponent
):
357 self
.assertTrue(inited
)
359 def test_init_raises(self
):
360 class MySink(bt2
.UserSinkComponent
):
362 raise RuntimeError('oops')
367 with self
.assertRaises(RuntimeError):
370 def test_destroy(self
):
371 class MySink(bt2
.UserSinkComponent
):
382 self
.assertTrue(destroyed
)
384 def test_destroy_raises(self
):
385 class MySink(bt2
.UserSinkComponent
):
387 raise RuntimeError('oh oh')
397 self
.assertFalse(destroyed
)
399 def test_comp_attr_name(self
):
400 class MySink(bt2
.UserSinkComponent
):
404 sink
= MySink(name
='salut')
405 self
.assertEqual(sink
.name
, 'salut')
407 def test_comp_attr_no_name(self
):
408 class MySink(bt2
.UserSinkComponent
):
413 self
.assertIsNone(sink
.name
)
415 def test_comp_attr_class(self
):
416 class MySink(bt2
.UserSinkComponent
):
421 self
.assertEqual(sink
.component_class
.name
, 'MySink')
424 class UserSinkCompClassTestCase(unittest
.TestCase
):
425 def test_missing_consume(self
):
426 with self
.assertRaises(bt2
.IncompleteUserClassError
):
427 class MySink(bt2
.UserSinkComponent
):
430 def test_set_min_input(self
):
431 class MySink(bt2
.UserSinkComponent
):
433 self
._maximum
_input
_notification
_iterator
_count
= 10
434 self
._minimum
_input
_notification
_iterator
_count
= 5
441 def test_set_max_input(self
):
442 class MySink(bt2
.UserSinkComponent
):
444 self
._maximum
_input
_notification
_iterator
_count
= 5
451 def test_consume(self
):
452 class MySink(bt2
.UserSinkComponent
):
462 source
= _create_source()
463 notif_iter
= source
.create_notification_iterator()
464 sink
.add_notification_iterator(notif_iter
)
466 self
.assertTrue(consumed
)
468 def test_consume_raises(self
):
469 class MySink(bt2
.UserSinkComponent
):
474 raise RuntimeError('hello')
477 source
= _create_source()
478 notif_iter
= source
.create_notification_iterator()
479 sink
.add_notification_iterator(notif_iter
)
481 with self
.assertRaises(bt2
.Error
):
484 def test_add_notification_iterator(self
):
485 class MySink(bt2
.UserSinkComponent
):
492 def _add_notification_iterator(self
, notif_iter
):
498 source
= _create_source()
499 notif_iter
= source
.create_notification_iterator()
500 sink
.add_notification_iterator(notif_iter
)
501 self
.assertTrue(added
)
503 def test_input_notif_iterators(self
):
504 class MySink(bt2
.UserSinkComponent
):
506 self
._maximum
_input
_notification
_iterator
_count
= 5
512 for notif_iter
in self
._input
_notification
_iterators
:
517 source
= _create_source()
518 notif_iter1
= source
.create_notification_iterator()
519 notif_iter2
= source
.create_notification_iterator()
520 sink
.add_notification_iterator(notif_iter1
)
521 sink
.add_notification_iterator(notif_iter2
)
523 self
.assertEqual(count
, 2)
526 class UserSourceCompClassTestCase(unittest
.TestCase
):
527 def test_missing_notif_iterator_class(self
):
528 with self
.assertRaises(bt2
.IncompleteUserClassError
):
529 class MySource(bt2
.UserSourceComponent
):
532 def test_invalid_notif_iterator_class(self
):
536 with self
.assertRaises(bt2
.IncompleteUserClassError
):
537 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=Lel
):
540 def test_notif_iterator_class_missing_get(self
):
541 class MyIter(bt2
.UserNotificationIterator
):
545 with self
.assertRaises(bt2
.IncompleteUserClassError
):
546 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
549 def test_notif_iterator_class_missing_next(self
):
550 class MyIter(bt2
.UserNotificationIterator
):
554 with self
.assertRaises(bt2
.IncompleteUserClassError
):
555 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
558 def test_create_notification_iterator(self
):
559 class MyIter(bt2
.UserNotificationIterator
):
570 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
575 notif_iter
= source
.create_notification_iterator()
576 self
.assertTrue(created
)
579 class UserSourceCompClassTestCase(unittest
.TestCase
):
580 def test_missing_notif_iterator_class(self
):
581 with self
.assertRaises(bt2
.IncompleteUserClassError
):
582 class MyFilter(bt2
.UserFilterComponent
):
585 def test_invalid_notif_iterator_class(self
):
589 with self
.assertRaises(bt2
.IncompleteUserClassError
):
590 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=Lel
):
593 def test_notif_iterator_class_missing_get(self
):
594 class MyIter(bt2
.UserNotificationIterator
):
598 with self
.assertRaises(bt2
.IncompleteUserClassError
):
599 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
602 def test_notif_iterator_class_missing_next(self
):
603 class MyIter(bt2
.UserNotificationIterator
):
607 with self
.assertRaises(bt2
.IncompleteUserClassError
):
608 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
611 def test_create_notification_iterator(self
):
612 class MyIter(bt2
.UserNotificationIterator
):
623 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
628 notif_iter
= filter.create_notification_iterator()
629 self
.assertTrue(created
)
631 def test_set_min_input(self
):
632 class MyIter(bt2
.UserNotificationIterator
):
639 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
641 self
._maximum
_input
_notification
_iterator
_count
= 10
642 self
._minimum
_input
_notification
_iterator
_count
= 5
646 def test_set_max_input(self
):
647 class MyIter(bt2
.UserNotificationIterator
):
654 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
656 self
._maximum
_input
_notification
_iterator
_count
= 5
660 def test_add_notification_iterator(self
):
661 class MyIter(bt2
.UserNotificationIterator
):
668 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
672 def _add_notification_iterator(self
, notif_iter
):
678 source
= _create_source()
679 notif_iter
= source
.create_notification_iterator()
680 filter.add_notification_iterator(notif_iter
)
681 self
.assertTrue(added
)
683 def test_input_notif_iterators(self
):
684 class MyIter(bt2
.UserNotificationIterator
):
691 class MyFilter(bt2
.UserFilterComponent
, notification_iterator_class
=MyIter
):
693 self
._maximum
_input
_notification
_iterator
_count
= 5
699 for notif_iter
in self
._input
_notification
_iterators
:
704 source
= _create_source()
705 notif_iter1
= source
.create_notification_iterator()
706 notif_iter2
= source
.create_notification_iterator()
707 filter.add_notification_iterator(notif_iter1
)
708 filter.add_notification_iterator(notif_iter2
)
710 self
.assertEqual(count
, 2)
713 class UserNotificationIteratorTestCase(unittest
.TestCase
):
715 class MyIter(bt2
.UserNotificationIterator
):
726 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
731 notif_iter
= source
.create_notification_iterator()
732 self
.assertTrue(inited
)
734 def test_destroy(self
):
735 class MyIter(bt2
.UserNotificationIterator
):
746 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
750 notif_iter
= source
.create_notification_iterator()
753 self
.assertTrue(destroyed
)
755 def test_attr_component(self
):
756 class MyIter(bt2
.UserNotificationIterator
):
758 nonlocal source
, is_same
759 is_same
= self
.component
is source
767 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
772 notif_iter
= source
.create_notification_iterator()
773 self
.assertTrue(is_same
)
775 def test_next_raises_stop(self
):
776 class MyIter(bt2
.UserNotificationIterator
):
783 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
788 notif_iter
= source
.create_notification_iterator()
790 with self
.assertRaises(bt2
.Stop
):
793 def test_next_raises_unsupported_feature(self
):
794 class MyIter(bt2
.UserNotificationIterator
):
796 raise bt2
.UnsupportedFeature
801 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
806 notif_iter
= source
.create_notification_iterator()
808 with self
.assertRaises(bt2
.UnsupportedFeature
):
811 def test_next_raises_unknown(self
):
812 class MyIter(bt2
.UserNotificationIterator
):
814 raise TypeError('lel')
819 class MySource(bt2
.UserSourceComponent
, notification_iterator_class
=MyIter
):
824 notif_iter
= source
.create_notification_iterator()
826 with self
.assertRaises(bt2
.Error
):
829 def test_iteration(self
):
830 source
= _create_source()
831 notif_iter
= source
.create_notification_iterator()
834 for index
, notif
in enumerate(notif_iter
):
836 self
.assertIsInstance(notif
, bt2
.BeginningOfPacketNotification
)
837 self
.assertEqual(notif
.packet
.header_field
['magic'], 0xc1fc1fc1)
838 self
.assertEqual(notif
.packet
.context_field
['something'], 194)
839 pkt_addr
= notif
.packet
.addr
841 self
.assertIsInstance(notif
, bt2
.TraceEventNotification
)
842 self
.assertEqual(notif
.event
.header_field
['ts'], 19487)
843 self
.assertEqual(notif
.event
.payload_field
['mosquito'], 'at {}'.format(index
))
845 self
.assertIsInstance(notif
, bt2
.EndOfPacketNotification
)
846 self
.assertEqual(notif
.packet
.header_field
['magic'], 0xc1fc1fc1)
847 self
.assertEqual(notif
.packet
.context_field
['something'], 194)
848 self
.assertEqual(notif
.packet
.addr
, pkt_addr
)
850 self
.assertIsInstance(notif
, bt2
.EndOfStreamNotification
)
851 self
.assertEqual(notif
.stream
.name
, 'abcdef')
853 raise RuntimeError('FAIL')
856 class GenNotificationIteratorTestCase(unittest
.TestCase
):
857 def test_attr_component(self
):
858 source
= _create_source()
859 notif_iter
= source
.create_notification_iterator()
860 self
.assertIsInstance(notif_iter
, bt2
.notification_iterator
._GenericNotificationIterator
)
861 self
.assertEqual(notif_iter
.component
.addr
, source
.addr
)