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             '''This is the description.
 126
 127             This is the help.
 128
 129             This too:
 130
 131                 * And this.
 132                 * And also that.
 133
 134             Voilà.
 135             '''
 136
 137             def _consume(self):
 138                 pass
 139
 140         sink = MySink()
 141         self.assertEqual(sink.component_class.description,
 142                          'This is the description.')
 143
 144     def test_attr_help(self):
 145         class MySink(bt2.UserSinkComponent):
 146             '''This is the description.
 147
 148             This is the help.
 149
 150             This too:
 151
 152                 * And this.
 153                 * And also that.
 154
 155             Voilà.
 156             '''
 157
 158             def _consume(self):
 159                 pass
 160
 161         sink = MySink()
 162         expected_help = '''This is the help.
 163
 164 This too:
 165
 166     * And this.
 167     * And also that.
 168
 169 Voilà.'''
 170         self.assertEqual(sink.component_class.help, expected_help)
 171
 172     def test_instantiate(self):
 173         class MySink(bt2.UserSinkComponent):
 174             'hello'
 175
 176             def __init__(self, params=None, name=None):
 177                 if params is None or name is None:
 178                     return
 179
 180                 nonlocal nl_params_a, nl_name
 181                 nl_params_a = params['a']
 182                 nl_name = name
 183
 184             def _consume(self):
 185                 pass
 186
 187         nl_params_a = None
 188         nl_name = None
 189         sink = MySink()
 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')
 196
 197
 198 class UserCompClassTestCase(unittest.TestCase):
 199     def test_attr_name(self):
 200         class MySink(bt2.UserSinkComponent):
 201             def _consume(self):
 202                 pass
 203
 204         self.assertEqual(MySink.name, 'MySink')
 205
 206     def test_attr_name_param(self):
 207         class MySink(bt2.UserSinkComponent, name='my custom name'):
 208             def _consume(self):
 209                 pass
 210
 211         self.assertEqual(MySink.name, 'my custom name')
 212
 213     def test_attr_description(self):
 214         class MySink(bt2.UserSinkComponent):
 215             '''This is the description.
 216
 217             This is the help.
 218
 219             This too:
 220
 221                 * And this.
 222                 * And also that.
 223
 224             Voilà.
 225             '''
 226
 227             def _consume(self):
 228                 pass
 229
 230         self.assertEqual(MySink.description, 'This is the description.')
 231
 232     def test_attr_help(self):
 233         class MySink(bt2.UserSinkComponent):
 234             '''This is the description.
 235
 236             This is the help.
 237
 238             This too:
 239
 240                 * And this.
 241                 * And also that.
 242
 243             Voilà.
 244             '''
 245
 246             def _consume(self):
 247                 pass
 248
 249         expected_help = '''This is the help.
 250
 251 This too:
 252
 253     * And this.
 254     * And also that.
 255
 256 Voilà.'''
 257         self.assertEqual(MySink.help, expected_help)
 258
 259     def test_query_missing(self):
 260         class MySink(bt2.UserSinkComponent):
 261             def _consume(self):
 262                 pass
 263
 264         with self.assertRaises(bt2.Error):
 265             MySink.query('salut')
 266
 267     def test_query_raises(self):
 268         class MySink(bt2.UserSinkComponent):
 269             def _consume(self):
 270                 pass
 271
 272             @staticmethod
 273             def _query(obj, params):
 274                 raise ValueError
 275
 276         with self.assertRaises(bt2.Error):
 277             MySink.query('salut')
 278
 279     def test_query_gets_none_params(self):
 280         class MySink(bt2.UserSinkComponent):
 281             def _consume(self):
 282                 pass
 283
 284             @staticmethod
 285             def _query(obj, params):
 286                 nonlocal recv_params
 287                 recv_params = params
 288
 289         recv_params = NotImplemented
 290         MySink.query('allo', None)
 291         self.assertIsNone(recv_params)
 292
 293     def test_query_gets_same_params(self):
 294         class MySink(bt2.UserSinkComponent):
 295             def _consume(self):
 296                 pass
 297
 298             @staticmethod
 299             def _query(obj, params):
 300                 nonlocal recv_params
 301                 recv_params = params
 302
 303         recv_params = NotImplemented
 304         params = bt2.create_value(23)
 305         MySink.query('allo', params)
 306         self.assertEqual(recv_params.addr, params.addr)
 307
 308     def test_query_obj(self):
 309         class MySink(bt2.UserSinkComponent):
 310             def _consume(self):
 311                 pass
 312
 313             @staticmethod
 314             def _query(obj, params):
 315                 nonlocal recv_obj
 316                 recv_obj = obj
 317
 318         recv_obj = None
 319         MySink.query('salut')
 320         self.assertEqual(recv_obj, 'salut')
 321
 322     def test_query_returns_none(self):
 323         class MySink(bt2.UserSinkComponent):
 324             def _consume(self):
 325                 pass
 326
 327             @staticmethod
 328             def _query(obj, params):
 329                 pass
 330
 331         self.assertIsNone(MySink.query('allo', 177))
 332
 333     def test_query_returns_params(self):
 334         class MySink(bt2.UserSinkComponent):
 335             def _consume(self):
 336                 pass
 337
 338             @staticmethod
 339             def _query(obj, params):
 340                 return {'obj': obj, 'params': params}
 341
 342         results = MySink.query('hello', (45, 'lol'))
 343         self.assertEqual(results['obj'], 'hello')
 344         self.assertEqual(results['params'], (45, 'lol'))
 345
 346     def test_init(self):
 347         class MySink(bt2.UserSinkComponent):
 348             def __init__(self):
 349                 nonlocal inited
 350                 inited = True
 351
 352             def _consume(self):
 353                 pass
 354
 355         inited = False
 356         sink = MySink()
 357         self.assertTrue(inited)
 358
 359     def test_init_raises(self):
 360         class MySink(bt2.UserSinkComponent):
 361             def __init__(self):
 362                 raise RuntimeError('oops')
 363
 364             def _consume(self):
 365                 pass
 366
 367         with self.assertRaises(RuntimeError):
 368             sink = MySink()
 369
 370     def test_destroy(self):
 371         class MySink(bt2.UserSinkComponent):
 372             def _destroy(self):
 373                 nonlocal destroyed
 374                 destroyed = True
 375
 376             def _consume(self):
 377                 pass
 378
 379         destroyed = False
 380         sink = MySink()
 381         del sink
 382         self.assertTrue(destroyed)
 383
 384     def test_destroy_raises(self):
 385         class MySink(bt2.UserSinkComponent):
 386             def _destroy(self):
 387                 raise RuntimeError('oh oh')
 388                 nonlocal destroyed
 389                 destroyed = True
 390
 391             def _consume(self):
 392                 pass
 393
 394         destroyed = False
 395         sink = MySink()
 396         del sink
 397         self.assertFalse(destroyed)
 398
 399     def test_comp_attr_name(self):
 400         class MySink(bt2.UserSinkComponent):
 401             def _consume(self):
 402                 pass
 403
 404         sink = MySink(name='salut')
 405         self.assertEqual(sink.name, 'salut')
 406
 407     def test_comp_attr_no_name(self):
 408         class MySink(bt2.UserSinkComponent):
 409             def _consume(self):
 410                 pass
 411
 412         sink = MySink()
 413         self.assertIsNone(sink.name)
 414
 415     def test_comp_attr_class(self):
 416         class MySink(bt2.UserSinkComponent):
 417             def _consume(self):
 418                 pass
 419
 420         sink = MySink()
 421         self.assertEqual(sink.component_class.name, 'MySink')
 422
 423
 424 class UserSinkCompClassTestCase(unittest.TestCase):
 425     def test_missing_consume(self):
 426         with self.assertRaises(bt2.IncompleteUserClassError):
 427             class MySink(bt2.UserSinkComponent):
 428                 pass
 429
 430     def test_set_min_input(self):
 431         class MySink(bt2.UserSinkComponent):
 432             def __init__(self):
 433                 self._maximum_input_notification_iterator_count = 10
 434                 self._minimum_input_notification_iterator_count = 5
 435
 436             def _consume(self):
 437                 pass
 438
 439         sink = MySink()
 440
 441     def test_set_max_input(self):
 442         class MySink(bt2.UserSinkComponent):
 443             def __init__(self):
 444                 self._maximum_input_notification_iterator_count = 5
 445
 446             def _consume(self):
 447                 pass
 448
 449         sink = MySink()
 450
 451     def test_consume(self):
 452         class MySink(bt2.UserSinkComponent):
 453             def __init__(self):
 454                 pass
 455
 456             def _consume(self):
 457                 nonlocal consumed
 458                 consumed = True
 459
 460         sink = MySink()
 461         consumed = False
 462         source = _create_source()
 463         notif_iter = source.create_notification_iterator()
 464         sink.add_notification_iterator(notif_iter)
 465         sink.consume()
 466         self.assertTrue(consumed)
 467
 468     def test_consume_raises(self):
 469         class MySink(bt2.UserSinkComponent):
 470             def __init__(self):
 471                 pass
 472
 473             def _consume(self):
 474                 raise RuntimeError('hello')
 475
 476         sink = MySink()
 477         source = _create_source()
 478         notif_iter = source.create_notification_iterator()
 479         sink.add_notification_iterator(notif_iter)
 480
 481         with self.assertRaises(bt2.Error):
 482             sink.consume()
 483
 484     def test_add_notification_iterator(self):
 485         class MySink(bt2.UserSinkComponent):
 486             def __init__(self):
 487                 pass
 488
 489             def _consume(self):
 490                 pass
 491
 492             def _add_notification_iterator(self, notif_iter):
 493                 nonlocal added
 494                 added = True
 495
 496         sink = MySink()
 497         added = False
 498         source = _create_source()
 499         notif_iter = source.create_notification_iterator()
 500         sink.add_notification_iterator(notif_iter)
 501         self.assertTrue(added)
 502
 503     def test_input_notif_iterators(self):
 504         class MySink(bt2.UserSinkComponent):
 505             def __init__(self):
 506                 self._maximum_input_notification_iterator_count = 5
 507
 508             def _consume(self):
 509                 nonlocal count
 510                 count = 0
 511
 512                 for notif_iter in self._input_notification_iterators:
 513                     count += 1
 514
 515         sink = MySink()
 516         count = 0
 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)
 522         sink.consume()
 523         self.assertEqual(count, 2)
 524
 525
 526 class UserSourceCompClassTestCase(unittest.TestCase):
 527     def test_missing_notif_iterator_class(self):
 528         with self.assertRaises(bt2.IncompleteUserClassError):
 529             class MySource(bt2.UserSourceComponent):
 530                 pass
 531
 532     def test_invalid_notif_iterator_class(self):
 533         class Lel:
 534             pass
 535
 536         with self.assertRaises(bt2.IncompleteUserClassError):
 537             class MySource(bt2.UserSourceComponent, notification_iterator_class=Lel):
 538                 pass
 539
 540     def test_notif_iterator_class_missing_get(self):
 541         class MyIter(bt2.UserNotificationIterator):
 542             def _next(self):
 543                 pass
 544
 545         with self.assertRaises(bt2.IncompleteUserClassError):
 546             class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 547                 pass
 548
 549     def test_notif_iterator_class_missing_next(self):
 550         class MyIter(bt2.UserNotificationIterator):
 551             def _get(self):
 552                 pass
 553
 554         with self.assertRaises(bt2.IncompleteUserClassError):
 555             class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 556                 pass
 557
 558     def test_create_notification_iterator(self):
 559         class MyIter(bt2.UserNotificationIterator):
 560             def __init__(self):
 561                 nonlocal created
 562                 created = True
 563
 564             def _next(self):
 565                 pass
 566
 567             def _get(self):
 568                 pass
 569
 570         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 571             pass
 572
 573         created = False
 574         source = MySource()
 575         notif_iter = source.create_notification_iterator()
 576         self.assertTrue(created)
 577
 578
 579 class UserSourceCompClassTestCase(unittest.TestCase):
 580     def test_missing_notif_iterator_class(self):
 581         with self.assertRaises(bt2.IncompleteUserClassError):
 582             class MyFilter(bt2.UserFilterComponent):
 583                 pass
 584
 585     def test_invalid_notif_iterator_class(self):
 586         class Lel:
 587             pass
 588
 589         with self.assertRaises(bt2.IncompleteUserClassError):
 590             class MyFilter(bt2.UserFilterComponent, notification_iterator_class=Lel):
 591                 pass
 592
 593     def test_notif_iterator_class_missing_get(self):
 594         class MyIter(bt2.UserNotificationIterator):
 595             def _next(self):
 596                 pass
 597
 598         with self.assertRaises(bt2.IncompleteUserClassError):
 599             class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 600                 pass
 601
 602     def test_notif_iterator_class_missing_next(self):
 603         class MyIter(bt2.UserNotificationIterator):
 604             def _get(self):
 605                 pass
 606
 607         with self.assertRaises(bt2.IncompleteUserClassError):
 608             class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 609                 pass
 610
 611     def test_create_notification_iterator(self):
 612         class MyIter(bt2.UserNotificationIterator):
 613             def __init__(self):
 614                 nonlocal created
 615                 created = True
 616
 617             def _next(self):
 618                 pass
 619
 620             def _get(self):
 621                 pass
 622
 623         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 624             pass
 625
 626         created = False
 627         filter = MyFilter()
 628         notif_iter = filter.create_notification_iterator()
 629         self.assertTrue(created)
 630
 631     def test_set_min_input(self):
 632         class MyIter(bt2.UserNotificationIterator):
 633             def _next(self):
 634                 pass
 635
 636             def _get(self):
 637                 pass
 638
 639         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 640             def __init__(self):
 641                 self._maximum_input_notification_iterator_count = 10
 642                 self._minimum_input_notification_iterator_count = 5
 643
 644         filter = MyFilter()
 645
 646     def test_set_max_input(self):
 647         class MyIter(bt2.UserNotificationIterator):
 648             def _next(self):
 649                 pass
 650
 651             def _get(self):
 652                 pass
 653
 654         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 655             def __init__(self):
 656                 self._maximum_input_notification_iterator_count = 5
 657
 658         filter = MyFilter()
 659
 660     def test_add_notification_iterator(self):
 661         class MyIter(bt2.UserNotificationIterator):
 662             def _next(self):
 663                 pass
 664
 665             def _get(self):
 666                 pass
 667
 668         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 669             def __init__(self):
 670                 pass
 671
 672             def _add_notification_iterator(self, notif_iter):
 673                 nonlocal added
 674                 added = True
 675
 676         filter = MyFilter()
 677         added = False
 678         source = _create_source()
 679         notif_iter = source.create_notification_iterator()
 680         filter.add_notification_iterator(notif_iter)
 681         self.assertTrue(added)
 682
 683     def test_input_notif_iterators(self):
 684         class MyIter(bt2.UserNotificationIterator):
 685             def _next(self):
 686                 pass
 687
 688             def _get(self):
 689                 pass
 690
 691         class MyFilter(bt2.UserFilterComponent, notification_iterator_class=MyIter):
 692             def __init__(self):
 693                 self._maximum_input_notification_iterator_count = 5
 694
 695             def _test(self):
 696                 nonlocal count
 697                 count = 0
 698
 699                 for notif_iter in self._input_notification_iterators:
 700                     count += 1
 701
 702         filter = MyFilter()
 703         count = 0
 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)
 709         filter._test()
 710         self.assertEqual(count, 2)
 711
 712
 713 class UserNotificationIteratorTestCase(unittest.TestCase):
 714     def test_init(self):
 715         class MyIter(bt2.UserNotificationIterator):
 716             def __init__(self):
 717                 nonlocal inited
 718                 inited = True
 719
 720             def _next(self):
 721                 pass
 722
 723             def _get(self):
 724                 pass
 725
 726         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 727             pass
 728
 729         inited = False
 730         source = MySource()
 731         notif_iter = source.create_notification_iterator()
 732         self.assertTrue(inited)
 733
 734     def test_destroy(self):
 735         class MyIter(bt2.UserNotificationIterator):
 736             def _next(self):
 737                 pass
 738
 739             def _get(self):
 740                 pass
 741
 742             def _destroy(self):
 743                 nonlocal destroyed
 744                 destroyed = True
 745
 746         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 747             pass
 748
 749         source = MySource()
 750         notif_iter = source.create_notification_iterator()
 751         destroyed = False
 752         del notif_iter
 753         self.assertTrue(destroyed)
 754
 755     def test_attr_component(self):
 756         class MyIter(bt2.UserNotificationIterator):
 757             def __init__(self):
 758                 nonlocal source, is_same
 759                 is_same = self.component is source
 760
 761             def _next(self):
 762                 pass
 763
 764             def _get(self):
 765                 pass
 766
 767         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 768             pass
 769
 770         source = MySource()
 771         is_same = False
 772         notif_iter = source.create_notification_iterator()
 773         self.assertTrue(is_same)
 774
 775     def test_next_raises_stop(self):
 776         class MyIter(bt2.UserNotificationIterator):
 777             def _next(self):
 778                 raise bt2.Stop
 779
 780             def _get(self):
 781                 pass
 782
 783         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 784             pass
 785
 786         source = MySource()
 787         is_same = False
 788         notif_iter = source.create_notification_iterator()
 789
 790         with self.assertRaises(bt2.Stop):
 791             notif_iter.next()
 792
 793     def test_next_raises_unsupported_feature(self):
 794         class MyIter(bt2.UserNotificationIterator):
 795             def _next(self):
 796                 raise bt2.UnsupportedFeature
 797
 798             def _get(self):
 799                 pass
 800
 801         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 802             pass
 803
 804         source = MySource()
 805         is_same = False
 806         notif_iter = source.create_notification_iterator()
 807
 808         with self.assertRaises(bt2.UnsupportedFeature):
 809             notif_iter.next()
 810
 811     def test_next_raises_unknown(self):
 812         class MyIter(bt2.UserNotificationIterator):
 813             def _next(self):
 814                 raise TypeError('lel')
 815
 816             def _get(self):
 817                 pass
 818
 819         class MySource(bt2.UserSourceComponent, notification_iterator_class=MyIter):
 820             pass
 821
 822         source = MySource()
 823         is_same = False
 824         notif_iter = source.create_notification_iterator()
 825
 826         with self.assertRaises(bt2.Error):
 827             notif_iter.next()
 828
 829     def test_iteration(self):
 830         source = _create_source()
 831         notif_iter = source.create_notification_iterator()
 832         pkt_addr = None
 833
 834         for index, notif in enumerate(notif_iter):
 835             if index == 0:
 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
 840             elif index < 5:
 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))
 844             elif index == 5:
 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)
 849             elif index == 6:
 850                 self.assertIsInstance(notif, bt2.EndOfStreamNotification)
 851                 self.assertEqual(notif.stream.name, 'abcdef')
 852             else:
 853                 raise RuntimeError('FAIL')
 854
 855
 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)