tests/bindings/python/bt2/test_port.py

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