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