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