tests/bindings/python/bt2/test_port.py

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