tests/bindings/python/bt2/test_port.py

   1 from bt2 import values
   2 import unittest
   3 import copy
   4 import bt2
   5
   6
   7 class PortTestCase(unittest.TestCase):
   8     @staticmethod
   9     def _create_comp(comp_cls, name=None):
  10         graph = bt2.Graph()
  11
  12         if name is None:
  13             name = 'comp'
  14
  15         return graph.add_component(comp_cls, name)
  16
  17     def test_src_add_output_port(self):
  18         class MyIter(bt2._UserNotificationIterator):
  19             def __next__(self):
  20                 raise bt2.Stop
  21
  22         class MySource(bt2._UserSourceComponent,
  23                        notification_iterator_class=MyIter):
  24             def __init__(comp_self, params):
  25                 port = comp_self._add_output_port('out')
  26                 self.assertEqual(port.name, 'out')
  27
  28         comp = self._create_comp(MySource)
  29         self.assertEqual(len(comp.output_ports), 1)
  30
  31
  32     def test_flt_add_output_port(self):
  33         class MyIter(bt2._UserNotificationIterator):
  34             def __next__(self):
  35                 raise bt2.Stop
  36
  37         class MyFilter(bt2._UserFilterComponent,
  38                        notification_iterator_class=MyIter):
  39             def __init__(comp_self, params):
  40                 port = comp_self._add_output_port('out')
  41                 self.assertEqual(port.name, 'out')
  42
  43         comp = self._create_comp(MyFilter)
  44         self.assertEqual(len(comp.output_ports), 1)
  45
  46     def test_flt_add_input_port(self):
  47         class MyIter(bt2._UserNotificationIterator):
  48             def __next__(self):
  49                 raise bt2.Stop
  50
  51         class MyFilter(bt2._UserFilterComponent,
  52                        notification_iterator_class=MyIter):
  53             def __init__(comp_self, params):
  54                 port = comp_self._add_input_port('in')
  55                 self.assertEqual(port.name, 'in')
  56
  57         comp = self._create_comp(MyFilter)
  58         self.assertEqual(len(comp.input_ports), 1)
  59
  60     def test_sink_add_input_port(self):
  61         class MySink(bt2._UserSinkComponent):
  62             def __init__(comp_self, params):
  63                 port = comp_self._add_input_port('in')
  64                 self.assertEqual(port.name, 'in')
  65
  66             def _consume(self):
  67                 pass
  68
  69         comp = self._create_comp(MySink)
  70         self.assertEqual(len(comp.input_ports), 1)
  71
  72     def test_user_src_output_ports_getitem(self):
  73         class MyIter(bt2._UserNotificationIterator):
  74             def __next__(self):
  75                 raise bt2.Stop
  76
  77         class MySource(bt2._UserSourceComponent,
  78                        notification_iterator_class=MyIter):
  79             def __init__(comp_self, params):
  80                 port1 = comp_self._add_output_port('clear')
  81                 port2 = comp_self._add_output_port('print')
  82                 port3 = comp_self._add_output_port('insert')
  83                 self.assertEqual(port3.addr, comp_self._output_ports['insert'].addr)
  84                 self.assertEqual(port2.addr, comp_self._output_ports['print'].addr)
  85                 self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr)
  86
  87         comp = self._create_comp(MySource)
  88
  89     def test_user_flt_output_ports_getitem(self):
  90         class MyIter(bt2._UserNotificationIterator):
  91             def __next__(self):
  92                 raise bt2.Stop
  93
  94         class MyFilter(bt2._UserFilterComponent,
  95                        notification_iterator_class=MyIter):
  96             def __init__(comp_self, params):
  97                 port1 = comp_self._add_output_port('clear')
  98                 port2 = comp_self._add_output_port('print')
  99                 port3 = comp_self._add_output_port('insert')
 100                 self.assertEqual(port3.addr, comp_self._output_ports['insert'].addr)
 101                 self.assertEqual(port2.addr, comp_self._output_ports['print'].addr)
 102                 self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr)
 103
 104         comp = self._create_comp(MyFilter)
 105
 106     def test_user_flt_input_ports_getitem(self):
 107         class MyIter(bt2._UserNotificationIterator):
 108             def __next__(self):
 109                 raise bt2.Stop
 110
 111         class MyFilter(bt2._UserFilterComponent,
 112                        notification_iterator_class=MyIter):
 113             def __init__(comp_self, params):
 114                 port1 = comp_self._add_input_port('clear')
 115                 port2 = comp_self._add_input_port('print')
 116                 port3 = comp_self._add_input_port('insert')
 117                 self.assertEqual(port3.addr, comp_self._input_ports['insert'].addr)
 118                 self.assertEqual(port2.addr, comp_self._input_ports['print'].addr)
 119                 self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr)
 120
 121         comp = self._create_comp(MyFilter)
 122
 123     def test_user_sink_input_ports_getitem(self):
 124         class MySink(bt2._UserSinkComponent):
 125             def __init__(comp_self, params):
 126                 port1 = comp_self._add_input_port('clear')
 127                 port2 = comp_self._add_input_port('print')
 128                 port3 = comp_self._add_input_port('insert')
 129                 self.assertEqual(port3.addr, comp_self._input_ports['insert'].addr)
 130                 self.assertEqual(port2.addr, comp_self._input_ports['print'].addr)
 131                 self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr)
 132
 133             def _consume(self):
 134                 pass
 135
 136         comp = self._create_comp(MySink)
 137
 138     def test_user_src_output_ports_getitem_invalid_key(self):
 139         class MyIter(bt2._UserNotificationIterator):
 140             def __next__(self):
 141                 raise bt2.Stop
 142
 143         class MySource(bt2._UserSourceComponent,
 144                        notification_iterator_class=MyIter):
 145             def __init__(comp_self, params):
 146                 comp_self._add_output_port('clear')
 147                 comp_self._add_output_port('print')
 148                 comp_self._add_output_port('insert')
 149
 150                 with self.assertRaises(KeyError):
 151                     comp_self._output_ports['hello']
 152
 153         comp = self._create_comp(MySource)
 154
 155     def test_user_flt_output_ports_getitem_invalid_key(self):
 156         class MyIter(bt2._UserNotificationIterator):
 157             def __next__(self):
 158                 raise bt2.Stop
 159
 160         class MyFilter(bt2._UserFilterComponent,
 161                        notification_iterator_class=MyIter):
 162             def __init__(comp_self, params):
 163                 comp_self._add_output_port('clear')
 164                 comp_self._add_output_port('print')
 165                 comp_self._add_output_port('insert')
 166
 167                 with self.assertRaises(KeyError):
 168                     comp_self._output_ports['hello']
 169
 170         comp = self._create_comp(MyFilter)
 171
 172     def test_user_flt_input_ports_getitem_invalid_key(self):
 173         class MyIter(bt2._UserNotificationIterator):
 174             def __next__(self):
 175                 raise bt2.Stop
 176
 177         class MyFilter(bt2._UserFilterComponent,
 178                        notification_iterator_class=MyIter):
 179             def __init__(comp_self, params):
 180                 comp_self._add_input_port('clear')
 181                 comp_self._add_input_port('print')
 182                 comp_self._add_input_port('insert')
 183
 184                 with self.assertRaises(KeyError):
 185                     comp_self._input_ports['hello']
 186
 187         comp = self._create_comp(MyFilter)
 188
 189     def test_user_sink_input_ports_getitem_invalid_key(self):
 190         class MySink(bt2._UserSinkComponent):
 191             def __init__(comp_self, params):
 192                 comp_self._add_input_port('clear')
 193                 comp_self._add_input_port('print')
 194                 comp_self._add_input_port('insert')
 195
 196                 with self.assertRaises(KeyError):
 197                     comp_self._input_ports['hello']
 198
 199             def _consume(self):
 200                 pass
 201
 202         comp = self._create_comp(MySink)
 203
 204     def test_user_src_output_ports_len(self):
 205         class MyIter(bt2._UserNotificationIterator):
 206             def __next__(self):
 207                 raise bt2.Stop
 208
 209         class MySource(bt2._UserSourceComponent,
 210                        notification_iterator_class=MyIter):
 211             def __init__(comp_self, params):
 212                 comp_self._add_output_port('clear')
 213                 comp_self._add_output_port('print')
 214                 comp_self._add_output_port('insert')
 215                 self.assertEqual(len(comp_self._output_ports), 3)
 216
 217         comp = self._create_comp(MySource)
 218
 219     def test_user_flt_output_ports_len(self):
 220         class MyIter(bt2._UserNotificationIterator):
 221             def __next__(self):
 222                 raise bt2.Stop
 223
 224         class MyFilter(bt2._UserFilterComponent,
 225                        notification_iterator_class=MyIter):
 226             def __init__(comp_self, params):
 227                 comp_self._add_output_port('clear')
 228                 comp_self._add_output_port('print')
 229                 comp_self._add_output_port('insert')
 230                 self.assertEqual(len(comp_self._output_ports), 3)
 231
 232         comp = self._create_comp(MyFilter)
 233
 234     def test_user_flt_input_ports_len(self):
 235         class MyIter(bt2._UserNotificationIterator):
 236             def __next__(self):
 237                 raise bt2.Stop
 238
 239         class MyFilter(bt2._UserFilterComponent,
 240                        notification_iterator_class=MyIter):
 241             def __init__(comp_self, params):
 242                 comp_self._add_input_port('clear')
 243                 comp_self._add_input_port('print')
 244                 comp_self._add_input_port('insert')
 245                 self.assertEqual(len(comp_self._input_ports), 3)
 246
 247         comp = self._create_comp(MyFilter)
 248
 249     def test_user_sink_input_ports_len(self):
 250         class MySink(bt2._UserSinkComponent):
 251             def __init__(comp_self, params):
 252                 comp_self._add_input_port('clear')
 253                 comp_self._add_input_port('print')
 254                 comp_self._add_input_port('insert')
 255                 self.assertEqual(len(comp_self._input_ports), 3)
 256
 257             def _consume(self):
 258                 pass
 259
 260         comp = self._create_comp(MySink)
 261
 262     def test_user_src_output_ports_iter(self):
 263         class MyIter(bt2._UserNotificationIterator):
 264             def __next__(self):
 265                 raise bt2.Stop
 266
 267         class MySource(bt2._UserSourceComponent,
 268                        notification_iterator_class=MyIter):
 269             def __init__(comp_self, params):
 270                 port1 = comp_self._add_output_port('clear')
 271                 port2 = comp_self._add_output_port('print')
 272                 port3 = comp_self._add_output_port('insert')
 273                 ports = []
 274
 275                 for port_name, port in comp_self._output_ports.items():
 276                     ports.append((port_name, port))
 277
 278                 self.assertEqual(ports[0][0], 'clear')
 279                 self.assertEqual(ports[0][1].addr, port1.addr)
 280                 self.assertEqual(ports[1][0], 'print')
 281                 self.assertEqual(ports[1][1].addr, port2.addr)
 282                 self.assertEqual(ports[2][0], 'insert')
 283                 self.assertEqual(ports[2][1].addr, port3.addr)
 284
 285         comp = self._create_comp(MySource)
 286
 287     def test_user_flt_output_ports_iter(self):
 288         class MyIter(bt2._UserNotificationIterator):
 289             def __next__(self):
 290                 raise bt2.Stop
 291
 292         class MyFilter(bt2._UserFilterComponent,
 293                        notification_iterator_class=MyIter):
 294             def __init__(comp_self, params):
 295                 port1 = comp_self._add_output_port('clear')
 296                 port2 = comp_self._add_output_port('print')
 297                 port3 = comp_self._add_output_port('insert')
 298                 ports = []
 299
 300                 for port_name, port in comp_self._output_ports.items():
 301                     ports.append((port_name, port))
 302
 303                 self.assertEqual(ports[0][0], 'clear')
 304                 self.assertEqual(ports[0][1].addr, port1.addr)
 305                 self.assertEqual(ports[1][0], 'print')
 306                 self.assertEqual(ports[1][1].addr, port2.addr)
 307                 self.assertEqual(ports[2][0], 'insert')
 308                 self.assertEqual(ports[2][1].addr, port3.addr)
 309
 310         comp = self._create_comp(MyFilter)
 311
 312     def test_user_flt_input_ports_iter(self):
 313         class MyIter(bt2._UserNotificationIterator):
 314             def __next__(self):
 315                 raise bt2.Stop
 316
 317         class MyFilter(bt2._UserFilterComponent,
 318                        notification_iterator_class=MyIter):
 319             def __init__(comp_self, params):
 320                 port1 = comp_self._add_input_port('clear')
 321                 port2 = comp_self._add_input_port('print')
 322                 port3 = comp_self._add_input_port('insert')
 323                 ports = []
 324
 325                 for port_name, port in comp_self._input_ports.items():
 326                     ports.append((port_name, port))
 327
 328                 self.assertEqual(ports[0][0], 'clear')
 329                 self.assertEqual(ports[0][1].addr, port1.addr)
 330                 self.assertEqual(ports[1][0], 'print')
 331                 self.assertEqual(ports[1][1].addr, port2.addr)
 332                 self.assertEqual(ports[2][0], 'insert')
 333                 self.assertEqual(ports[2][1].addr, port3.addr)
 334
 335         comp = self._create_comp(MyFilter)
 336
 337     def test_user_sink_input_ports_iter(self):
 338         class MySink(bt2._UserSinkComponent):
 339             def __init__(comp_self, params):
 340                 port1 = comp_self._add_input_port('clear')
 341                 port2 = comp_self._add_input_port('print')
 342                 port3 = comp_self._add_input_port('insert')
 343                 ports = []
 344
 345                 for port_name, port in comp_self._input_ports.items():
 346                     ports.append((port_name, port))
 347
 348                 self.assertEqual(ports[0][0], 'clear')
 349                 self.assertEqual(ports[0][1].addr, port1.addr)
 350                 self.assertEqual(ports[1][0], 'print')
 351                 self.assertEqual(ports[1][1].addr, port2.addr)
 352                 self.assertEqual(ports[2][0], 'insert')
 353                 self.assertEqual(ports[2][1].addr, port3.addr)
 354
 355             def _consume(self):
 356                 pass
 357
 358         comp = self._create_comp(MySink)
 359
 360     def test_gen_src_output_ports_getitem(self):
 361         class MyIter(bt2._UserNotificationIterator):
 362             def __next__(self):
 363                 raise bt2.Stop
 364
 365         port1 = None
 366         port2 = None
 367         port3 = None
 368
 369         class MySource(bt2._UserSourceComponent,
 370                        notification_iterator_class=MyIter):
 371             def __init__(comp_self, params):
 372                 nonlocal port1, port2, port3
 373                 port1 = comp_self._add_output_port('clear')
 374                 port2 = comp_self._add_output_port('print')
 375                 port3 = comp_self._add_output_port('insert')
 376
 377         comp = self._create_comp(MySource)
 378         self.assertEqual(port3.addr, comp.output_ports['insert'].addr)
 379         self.assertEqual(port2.addr, comp.output_ports['print'].addr)
 380         self.assertEqual(port1.addr, comp.output_ports['clear'].addr)
 381         del port1
 382         del port2
 383         del port3
 384
 385     def test_gen_flt_output_ports_getitem(self):
 386         class MyIter(bt2._UserNotificationIterator):
 387             def __next__(self):
 388                 raise bt2.Stop
 389
 390         port1 = None
 391         port2 = None
 392         port3 = None
 393
 394         class MyFilter(bt2._UserFilterComponent,
 395                        notification_iterator_class=MyIter):
 396             def __init__(comp_self, params):
 397                 nonlocal port1, port2, port3
 398                 port1 = comp_self._add_output_port('clear')
 399                 port2 = comp_self._add_output_port('print')
 400                 port3 = comp_self._add_output_port('insert')
 401
 402         comp = self._create_comp(MyFilter)
 403         self.assertEqual(port3.addr, comp.output_ports['insert'].addr)
 404         self.assertEqual(port2.addr, comp.output_ports['print'].addr)
 405         self.assertEqual(port1.addr, comp.output_ports['clear'].addr)
 406         del port1
 407         del port2
 408         del port3
 409
 410     def test_gen_flt_input_ports_getitem(self):
 411         class MyIter(bt2._UserNotificationIterator):
 412             def __next__(self):
 413                 raise bt2.Stop
 414
 415         port1 = None
 416         port2 = None
 417         port3 = None
 418
 419         class MyFilter(bt2._UserFilterComponent,
 420                        notification_iterator_class=MyIter):
 421             def __init__(comp_self, params):
 422                 nonlocal port1, port2, port3
 423                 port1 = comp_self._add_input_port('clear')
 424                 port2 = comp_self._add_input_port('print')
 425                 port3 = comp_self._add_input_port('insert')
 426
 427         comp = self._create_comp(MyFilter)
 428         self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
 429         self.assertEqual(port2.addr, comp.input_ports['print'].addr)
 430         self.assertEqual(port1.addr, comp.input_ports['clear'].addr)
 431         del port1
 432         del port2
 433         del port3
 434
 435     def test_gen_sink_input_ports_getitem(self):
 436         port1 = None
 437         port2 = None
 438         port3 = None
 439
 440         class MySink(bt2._UserSinkComponent):
 441             def __init__(comp_self, params):
 442                 nonlocal port1, port2, port3
 443                 port1 = comp_self._add_input_port('clear')
 444                 port2 = comp_self._add_input_port('print')
 445                 port3 = comp_self._add_input_port('insert')
 446
 447             def _consume(self):
 448                 pass
 449
 450         comp = self._create_comp(MySink)
 451         self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
 452         self.assertEqual(port2.addr, comp.input_ports['print'].addr)
 453         self.assertEqual(port1.addr, comp.input_ports['clear'].addr)
 454         del port1
 455         del port2
 456         del port3
 457
 458     def test_gen_src_output_ports_getitem_invalid_key(self):
 459         class MyIter(bt2._UserNotificationIterator):
 460             def __next__(self):
 461                 raise bt2.Stop
 462
 463         class MySource(bt2._UserSourceComponent,
 464                        notification_iterator_class=MyIter):
 465             def __init__(comp_self, params):
 466                 comp_self._add_output_port('clear')
 467                 comp_self._add_output_port('print')
 468                 comp_self._add_output_port('insert')
 469
 470         comp = self._create_comp(MySource)
 471
 472         with self.assertRaises(KeyError):
 473             comp.output_ports['hello']
 474
 475     def test_gen_flt_output_ports_getitem_invalid_key(self):
 476         class MyIter(bt2._UserNotificationIterator):
 477             def __next__(self):
 478                 raise bt2.Stop
 479
 480         class MyFilter(bt2._UserFilterComponent,
 481                        notification_iterator_class=MyIter):
 482             def __init__(comp_self, params):
 483                 comp_self._add_output_port('clear')
 484                 comp_self._add_output_port('print')
 485                 comp_self._add_output_port('insert')
 486
 487         comp = self._create_comp(MyFilter)
 488
 489         with self.assertRaises(KeyError):
 490             comp.output_ports['hello']
 491
 492     def test_gen_flt_input_ports_getitem_invalid_key(self):
 493         class MyIter(bt2._UserNotificationIterator):
 494             def __next__(self):
 495                 raise bt2.Stop
 496
 497         class MyFilter(bt2._UserFilterComponent,
 498                        notification_iterator_class=MyIter):
 499             def __init__(comp_self, params):
 500                 comp_self._add_input_port('clear')
 501                 comp_self._add_input_port('print')
 502                 comp_self._add_input_port('insert')
 503
 504         comp = self._create_comp(MyFilter)
 505
 506         with self.assertRaises(KeyError):
 507             comp.input_ports['hello']
 508
 509     def test_gen_sink_input_ports_getitem_invalid_key(self):
 510         class MySink(bt2._UserSinkComponent):
 511             def __init__(comp_self, params):
 512                 comp_self._add_input_port('clear')
 513                 comp_self._add_input_port('print')
 514                 comp_self._add_input_port('insert')
 515
 516                 with self.assertRaises(KeyError):
 517                     comp_self._input_ports['hello']
 518
 519             def _consume(self):
 520                 pass
 521
 522         comp = self._create_comp(MySink)
 523
 524         with self.assertRaises(KeyError):
 525             comp.input_ports['hello']
 526
 527     def test_gen_src_output_ports_len(self):
 528         class MyIter(bt2._UserNotificationIterator):
 529             def __next__(self):
 530                 raise bt2.Stop
 531
 532         class MySource(bt2._UserSourceComponent,
 533                        notification_iterator_class=MyIter):
 534             def __init__(comp_self, params):
 535                 comp_self._add_output_port('clear')
 536                 comp_self._add_output_port('print')
 537                 comp_self._add_output_port('insert')
 538
 539         comp = self._create_comp(MySource)
 540         self.assertEqual(len(comp.output_ports), 3)
 541
 542     def test_gen_flt_output_ports_len(self):
 543         class MyIter(bt2._UserNotificationIterator):
 544             def __next__(self):
 545                 raise bt2.Stop
 546
 547         class MyFilter(bt2._UserFilterComponent,
 548                        notification_iterator_class=MyIter):
 549             def __init__(comp_self, params):
 550                 comp_self._add_output_port('clear')
 551                 comp_self._add_output_port('print')
 552                 comp_self._add_output_port('insert')
 553
 554         comp = self._create_comp(MyFilter)
 555         self.assertEqual(len(comp.output_ports), 3)
 556
 557     def test_gen_flt_input_ports_len(self):
 558         class MyIter(bt2._UserNotificationIterator):
 559             def __next__(self):
 560                 raise bt2.Stop
 561
 562         class MyFilter(bt2._UserFilterComponent,
 563                        notification_iterator_class=MyIter):
 564             def __init__(comp_self, params):
 565                 comp_self._add_input_port('clear')
 566                 comp_self._add_input_port('print')
 567                 comp_self._add_input_port('insert')
 568
 569         comp = self._create_comp(MyFilter)
 570         self.assertEqual(len(comp.input_ports), 3)
 571
 572     def test_gen_sink_input_ports_len(self):
 573         class MySink(bt2._UserSinkComponent):
 574             def __init__(comp_self, params):
 575                 comp_self._add_input_port('clear')
 576                 comp_self._add_input_port('print')
 577                 comp_self._add_input_port('insert')
 578
 579             def _consume(self):
 580                 pass
 581
 582         comp = self._create_comp(MySink)
 583         self.assertEqual(len(comp.input_ports), 3)
 584
 585     def test_gen_src_output_ports_iter(self):
 586         class MyIter(bt2._UserNotificationIterator):
 587             def __next__(self):
 588                 raise bt2.Stop
 589
 590         port1 = None
 591         port2 = None
 592         port3 = None
 593
 594         class MySource(bt2._UserSourceComponent,
 595                        notification_iterator_class=MyIter):
 596             def __init__(comp_self, params):
 597                 nonlocal port1, port2, port3
 598                 port1 = comp_self._add_output_port('clear')
 599                 port2 = comp_self._add_output_port('print')
 600                 port3 = comp_self._add_output_port('insert')
 601
 602         comp = self._create_comp(MySource)
 603         ports = []
 604
 605         for port_name, port in comp.output_ports.items():
 606             ports.append((port_name, port))
 607
 608         self.assertEqual(ports[0][0], 'clear')
 609         self.assertEqual(ports[0][1].addr, port1.addr)
 610         self.assertEqual(ports[1][0], 'print')
 611         self.assertEqual(ports[1][1].addr, port2.addr)
 612         self.assertEqual(ports[2][0], 'insert')
 613         self.assertEqual(ports[2][1].addr, port3.addr)
 614         del port1
 615         del port2
 616         del port3
 617
 618     def test_gen_flt_output_ports_iter(self):
 619         class MyIter(bt2._UserNotificationIterator):
 620             def __next__(self):
 621                 raise bt2.Stop
 622
 623         port1 = None
 624         port2 = None
 625         port3 = None
 626
 627         class MyFilter(bt2._UserFilterComponent,
 628                        notification_iterator_class=MyIter):
 629             def __init__(comp_self, params):
 630                 nonlocal port1, port2, port3
 631                 port1 = comp_self._add_output_port('clear')
 632                 port2 = comp_self._add_output_port('print')
 633                 port3 = comp_self._add_output_port('insert')
 634
 635         comp = self._create_comp(MyFilter)
 636         ports = []
 637
 638         for port_name, port in comp.output_ports.items():
 639             ports.append((port_name, port))
 640
 641         self.assertEqual(ports[0][0], 'clear')
 642         self.assertEqual(ports[0][1].addr, port1.addr)
 643         self.assertEqual(ports[1][0], 'print')
 644         self.assertEqual(ports[1][1].addr, port2.addr)
 645         self.assertEqual(ports[2][0], 'insert')
 646         self.assertEqual(ports[2][1].addr, port3.addr)
 647         del port1
 648         del port2
 649         del port3
 650
 651     def test_gen_flt_input_ports_iter(self):
 652         class MyIter(bt2._UserNotificationIterator):
 653             def __next__(self):
 654                 raise bt2.Stop
 655
 656         port1 = None
 657         port2 = None
 658         port3 = None
 659
 660         class MyFilter(bt2._UserFilterComponent,
 661                        notification_iterator_class=MyIter):
 662             def __init__(comp_self, params):
 663                 nonlocal port1, port2, port3
 664                 port1 = comp_self._add_input_port('clear')
 665                 port2 = comp_self._add_input_port('print')
 666                 port3 = comp_self._add_input_port('insert')
 667
 668         comp = self._create_comp(MyFilter)
 669         ports = []
 670
 671         for port_name, port in comp.input_ports.items():
 672             ports.append((port_name, port))
 673
 674         self.assertEqual(ports[0][0], 'clear')
 675         self.assertEqual(ports[0][1].addr, port1.addr)
 676         self.assertEqual(ports[1][0], 'print')
 677         self.assertEqual(ports[1][1].addr, port2.addr)
 678         self.assertEqual(ports[2][0], 'insert')
 679         self.assertEqual(ports[2][1].addr, port3.addr)
 680         del port1
 681         del port2
 682         del port3
 683
 684     def test_gen_sink_input_ports_iter(self):
 685         port1 = None
 686         port2 = None
 687         port3 = None
 688
 689         class MySink(bt2._UserSinkComponent):
 690             def __init__(comp_self, params):
 691                 nonlocal port1, port2, port3
 692                 port1 = comp_self._add_input_port('clear')
 693                 port2 = comp_self._add_input_port('print')
 694                 port3 = comp_self._add_input_port('insert')
 695
 696             def _consume(self):
 697                 pass
 698
 699         comp = self._create_comp(MySink)
 700         ports = []
 701
 702         for port_name, port in comp.input_ports.items():
 703             ports.append((port_name, port))
 704
 705         self.assertEqual(ports[0][0], 'clear')
 706         self.assertEqual(ports[0][1].addr, port1.addr)
 707         self.assertEqual(ports[1][0], 'print')
 708         self.assertEqual(ports[1][1].addr, port2.addr)
 709         self.assertEqual(ports[2][0], 'insert')
 710         self.assertEqual(ports[2][1].addr, port3.addr)
 711         del port1
 712         del port2
 713         del port3
 714
 715     def test_name(self):
 716         class MySink(bt2._UserSinkComponent):
 717             def __init__(comp_self, params):
 718                 comp_self._add_input_port('clear')
 719
 720             def _consume(self):
 721                 pass
 722
 723         comp = self._create_comp(MySink)
 724         self.assertEqual(comp.input_ports['clear'].name, 'clear')
 725
 726     def test_component(self):
 727         class MySink(bt2._UserSinkComponent):
 728             def __init__(comp_self, params):
 729                 comp_self._add_input_port('clear')
 730
 731             def _consume(self):
 732                 pass
 733
 734         comp = self._create_comp(MySink)
 735         self.assertEqual(comp.input_ports['clear'].component.addr, comp.addr)
 736
 737     def test_connection_none(self):
 738         class MySink(bt2._UserSinkComponent):
 739             def __init__(comp_self, params):
 740                 comp_self._add_input_port('clear')
 741
 742             def _consume(self):
 743                 pass
 744
 745         comp = self._create_comp(MySink)
 746         self.assertIsNone(comp.input_ports['clear'].connection)
 747
 748     def test_is_connected_false(self):
 749         class MySink(bt2._UserSinkComponent):
 750             def __init__(comp_self, params):
 751                 comp_self._add_input_port('clear')
 752
 753             def _consume(self):
 754                 pass
 755
 756         comp = self._create_comp(MySink)
 757         self.assertFalse(comp.input_ports['clear'].is_connected)
 758
 759     def test_eq(self):
 760         class MySink(bt2._UserSinkComponent):
 761             def __init__(comp_self, params):
 762                 comp_self._add_input_port('clear')
 763
 764             def _consume(self):
 765                 pass
 766
 767         comp = self._create_comp(MySink)
 768         self.assertEqual(comp.input_ports['clear'],
 769                          comp.input_ports['clear'])
 770
 771     def test_eq_invalid(self):
 772         class MySink(bt2._UserSinkComponent):
 773             def __init__(comp_self, params):
 774                 comp_self._add_input_port('clear')
 775
 776             def _consume(self):
 777                 pass
 778
 779         comp = self._create_comp(MySink)
 780         self.assertNotEqual(comp.input_ports['clear'], 23)
 781
 782     def test_disconnect_no_connection(self):
 783         class MySink(bt2._UserSinkComponent):
 784             def __init__(comp_self, params):
 785                 port = comp_self._add_input_port('clear')
 786
 787             def _consume(self):
 788                 pass
 789
 790         comp = self._create_comp(MySink)
 791         comp.input_ports['clear'].disconnect()
 792
 793     def test_priv_name(self):
 794         class MySink(bt2._UserSinkComponent):
 795             def __init__(comp_self, params):
 796                 port = comp_self._add_input_port('clear')
 797                 self.assertEqual(port.name, 'clear')
 798
 799             def _consume(self):
 800                 pass
 801
 802         comp = self._create_comp(MySink)
 803
 804     def test_priv_component(self):
 805         class MySink(bt2._UserSinkComponent):
 806             def __init__(comp_self, params):
 807                 port = comp_self._add_input_port('clear')
 808                 self.assertEqual(port.component, comp_self)
 809
 810             def _consume(self):
 811                 pass
 812
 813         comp = self._create_comp(MySink)
 814
 815     def test_priv_connection_none(self):
 816         class MySink(bt2._UserSinkComponent):
 817             def __init__(comp_self, params):
 818                 port = comp_self._add_input_port('clear')
 819                 self.assertIsNone(port.connection)
 820
 821             def _consume(self):
 822                 pass
 823
 824         comp = self._create_comp(MySink)
 825
 826     def test_priv_is_connected_false(self):
 827         class MySink(bt2._UserSinkComponent):
 828             def __init__(comp_self, params):
 829                 port = comp_self._add_input_port('clear')
 830                 self.assertFalse(port.is_connected)
 831
 832             def _consume(self):
 833                 pass
 834
 835         comp = self._create_comp(MySink)
 836
 837     def test_priv_eq(self):
 838         class MySink(bt2._UserSinkComponent):
 839             def __init__(comp_self, params):
 840                 port = comp_self._add_input_port('clear')
 841                 self.assertEqual(port, port)
 842
 843             def _consume(self):
 844                 pass
 845
 846         comp = self._create_comp(MySink)
 847
 848     def test_priv_eq_invalid(self):
 849         class MySink(bt2._UserSinkComponent):
 850             def __init__(comp_self, params):
 851                 port = comp_self._add_input_port('clear')
 852                 self.assertNotEqual(port, 23)
 853
 854             def _consume(self):
 855                 pass
 856
 857         comp = self._create_comp(MySink)
 858
 859     def test_priv_disconnect_no_connection(self):
 860         class MySink(bt2._UserSinkComponent):
 861             def __init__(comp_self, params):
 862                 port = comp_self._add_input_port('clear')
 863                 port.disconnect()
 864
 865             def _consume(self):
 866                 pass
 867
 868         comp = self._create_comp(MySink)
 869
 870     def test_priv_remove_from_component(self):
 871         class MySink(bt2._UserSinkComponent):
 872             def __init__(comp_self, params):
 873                 port = comp_self._add_input_port('clear')
 874                 self.assertEqual(len(comp_self._input_ports), 1)
 875
 876                 try:
 877                     port.remove_from_component()
 878                 except:
 879                     import traceback
 880                     traceback.print_exc()
 881
 882                 self.assertEqual(len(comp_self._input_ports), 0)
 883                 self.assertIsNone(port.component)
 884
 885             def _consume(self):
 886                 pass
 887
 888         comp = self._create_comp(MySink)