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