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):
  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):
  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):
  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):
  75                 port = comp_self._add_input_port('in')
  76                 self.assertEqual(port.name, 'in')
  77
  78             def _consume(self):
  79                 pass
  80
  81             def _graph_is_configured(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):
  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):
 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):
 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):
 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 _consume(self):
 146                 pass
 147
 148             def _graph_is_configured(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, message_iterator_class=MyIter):
 159             def __init__(comp_self, params):
 160                 comp_self._add_output_port('clear')
 161                 comp_self._add_output_port('print')
 162                 comp_self._add_output_port('insert')
 163
 164                 with self.assertRaises(KeyError):
 165                     comp_self._output_ports['hello']
 166
 167         self._create_comp(MySource)
 168
 169     def test_user_flt_output_ports_getitem_invalid_key(self):
 170         class MyIter(bt2._UserMessageIterator):
 171             def __next__(self):
 172                 raise bt2.Stop
 173
 174         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 175             def __init__(comp_self, params):
 176                 comp_self._add_output_port('clear')
 177                 comp_self._add_output_port('print')
 178                 comp_self._add_output_port('insert')
 179
 180                 with self.assertRaises(KeyError):
 181                     comp_self._output_ports['hello']
 182
 183         self._create_comp(MyFilter)
 184
 185     def test_user_flt_input_ports_getitem_invalid_key(self):
 186         class MyIter(bt2._UserMessageIterator):
 187             def __next__(self):
 188                 raise bt2.Stop
 189
 190         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 191             def __init__(comp_self, params):
 192                 comp_self._add_input_port('clear')
 193                 comp_self._add_input_port('print')
 194                 comp_self._add_input_port('insert')
 195
 196                 with self.assertRaises(KeyError):
 197                     comp_self._input_ports['hello']
 198
 199         self._create_comp(MyFilter)
 200
 201     def test_user_sink_input_ports_getitem_invalid_key(self):
 202         class MySink(bt2._UserSinkComponent):
 203             def __init__(comp_self, params):
 204                 comp_self._add_input_port('clear')
 205                 comp_self._add_input_port('print')
 206                 comp_self._add_input_port('insert')
 207
 208                 with self.assertRaises(KeyError):
 209                     comp_self._input_ports['hello']
 210
 211             def _consume(self):
 212                 pass
 213
 214             def _graph_is_configured(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, message_iterator_class=MyIter):
 225             def __init__(comp_self, params):
 226                 comp_self._add_output_port('clear')
 227                 comp_self._add_output_port('print')
 228                 comp_self._add_output_port('insert')
 229                 self.assertEqual(len(comp_self._output_ports), 3)
 230
 231         self._create_comp(MySource)
 232
 233     def test_user_flt_output_ports_len(self):
 234         class MyIter(bt2._UserMessageIterator):
 235             def __next__(self):
 236                 raise bt2.Stop
 237
 238         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 239             def __init__(comp_self, params):
 240                 comp_self._add_output_port('clear')
 241                 comp_self._add_output_port('print')
 242                 comp_self._add_output_port('insert')
 243                 self.assertEqual(len(comp_self._output_ports), 3)
 244
 245         self._create_comp(MyFilter)
 246
 247     def test_user_flt_input_ports_len(self):
 248         class MyIter(bt2._UserMessageIterator):
 249             def __next__(self):
 250                 raise bt2.Stop
 251
 252         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 253             def __init__(comp_self, params):
 254                 comp_self._add_input_port('clear')
 255                 comp_self._add_input_port('print')
 256                 comp_self._add_input_port('insert')
 257                 self.assertEqual(len(comp_self._input_ports), 3)
 258
 259         self._create_comp(MyFilter)
 260
 261     def test_user_sink_input_ports_len(self):
 262         class MySink(bt2._UserSinkComponent):
 263             def __init__(comp_self, params):
 264                 comp_self._add_input_port('clear')
 265                 comp_self._add_input_port('print')
 266                 comp_self._add_input_port('insert')
 267                 self.assertEqual(len(comp_self._input_ports), 3)
 268
 269             def _consume(self):
 270                 pass
 271
 272             def _graph_is_configured(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, message_iterator_class=MyIter):
 283             def __init__(comp_self, params):
 284                 port1 = comp_self._add_output_port('clear')
 285                 port2 = comp_self._add_output_port('print')
 286                 port3 = comp_self._add_output_port('insert')
 287                 ports = []
 288
 289                 for port_name, port in comp_self._output_ports.items():
 290                     ports.append((port_name, port))
 291
 292                 self.assertEqual(ports[0][0], 'clear')
 293                 self.assertEqual(ports[0][1].addr, port1.addr)
 294                 self.assertEqual(ports[1][0], 'print')
 295                 self.assertEqual(ports[1][1].addr, port2.addr)
 296                 self.assertEqual(ports[2][0], 'insert')
 297                 self.assertEqual(ports[2][1].addr, port3.addr)
 298
 299         self._create_comp(MySource)
 300
 301     def test_user_flt_output_ports_iter(self):
 302         class MyIter(bt2._UserMessageIterator):
 303             def __next__(self):
 304                 raise bt2.Stop
 305
 306         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 307             def __init__(comp_self, params):
 308                 port1 = comp_self._add_output_port('clear')
 309                 port2 = comp_self._add_output_port('print')
 310                 port3 = comp_self._add_output_port('insert')
 311                 ports = []
 312
 313                 for port_name, port in comp_self._output_ports.items():
 314                     ports.append((port_name, port))
 315
 316                 self.assertEqual(ports[0][0], 'clear')
 317                 self.assertEqual(ports[0][1].addr, port1.addr)
 318                 self.assertEqual(ports[1][0], 'print')
 319                 self.assertEqual(ports[1][1].addr, port2.addr)
 320                 self.assertEqual(ports[2][0], 'insert')
 321                 self.assertEqual(ports[2][1].addr, port3.addr)
 322
 323         self._create_comp(MyFilter)
 324
 325     def test_user_flt_input_ports_iter(self):
 326         class MyIter(bt2._UserMessageIterator):
 327             def __next__(self):
 328                 raise bt2.Stop
 329
 330         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 331             def __init__(comp_self, params):
 332                 port1 = comp_self._add_input_port('clear')
 333                 port2 = comp_self._add_input_port('print')
 334                 port3 = comp_self._add_input_port('insert')
 335                 ports = []
 336
 337                 for port_name, port in comp_self._input_ports.items():
 338                     ports.append((port_name, port))
 339
 340                 self.assertEqual(ports[0][0], 'clear')
 341                 self.assertEqual(ports[0][1].addr, port1.addr)
 342                 self.assertEqual(ports[1][0], 'print')
 343                 self.assertEqual(ports[1][1].addr, port2.addr)
 344                 self.assertEqual(ports[2][0], 'insert')
 345                 self.assertEqual(ports[2][1].addr, port3.addr)
 346
 347         self._create_comp(MyFilter)
 348
 349     def test_user_sink_input_ports_iter(self):
 350         class MySink(bt2._UserSinkComponent):
 351             def __init__(comp_self, params):
 352                 port1 = comp_self._add_input_port('clear')
 353                 port2 = comp_self._add_input_port('print')
 354                 port3 = comp_self._add_input_port('insert')
 355                 ports = []
 356
 357                 for port_name, port in comp_self._input_ports.items():
 358                     ports.append((port_name, port))
 359
 360                 self.assertEqual(ports[0][0], 'clear')
 361                 self.assertEqual(ports[0][1].addr, port1.addr)
 362                 self.assertEqual(ports[1][0], 'print')
 363                 self.assertEqual(ports[1][1].addr, port2.addr)
 364                 self.assertEqual(ports[2][0], 'insert')
 365                 self.assertEqual(ports[2][1].addr, port3.addr)
 366
 367             def _consume(self):
 368                 pass
 369
 370             def _graph_is_configured(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, message_iterator_class=MyIter):
 385             def __init__(comp_self, params):
 386                 nonlocal port1, port2, port3
 387                 port1 = comp_self._add_output_port('clear')
 388                 port2 = comp_self._add_output_port('print')
 389                 port3 = comp_self._add_output_port('insert')
 390
 391         comp = self._create_comp(MySource)
 392         self.assertEqual(port3.addr, comp.output_ports['insert'].addr)
 393         self.assertEqual(port2.addr, comp.output_ports['print'].addr)
 394         self.assertEqual(port1.addr, comp.output_ports['clear'].addr)
 395         del port1
 396         del port2
 397         del port3
 398
 399     def test_gen_flt_output_ports_getitem(self):
 400         class MyIter(bt2._UserMessageIterator):
 401             def __next__(self):
 402                 raise bt2.Stop
 403
 404         port1 = None
 405         port2 = None
 406         port3 = None
 407
 408         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 409             def __init__(comp_self, params):
 410                 nonlocal port1, port2, port3
 411                 port1 = comp_self._add_output_port('clear')
 412                 port2 = comp_self._add_output_port('print')
 413                 port3 = comp_self._add_output_port('insert')
 414
 415         comp = self._create_comp(MyFilter)
 416         self.assertEqual(port3.addr, comp.output_ports['insert'].addr)
 417         self.assertEqual(port2.addr, comp.output_ports['print'].addr)
 418         self.assertEqual(port1.addr, comp.output_ports['clear'].addr)
 419         del port1
 420         del port2
 421         del port3
 422
 423     def test_gen_flt_input_ports_getitem(self):
 424         class MyIter(bt2._UserMessageIterator):
 425             def __next__(self):
 426                 raise bt2.Stop
 427
 428         port1 = None
 429         port2 = None
 430         port3 = None
 431
 432         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 433             def __init__(comp_self, params):
 434                 nonlocal port1, port2, port3
 435                 port1 = comp_self._add_input_port('clear')
 436                 port2 = comp_self._add_input_port('print')
 437                 port3 = comp_self._add_input_port('insert')
 438
 439         comp = self._create_comp(MyFilter)
 440         self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
 441         self.assertEqual(port2.addr, comp.input_ports['print'].addr)
 442         self.assertEqual(port1.addr, comp.input_ports['clear'].addr)
 443         del port1
 444         del port2
 445         del port3
 446
 447     def test_gen_sink_input_ports_getitem(self):
 448         port1 = None
 449         port2 = None
 450         port3 = None
 451
 452         class MySink(bt2._UserSinkComponent):
 453             def __init__(comp_self, params):
 454                 nonlocal port1, port2, port3
 455                 port1 = comp_self._add_input_port('clear')
 456                 port2 = comp_self._add_input_port('print')
 457                 port3 = comp_self._add_input_port('insert')
 458
 459             def _consume(self):
 460                 pass
 461
 462             def _graph_is_configured(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, message_iterator_class=MyIter):
 479             def __init__(comp_self, params):
 480                 comp_self._add_output_port('clear')
 481                 comp_self._add_output_port('print')
 482                 comp_self._add_output_port('insert')
 483
 484         comp = self._create_comp(MySource)
 485
 486         with self.assertRaises(KeyError):
 487             comp.output_ports['hello']
 488
 489     def test_gen_flt_output_ports_getitem_invalid_key(self):
 490         class MyIter(bt2._UserMessageIterator):
 491             def __next__(self):
 492                 raise bt2.Stop
 493
 494         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 495             def __init__(comp_self, params):
 496                 comp_self._add_output_port('clear')
 497                 comp_self._add_output_port('print')
 498                 comp_self._add_output_port('insert')
 499
 500         comp = self._create_comp(MyFilter)
 501
 502         with self.assertRaises(KeyError):
 503             comp.output_ports['hello']
 504
 505     def test_gen_flt_input_ports_getitem_invalid_key(self):
 506         class MyIter(bt2._UserMessageIterator):
 507             def __next__(self):
 508                 raise bt2.Stop
 509
 510         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 511             def __init__(comp_self, params):
 512                 comp_self._add_input_port('clear')
 513                 comp_self._add_input_port('print')
 514                 comp_self._add_input_port('insert')
 515
 516         comp = self._create_comp(MyFilter)
 517
 518         with self.assertRaises(KeyError):
 519             comp.input_ports['hello']
 520
 521     def test_gen_sink_input_ports_getitem_invalid_key(self):
 522         class MySink(bt2._UserSinkComponent):
 523             def __init__(comp_self, params):
 524                 comp_self._add_input_port('clear')
 525                 comp_self._add_input_port('print')
 526                 comp_self._add_input_port('insert')
 527
 528                 with self.assertRaises(KeyError):
 529                     comp_self._input_ports['hello']
 530
 531             def _consume(self):
 532                 pass
 533
 534             def _graph_is_configured(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, message_iterator_class=MyIter):
 548             def __init__(comp_self, params):
 549                 comp_self._add_output_port('clear')
 550                 comp_self._add_output_port('print')
 551                 comp_self._add_output_port('insert')
 552
 553         comp = self._create_comp(MySource)
 554         self.assertEqual(len(comp.output_ports), 3)
 555
 556     def test_gen_flt_output_ports_len(self):
 557         class MyIter(bt2._UserMessageIterator):
 558             def __next__(self):
 559                 raise bt2.Stop
 560
 561         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 562             def __init__(comp_self, params):
 563                 comp_self._add_output_port('clear')
 564                 comp_self._add_output_port('print')
 565                 comp_self._add_output_port('insert')
 566
 567         comp = self._create_comp(MyFilter)
 568         self.assertEqual(len(comp.output_ports), 3)
 569
 570     def test_gen_flt_input_ports_len(self):
 571         class MyIter(bt2._UserMessageIterator):
 572             def __next__(self):
 573                 raise bt2.Stop
 574
 575         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 576             def __init__(comp_self, params):
 577                 comp_self._add_input_port('clear')
 578                 comp_self._add_input_port('print')
 579                 comp_self._add_input_port('insert')
 580
 581         comp = self._create_comp(MyFilter)
 582         self.assertEqual(len(comp.input_ports), 3)
 583
 584     def test_gen_sink_input_ports_len(self):
 585         class MySink(bt2._UserSinkComponent):
 586             def __init__(comp_self, params):
 587                 comp_self._add_input_port('clear')
 588                 comp_self._add_input_port('print')
 589                 comp_self._add_input_port('insert')
 590
 591             def _consume(self):
 592                 pass
 593
 594             def _graph_is_configured(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, message_iterator_class=MyIter):
 610             def __init__(comp_self, params):
 611                 nonlocal port1, port2, port3
 612                 port1 = comp_self._add_output_port('clear')
 613                 port2 = comp_self._add_output_port('print')
 614                 port3 = comp_self._add_output_port('insert')
 615
 616         comp = self._create_comp(MySource)
 617         ports = []
 618
 619         for port_name, port in comp.output_ports.items():
 620             ports.append((port_name, port))
 621
 622         self.assertEqual(ports[0][0], 'clear')
 623         self.assertEqual(ports[0][1].addr, port1.addr)
 624         self.assertEqual(ports[1][0], 'print')
 625         self.assertEqual(ports[1][1].addr, port2.addr)
 626         self.assertEqual(ports[2][0], 'insert')
 627         self.assertEqual(ports[2][1].addr, port3.addr)
 628         del port1
 629         del port2
 630         del port3
 631
 632     def test_gen_flt_output_ports_iter(self):
 633         class MyIter(bt2._UserMessageIterator):
 634             def __next__(self):
 635                 raise bt2.Stop
 636
 637         port1 = None
 638         port2 = None
 639         port3 = None
 640
 641         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 642             def __init__(comp_self, params):
 643                 nonlocal port1, port2, port3
 644                 port1 = comp_self._add_output_port('clear')
 645                 port2 = comp_self._add_output_port('print')
 646                 port3 = comp_self._add_output_port('insert')
 647
 648         comp = self._create_comp(MyFilter)
 649         ports = []
 650
 651         for port_name, port in comp.output_ports.items():
 652             ports.append((port_name, port))
 653
 654         self.assertEqual(ports[0][0], 'clear')
 655         self.assertEqual(ports[0][1].addr, port1.addr)
 656         self.assertEqual(ports[1][0], 'print')
 657         self.assertEqual(ports[1][1].addr, port2.addr)
 658         self.assertEqual(ports[2][0], 'insert')
 659         self.assertEqual(ports[2][1].addr, port3.addr)
 660         del port1
 661         del port2
 662         del port3
 663
 664     def test_gen_flt_input_ports_iter(self):
 665         class MyIter(bt2._UserMessageIterator):
 666             def __next__(self):
 667                 raise bt2.Stop
 668
 669         port1 = None
 670         port2 = None
 671         port3 = None
 672
 673         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 674             def __init__(comp_self, params):
 675                 nonlocal port1, port2, port3
 676                 port1 = comp_self._add_input_port('clear')
 677                 port2 = comp_self._add_input_port('print')
 678                 port3 = comp_self._add_input_port('insert')
 679
 680         comp = self._create_comp(MyFilter)
 681         ports = []
 682
 683         for port_name, port in comp.input_ports.items():
 684             ports.append((port_name, port))
 685
 686         self.assertEqual(ports[0][0], 'clear')
 687         self.assertEqual(ports[0][1].addr, port1.addr)
 688         self.assertEqual(ports[1][0], 'print')
 689         self.assertEqual(ports[1][1].addr, port2.addr)
 690         self.assertEqual(ports[2][0], 'insert')
 691         self.assertEqual(ports[2][1].addr, port3.addr)
 692         del port1
 693         del port2
 694         del port3
 695
 696     def test_gen_sink_input_ports_iter(self):
 697         port1 = None
 698         port2 = None
 699         port3 = None
 700
 701         class MySink(bt2._UserSinkComponent):
 702             def __init__(comp_self, params):
 703                 nonlocal port1, port2, port3
 704                 port1 = comp_self._add_input_port('clear')
 705                 port2 = comp_self._add_input_port('print')
 706                 port3 = comp_self._add_input_port('insert')
 707
 708             def _consume(self):
 709                 pass
 710
 711             def _graph_is_configured(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             def _graph_is_configured(self):
 739                 pass
 740
 741         comp = self._create_comp(MySink)
 742         self.assertEqual(comp.input_ports['clear'].name, 'clear')
 743
 744     def test_connection_none(self):
 745         class MySink(bt2._UserSinkComponent):
 746             def __init__(comp_self, params):
 747                 comp_self._add_input_port('clear')
 748
 749             def _consume(self):
 750                 pass
 751
 752             def _graph_is_configured(self):
 753                 pass
 754
 755         comp = self._create_comp(MySink)
 756         self.assertIsNone(comp.input_ports['clear'].connection)
 757
 758     def test_is_connected_false(self):
 759         class MySink(bt2._UserSinkComponent):
 760             def __init__(comp_self, params):
 761                 comp_self._add_input_port('clear')
 762
 763             def _consume(self):
 764                 pass
 765
 766             def _graph_is_configured(self):
 767                 pass
 768
 769         comp = self._create_comp(MySink)
 770         self.assertFalse(comp.input_ports['clear'].is_connected)
 771
 772     def test_self_name(self):
 773         class MySink(bt2._UserSinkComponent):
 774             def __init__(comp_self, params):
 775                 port = comp_self._add_input_port('clear')
 776                 self.assertEqual(port.name, 'clear')
 777
 778             def _consume(self):
 779                 pass
 780
 781             def _graph_is_configured(self):
 782                 pass
 783
 784         self._create_comp(MySink)
 785
 786     def test_self_connection_none(self):
 787         class MySink(bt2._UserSinkComponent):
 788             def __init__(comp_self, params):
 789                 port = comp_self._add_input_port('clear')
 790                 self.assertIsNone(port.connection)
 791
 792             def _consume(self):
 793                 pass
 794
 795             def _graph_is_configured(self):
 796                 pass
 797
 798         self._create_comp(MySink)
 799
 800     def test_self_is_connected_false(self):
 801         class MySink(bt2._UserSinkComponent):
 802             def __init__(comp_self, params):
 803                 port = comp_self._add_input_port('clear')
 804                 self.assertFalse(port.is_connected)
 805
 806             def _consume(self):
 807                 pass
 808
 809             def _graph_is_configured(self):
 810                 pass
 811
 812         self._create_comp(MySink)
 813
 814     def test_source_self_port_user_data(self):
 815         class MyIter(bt2._UserMessageIterator):
 816             def __next__(self):
 817                 raise bt2.Stop
 818
 819         class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter):
 820             def __init__(comp_self, params):
 821                 nonlocal user_datas
 822
 823                 p = comp_self._add_output_port('port1')
 824                 user_datas.append(p.user_data)
 825                 p = comp_self._add_output_port('port2', 2)
 826                 user_datas.append(p.user_data)
 827
 828         user_datas = []
 829
 830         comp = self._create_comp(MySource)
 831         self.assertEqual(user_datas, [None, 2])
 832
 833     def test_filter_self_port_user_data(self):
 834         class MyIter(bt2._UserMessageIterator):
 835             def __next__(self):
 836                 raise bt2.Stop
 837
 838         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
 839             def __init__(comp_self, params):
 840                 nonlocal user_datas
 841
 842                 p = comp_self._add_output_port('port1')
 843                 user_datas.append(p.user_data)
 844                 p = comp_self._add_output_port('port2', 'user data string')
 845                 user_datas.append(p.user_data)
 846
 847                 p = comp_self._add_input_port('port3')
 848                 user_datas.append(p.user_data)
 849                 p = comp_self._add_input_port('port4', user_data={'user data': 'dict'})
 850                 user_datas.append(p.user_data)
 851
 852         user_datas = []
 853
 854         comp = self._create_comp(MyFilter)
 855         self.assertEqual(
 856             user_datas, [None, 'user data string', None, {'user data': 'dict'}]
 857         )
 858
 859     def test_sink_self_port_user_data(self):
 860         class MyIter(bt2._UserMessageIterator):
 861             def __next__(self):
 862                 raise bt2.Stop
 863
 864         class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter):
 865             def __init__(comp_self, params):
 866                 nonlocal user_datas
 867
 868                 p = comp_self._add_input_port('port1')
 869                 user_datas.append(p.user_data)
 870                 p = comp_self._add_input_port('port2', set())
 871                 user_datas.append(p.user_data)
 872
 873         user_datas = []
 874
 875         comp = self._create_comp(MySink)
 876         self.assertEqual(user_datas, [None, set()])