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