tests/bindings/python/bt2/test_comp_notif_iter.py

   1 from collections import OrderedDict
   2 import unittest
   3 import copy
   4 import bt2.notification_iterator
   5 import bt2
   6
   7
   8 def _create_ec():
   9     # clock class
  10     cc = bt2.ClockClass('salut_clock')
  11
  12     # event header
  13     eh = bt2.StructureFieldType()
  14     eh += OrderedDict((
  15         ('id', bt2.IntegerFieldType(8)),
  16         ('ts', bt2.IntegerFieldType(32, mapped_clock_class=cc)),
  17     ))
  18
  19     # packet context
  20     pc = bt2.StructureFieldType()
  21     pc += OrderedDict((
  22         ('something', bt2.IntegerFieldType(8)),
  23     ))
  24
  25     # stream class
  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
  30
  31     # event payload
  32     ep = bt2.StructureFieldType()
  33     ep += OrderedDict((
  34         ('mosquito', bt2.StringFieldType()),
  35     ))
  36
  37     # event class
  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)
  42
  43     # packet header
  44     ph = bt2.StructureFieldType()
  45     ph += OrderedDict((
  46         ('magic', bt2.IntegerFieldType(32)),
  47         ('stream_id', bt2.IntegerFieldType(16)),
  48     ))
  49
  50     # trace
  51     trace = bt2.Trace()
  52     trace.packet_header_field_type = ph
  53     trace.add_clock_class(cc)
  54     trace.add_stream_class(sc)
  55     return event_class
  56
  57
  58 def _create_event(event_class, msg):
  59     ev = event_class()
  60     ev.header_field['id'] = 0
  61     ev.header_field['ts'] = 19487
  62     ev.payload_field['mosquito'] = msg
  63     return ev
  64
  65
  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
  71     return packet
  72
  73
  74 def _create_source():
  75     class MyIter(bt2.UserNotificationIterator):
  76         def __init__(self):
  77             self._event_class = self.component._event_class
  78             self._stream = self.component._stream
  79             self._packet = _create_packet(self._stream)
  80             self._at = 0
  81             self._cur_notif = None
  82
  83         def _get(self):
  84             if self._cur_notif is None:
  85                 raise bt2.Error('nothing here!')
  86
  87             return self._cur_notif
  88
  89         def _next(self):
  90             if self._at == 0:
  91                 notif = bt2.BeginningOfPacketNotification(self._packet)
  92             elif self._at < 5:
  93                 ev = _create_event(self._event_class, 'at {}'.format(self._at))
  94                 ev.packet = self._packet
  95                 notif = bt2.TraceEventNotification(ev)
  96             elif self._at == 5:
  97                 notif = bt2.EndOfPacketNotification(self._packet)
  98             elif self._at == 6:
  99                 notif = bt2.EndOfStreamNotification(self._stream)
 100             else:
 101                 raise bt2.Stop
 102
 103             self._at += 1
 104             self._cur_notif = notif
 105
 106     class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 107         def __init__(self):
 108             self._event_class = _create_ec()
 109             self._stream = self._event_class.stream_class(name='abcdef')
 110
 111     return MySource()
 112
 113
 114 class GenCompClassTestCase(unittest.TestCase):
 115     def test_attr_name(self):
 116         class MySink(bt2.UserSinkComponent):
 117             def _consume(self):
 118                 pass
 119
 120         sink = MySink()
 121         self.assertEqual(sink.component_class.name, 'MySink')
 122
 123     def test_attr_description(self):
 124         class MySink(bt2.UserSinkComponent):
 125             'hello'
 126
 127             def _consume(self):
 128                 pass
 129
 130         sink = MySink()
 131         self.assertEqual(sink.component_class.description, 'hello')
 132
 133     def test_instantiate(self):
 134         class MySink(bt2.UserSinkComponent):
 135             'hello'
 136
 137             def __init__(self, params=None, name=None):
 138                 if params is None or name is None:
 139                     return
 140
 141                 nonlocal nl_params_a, nl_name
 142                 nl_params_a = params['a']
 143                 nl_name = name
 144
 145             def _consume(self):
 146                 pass
 147
 148         nl_params_a = None
 149         nl_name = None
 150         sink = MySink()
 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')
 157
 158
 159 class UserCompClassTestCase(unittest.TestCase):
 160     def test_attr_name(self):
 161         class MySink(bt2.UserSinkComponent):
 162             def _consume(self):
 163                 pass
 164
 165         self.assertEqual(MySink.name, 'MySink')
 166
 167     def test_attr_name_param(self):
 168         class MySink(bt2.UserSinkComponent, name='my custom name'):
 169             def _consume(self):
 170                 pass
 171
 172         self.assertEqual(MySink.name, 'my custom name')
 173
 174     def test_attr_description(self):
 175         class MySink(bt2.UserSinkComponent):
 176             'here is your description'
 177
 178             def _consume(self):
 179                 pass
 180
 181         self.assertEqual(MySink.description, 'here is your description')
 182
 183     def test_init(self):
 184         class MySink(bt2.UserSinkComponent):
 185             def __init__(self):
 186                 nonlocal inited
 187                 inited = True
 188
 189             def _consume(self):
 190                 pass
 191
 192         inited = False
 193         sink = MySink()
 194         self.assertTrue(inited)
 195
 196     def test_init_raises(self):
 197         class MySink(bt2.UserSinkComponent):
 198             def __init__(self):
 199                 raise RuntimeError('oops')
 200
 201             def _consume(self):
 202                 pass
 203
 204         with self.assertRaises(RuntimeError):
 205             sink = MySink()
 206
 207     def test_destroy(self):
 208         class MySink(bt2.UserSinkComponent):
 209             def _destroy(self):
 210                 nonlocal destroyed
 211                 destroyed = True
 212
 213             def _consume(self):
 214                 pass
 215
 216         destroyed = False
 217         sink = MySink()
 218         del sink
 219         self.assertTrue(destroyed)
 220
 221     def test_destroy_raises(self):
 222         class MySink(bt2.UserSinkComponent):
 223             def _destroy(self):
 224                 raise RuntimeError('oh oh')
 225                 nonlocal destroyed
 226                 destroyed = True
 227
 228             def _consume(self):
 229                 pass
 230
 231         destroyed = False
 232         sink = MySink()
 233         del sink
 234         self.assertFalse(destroyed)
 235
 236     def test_comp_attr_name(self):
 237         class MySink(bt2.UserSinkComponent):
 238             def _consume(self):
 239                 pass
 240
 241         sink = MySink(name='salut')
 242         self.assertEqual(sink.name, 'salut')
 243
 244     def test_comp_attr_no_name(self):
 245         class MySink(bt2.UserSinkComponent):
 246             def _consume(self):
 247                 pass
 248
 249         sink = MySink()
 250         self.assertIsNone(sink.name)
 251
 252     def test_comp_attr_class(self):
 253         class MySink(bt2.UserSinkComponent):
 254             def _consume(self):
 255                 pass
 256
 257         sink = MySink()
 258         self.assertEqual(sink.component_class.name, 'MySink')
 259
 260
 261 class UserSinkCompClassTestCase(unittest.TestCase):
 262     def test_missing_consume(self):
 263         with self.assertRaises(bt2.IncompleteUserClassError):
 264             class MySink(bt2.UserSinkComponent):
 265                 pass
 266
 267     def test_set_min_input(self):
 268         class MySink(bt2.UserSinkComponent):
 269             def __init__(self):
 270                 self._maximum_input_notification_iterator_count = 10
 271                 self._minimum_input_notification_iterator_count = 5
 272
 273             def _consume(self):
 274                 pass
 275
 276         sink = MySink()
 277
 278     def test_set_max_input(self):
 279         class MySink(bt2.UserSinkComponent):
 280             def __init__(self):
 281                 self._maximum_input_notification_iterator_count = 5
 282
 283             def _consume(self):
 284                 pass
 285
 286         sink = MySink()
 287
 288     def test_consume(self):
 289         class MySink(bt2.UserSinkComponent):
 290             def __init__(self):
 291                 pass
 292
 293             def _consume(self):
 294                 nonlocal consumed
 295                 consumed = True
 296
 297         sink = MySink()
 298         consumed = False
 299         source = _create_source()
 300         notif_iter = source.create_notification_iterator()
 301         sink.add_notification_iterator(notif_iter)
 302         sink.consume()
 303         self.assertTrue(consumed)
 304
 305     def test_consume_raises(self):
 306         class MySink(bt2.UserSinkComponent):
 307             def __init__(self):
 308                 pass
 309
 310             def _consume(self):
 311                 raise RuntimeError('hello')
 312
 313         sink = MySink()
 314         source = _create_source()
 315         notif_iter = source.create_notification_iterator()
 316         sink.add_notification_iterator(notif_iter)
 317
 318         with self.assertRaises(bt2.Error):
 319             sink.consume()
 320
 321     def test_add_notification_iterator(self):
 322         class MySink(bt2.UserSinkComponent):
 323             def __init__(self):
 324                 pass
 325
 326             def _consume(self):
 327                 pass
 328
 329             def _add_notification_iterator(self, notif_iter):
 330                 nonlocal added
 331                 added = True
 332
 333         sink = MySink()
 334         added = False
 335         source = _create_source()
 336         notif_iter = source.create_notification_iterator()
 337         sink.add_notification_iterator(notif_iter)
 338         self.assertTrue(added)
 339
 340     def test_input_notif_iterators(self):
 341         class MySink(bt2.UserSinkComponent):
 342             def __init__(self):
 343                 self._maximum_input_notification_iterator_count = 5
 344
 345             def _consume(self):
 346                 nonlocal count
 347                 count = 0
 348
 349                 for notif_iter in self._input_notification_iterators:
 350                     count += 1
 351
 352         sink = MySink()
 353         count = 0
 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)
 359         sink.consume()
 360         self.assertEqual(count, 2)
 361
 362
 363 class UserSourceCompClassTestCase(unittest.TestCase):
 364     def test_missing_notif_iterator_class(self):
 365         with self.assertRaises(bt2.IncompleteUserClassError):
 366             class MySource(bt2.UserSourceComponent):
 367                 pass
 368
 369     def test_invalid_notif_iterator_class(self):
 370         class Lel:
 371             pass
 372
 373         with self.assertRaises(bt2.IncompleteUserClassError):
 374             class MySource(bt2.UserSourceComponent, notification_iterator_class=Lel):
 375                 pass
 376
 377     def test_notif_iterator_class_missing_get(self):
 378         class MyIter(bt2.UserNotificationIterator):
 379             def _next(self):
 380                 pass
 381
 382         with self.assertRaises(bt2.IncompleteUserClassError):
 383             class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 384                 pass
 385
 386     def test_notif_iterator_class_missing_next(self):
 387         class MyIter(bt2.UserNotificationIterator):
 388             def _get(self):
 389                 pass
 390
 391         with self.assertRaises(bt2.IncompleteUserClassError):
 392             class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 393                 pass
 394
 395     def test_create_notification_iterator(self):
 396         class MyIter(bt2.UserNotificationIterator):
 397             def __init__(self):
 398                 nonlocal created
 399                 created = True
 400
 401             def _next(self):
 402                 pass
 403
 404             def _get(self):
 405                 pass
 406
 407         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 408             pass
 409
 410         created = False
 411         source = MySource()
 412         notif_iter = source.create_notification_iterator()
 413         self.assertTrue(created)
 414
 415
 416 class UserSourceCompClassTestCase(unittest.TestCase):
 417     def test_missing_notif_iterator_class(self):
 418         with self.assertRaises(bt2.IncompleteUserClassError):
 419             class MyFilter(bt2.UserFilterComponent):
 420                 pass
 421
 422     def test_invalid_notif_iterator_class(self):
 423         class Lel:
 424             pass
 425
 426         with self.assertRaises(bt2.IncompleteUserClassError):
 427             class MyFilter(bt2.UserFilterComponent, notification_iterator_class=Lel):
 428                 pass
 429
 430     def test_notif_iterator_class_missing_get(self):
 431         class MyIter(bt2.UserNotificationIterator):
 432             def _next(self):
 433                 pass
 434
 435         with self.assertRaises(bt2.IncompleteUserClassError):
 436             class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 437                 pass
 438
 439     def test_notif_iterator_class_missing_next(self):
 440         class MyIter(bt2.UserNotificationIterator):
 441             def _get(self):
 442                 pass
 443
 444         with self.assertRaises(bt2.IncompleteUserClassError):
 445             class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 446                 pass
 447
 448     def test_create_notification_iterator(self):
 449         class MyIter(bt2.UserNotificationIterator):
 450             def __init__(self):
 451                 nonlocal created
 452                 created = True
 453
 454             def _next(self):
 455                 pass
 456
 457             def _get(self):
 458                 pass
 459
 460         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 461             pass
 462
 463         created = False
 464         filter = MyFilter()
 465         notif_iter = filter.create_notification_iterator()
 466         self.assertTrue(created)
 467
 468     def test_set_min_input(self):
 469         class MyIter(bt2.UserNotificationIterator):
 470             def _next(self):
 471                 pass
 472
 473             def _get(self):
 474                 pass
 475
 476         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 477             def __init__(self):
 478                 self._maximum_input_notification_iterator_count = 10
 479                 self._minimum_input_notification_iterator_count = 5
 480
 481         filter = MyFilter()
 482
 483     def test_set_max_input(self):
 484         class MyIter(bt2.UserNotificationIterator):
 485             def _next(self):
 486                 pass
 487
 488             def _get(self):
 489                 pass
 490
 491         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 492             def __init__(self):
 493                 self._maximum_input_notification_iterator_count = 5
 494
 495         filter = MyFilter()
 496
 497     def test_add_notification_iterator(self):
 498         class MyIter(bt2.UserNotificationIterator):
 499             def _next(self):
 500                 pass
 501
 502             def _get(self):
 503                 pass
 504
 505         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 506             def __init__(self):
 507                 pass
 508
 509             def _add_notification_iterator(self, notif_iter):
 510                 nonlocal added
 511                 added = True
 512
 513         filter = MyFilter()
 514         added = False
 515         source = _create_source()
 516         notif_iter = source.create_notification_iterator()
 517         filter.add_notification_iterator(notif_iter)
 518         self.assertTrue(added)
 519
 520     def test_input_notif_iterators(self):
 521         class MyIter(bt2.UserNotificationIterator):
 522             def _next(self):
 523                 pass
 524
 525             def _get(self):
 526                 pass
 527
 528         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 529             def __init__(self):
 530                 self._maximum_input_notification_iterator_count = 5
 531
 532             def _test(self):
 533                 nonlocal count
 534                 count = 0
 535
 536                 for notif_iter in self._input_notification_iterators:
 537                     count += 1
 538
 539         filter = MyFilter()
 540         count = 0
 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)
 546         filter._test()
 547         self.assertEqual(count, 2)
 548
 549
 550 class UserNotificationIteratorTestCase(unittest.TestCase):
 551     def test_init(self):
 552         class MyIter(bt2.UserNotificationIterator):
 553             def __init__(self):
 554                 nonlocal inited
 555                 inited = True
 556
 557             def _next(self):
 558                 pass
 559
 560             def _get(self):
 561                 pass
 562
 563         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 564             pass
 565
 566         inited = False
 567         source = MySource()
 568         notif_iter = source.create_notification_iterator()
 569         self.assertTrue(inited)
 570
 571     def test_destroy(self):
 572         class MyIter(bt2.UserNotificationIterator):
 573             def _next(self):
 574                 pass
 575
 576             def _get(self):
 577                 pass
 578
 579             def _destroy(self):
 580                 nonlocal destroyed
 581                 destroyed = True
 582
 583         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 584             pass
 585
 586         source = MySource()
 587         notif_iter = source.create_notification_iterator()
 588         destroyed = False
 589         del notif_iter
 590         self.assertTrue(destroyed)
 591
 592     def test_attr_component(self):
 593         class MyIter(bt2.UserNotificationIterator):
 594             def __init__(self):
 595                 nonlocal source, is_same
 596                 is_same = self.component is source
 597
 598             def _next(self):
 599                 pass
 600
 601             def _get(self):
 602                 pass
 603
 604         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 605             pass
 606
 607         source = MySource()
 608         is_same = False
 609         notif_iter = source.create_notification_iterator()
 610         self.assertTrue(is_same)
 611
 612     def test_next_raises_stop(self):
 613         class MyIter(bt2.UserNotificationIterator):
 614             def _next(self):
 615                 raise bt2.Stop
 616
 617             def _get(self):
 618                 pass
 619
 620         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 621             pass
 622
 623         source = MySource()
 624         is_same = False
 625         notif_iter = source.create_notification_iterator()
 626
 627         with self.assertRaises(bt2.Stop):
 628             notif_iter.next()
 629
 630     def test_next_raises_unsupported_feature(self):
 631         class MyIter(bt2.UserNotificationIterator):
 632             def _next(self):
 633                 raise bt2.UnsupportedFeature
 634
 635             def _get(self):
 636                 pass
 637
 638         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 639             pass
 640
 641         source = MySource()
 642         is_same = False
 643         notif_iter = source.create_notification_iterator()
 644
 645         with self.assertRaises(bt2.UnsupportedFeature):
 646             notif_iter.next()
 647
 648     def test_next_raises_unknown(self):
 649         class MyIter(bt2.UserNotificationIterator):
 650             def _next(self):
 651                 raise TypeError('lel')
 652
 653             def _get(self):
 654                 pass
 655
 656         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 657             pass
 658
 659         source = MySource()
 660         is_same = False
 661         notif_iter = source.create_notification_iterator()
 662
 663         with self.assertRaises(bt2.Error):
 664             notif_iter.next()
 665
 666     def test_iteration(self):
 667         source = _create_source()
 668         notif_iter = source.create_notification_iterator()
 669         pkt_addr = None
 670
 671         for index, notif in enumerate(notif_iter):
 672             if index == 0:
 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
 677             elif index < 5:
 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))
 681             elif index == 5:
 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)
 686             elif index == 6:
 687                 self.assertIsInstance(notif, bt2.EndOfStreamNotification)
 688                 self.assertEqual(notif.stream.name, 'abcdef')
 689             else:
 690                 raise RuntimeError('FAIL')
 691
 692
 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)