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