tests/bindings/python/bt2/test_graph.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; only version 2
   7 # of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  17 #
  18
  19 import unittest
  20 import bt2
  21
  22
  23 class _MyIter(bt2._UserMessageIterator):
  24     def __init__(self, config, self_output_port):
  25         self._build_meta()
  26         self._at = 0
  27
  28     def _build_meta(self):
  29         self._tc = self._component._create_trace_class()
  30         self._t = self._tc()
  31         self._sc = self._tc.create_stream_class(supports_packets=True)
  32         self._ec = self._sc.create_event_class(name='salut')
  33         self._my_int_ft = self._tc.create_signed_integer_field_class(32)
  34         payload_ft = self._tc.create_structure_field_class()
  35         payload_ft += [('my_int', self._my_int_ft)]
  36         self._ec.payload_field_type = payload_ft
  37         self._stream = self._t.create_stream(self._sc)
  38         self._packet = self._stream.create_packet()
  39
  40     def _create_event(self, value):
  41         ev = self._ec()
  42         ev.payload_field['my_int'] = value
  43         ev.packet = self._packet
  44         return ev
  45
  46
  47 class GraphTestCase(unittest.TestCase):
  48     def setUp(self):
  49         self._graph = bt2.Graph()
  50
  51     def tearDown(self):
  52         del self._graph
  53
  54     def test_create_default(self):
  55         bt2.Graph()
  56
  57     def test_create_known_mip_version(self):
  58         bt2.Graph(0)
  59
  60     def test_create_invalid_mip_version_type(self):
  61         with self.assertRaises(TypeError):
  62             bt2.Graph('')
  63
  64     def test_create_unknown_mip_version(self):
  65         with self.assertRaisesRegex(ValueError, 'unknown MIP version'):
  66             bt2.Graph(1)
  67
  68     def test_add_component_user_cls(self):
  69         class MySink(bt2._UserSinkComponent):
  70             def _user_consume(self):
  71                 pass
  72
  73         comp = self._graph.add_component(MySink, 'salut')
  74         self.assertEqual(comp.name, 'salut')
  75
  76     def test_add_component_gen_cls(self):
  77         class MySink(bt2._UserSinkComponent):
  78             def _user_consume(self):
  79                 pass
  80
  81         comp = self._graph.add_component(MySink, 'salut')
  82         assert comp
  83         comp2 = self._graph.add_component(comp.cls, 'salut2')
  84         self.assertEqual(comp2.name, 'salut2')
  85
  86     def test_add_component_params(self):
  87         comp_params = None
  88
  89         class MySink(bt2._UserSinkComponent):
  90             def __init__(self, config, params, obj):
  91                 nonlocal comp_params
  92                 comp_params = params
  93
  94             def _user_consume(self):
  95                 pass
  96
  97         params = {'hello': 23, 'path': '/path/to/stuff'}
  98         self._graph.add_component(MySink, 'salut', params)
  99         self.assertEqual(params, comp_params)
 100         del comp_params
 101
 102     def test_add_component_obj_python_comp_cls(self):
 103         comp_obj = None
 104
 105         class MySink(bt2._UserSinkComponent):
 106             def __init__(self, config, params, obj):
 107                 nonlocal comp_obj
 108                 comp_obj = obj
 109
 110             def _user_consume(self):
 111                 pass
 112
 113         obj = object()
 114         self._graph.add_component(MySink, 'salut', obj=obj)
 115         self.assertIs(comp_obj, obj)
 116         del comp_obj
 117
 118     def test_add_component_obj_none_python_comp_cls(self):
 119         comp_obj = None
 120
 121         class MySink(bt2._UserSinkComponent):
 122             def __init__(self, config, params, obj):
 123                 nonlocal comp_obj
 124                 comp_obj = obj
 125
 126             def _user_consume(self):
 127                 pass
 128
 129         self._graph.add_component(MySink, 'salut')
 130         self.assertIsNone(comp_obj)
 131         del comp_obj
 132
 133     def test_add_component_obj_non_python_comp_cls(self):
 134         plugin = bt2.find_plugin('text', find_in_user_dir=False, find_in_sys_dir=False)
 135         assert plugin is not None
 136         cc = plugin.source_component_classes['dmesg']
 137         assert cc is not None
 138
 139         with self.assertRaises(ValueError):
 140             self._graph.add_component(cc, 'salut', obj=57)
 141
 142     def test_add_component_invalid_cls_type(self):
 143         with self.assertRaises(TypeError):
 144             self._graph.add_component(int, 'salut')
 145
 146     def test_add_component_invalid_logging_level_type(self):
 147         class MySink(bt2._UserSinkComponent):
 148             def _user_consume(self):
 149                 pass
 150
 151         with self.assertRaises(TypeError):
 152             self._graph.add_component(MySink, 'salut', logging_level='yo')
 153
 154     def test_add_component_invalid_logging_level_value(self):
 155         class MySink(bt2._UserSinkComponent):
 156             def _user_consume(self):
 157                 pass
 158
 159         with self.assertRaises(ValueError):
 160             self._graph.add_component(MySink, 'salut', logging_level=12345)
 161
 162     def test_add_component_invalid_params_type(self):
 163         class MySink(bt2._UserSinkComponent):
 164             def _user_consume(self):
 165                 pass
 166
 167         with self.assertRaises(TypeError):
 168             self._graph.add_component(MySink, 'salut', params=12)
 169
 170     def test_add_component_params_dict(self):
 171         params_obj = None
 172
 173         class MySink(bt2._UserSinkComponent):
 174             def __init__(self, config, params, obj):
 175                 nonlocal params_obj
 176                 params_obj = params
 177
 178             def _user_consume(self):
 179                 pass
 180
 181         params = {'plage': 12312}
 182         self._graph.add_component(MySink, 'salut', params=params)
 183
 184         # Check equality and not identity because `add_component()` method
 185         # converts the Python `dict` to a `bt2.MapValue`.
 186         self.assertEqual(params, params_obj)
 187
 188     def test_add_component_params_mapvalue(self):
 189         params_obj = None
 190
 191         class MySink(bt2._UserSinkComponent):
 192             def __init__(self, config, params, obj):
 193                 nonlocal params_obj
 194                 params_obj = params
 195
 196             def _user_consume(self):
 197                 pass
 198
 199         params = bt2.MapValue({'beachclub': '121'})
 200         self._graph.add_component(MySink, 'salut', params=params)
 201
 202         self.assertEqual(params, params_obj)
 203
 204     def test_add_component_logging_level(self):
 205         class MySink(bt2._UserSinkComponent):
 206             def _user_consume(self):
 207                 pass
 208
 209         comp = self._graph.add_component(
 210             MySink, 'salut', logging_level=bt2.LoggingLevel.DEBUG
 211         )
 212         self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)
 213
 214     def test_connect_ports(self):
 215         class MyIter(bt2._UserMessageIterator):
 216             def __next__(self):
 217                 raise bt2.Stop
 218
 219         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 220             def __init__(self, config, params, obj):
 221                 self._add_output_port('out')
 222
 223         class MySink(bt2._UserSinkComponent):
 224             def __init__(self, config, params, obj):
 225                 self._add_input_port('in')
 226
 227             def _user_consume(self):
 228                 raise bt2.Stop
 229
 230         src = self._graph.add_component(MySource, 'src')
 231         sink = self._graph.add_component(MySink, 'sink')
 232
 233         conn = self._graph.connect_ports(
 234             src.output_ports['out'], sink.input_ports['in']
 235         )
 236         self.assertTrue(src.output_ports['out'].is_connected)
 237         self.assertTrue(sink.input_ports['in'].is_connected)
 238         self.assertEqual(src.output_ports['out'].connection.addr, conn.addr)
 239         self.assertEqual(sink.input_ports['in'].connection.addr, conn.addr)
 240
 241     def test_connect_ports_invalid_direction(self):
 242         class MyIter(bt2._UserMessageIterator):
 243             def __next__(self):
 244                 raise bt2.Stop
 245
 246         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 247             def __init__(self, config, params, obj):
 248                 self._add_output_port('out')
 249
 250         class MySink(bt2._UserSinkComponent):
 251             def __init__(self, config, params, obj):
 252                 self._add_input_port('in')
 253
 254             def _user_consume(self):
 255                 raise bt2.Stop
 256
 257         src = self._graph.add_component(MySource, 'src')
 258         sink = self._graph.add_component(MySink, 'sink')
 259
 260         with self.assertRaises(TypeError):
 261             self._graph.connect_ports(sink.input_ports['in'], src.output_ports['out'])
 262
 263     def test_add_interrupter(self):
 264         class MyIter(bt2._UserMessageIterator):
 265             def __next__(self):
 266                 raise TypeError
 267
 268         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 269             def __init__(self, config, params, obj):
 270                 self._add_output_port('out')
 271
 272         class MySink(bt2._UserSinkComponent):
 273             def __init__(self, config, params, obj):
 274                 self._add_input_port('in')
 275
 276             def _user_consume(self):
 277                 next(self._msg_iter)
 278
 279             def _user_graph_is_configured(self):
 280                 self._msg_iter = self._create_input_port_message_iterator(
 281                     self._input_ports['in']
 282                 )
 283
 284         # add two interrupters, set one of them
 285         interrupter1 = bt2.Interrupter()
 286         interrupter2 = bt2.Interrupter()
 287         self._graph.add_interrupter(interrupter1)
 288         src = self._graph.add_component(MySource, 'src')
 289         sink = self._graph.add_component(MySink, 'sink')
 290         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 291         self._graph.add_interrupter(interrupter2)
 292
 293         with self.assertRaises(bt2._Error):
 294             self._graph.run()
 295
 296         interrupter2.set()
 297
 298         with self.assertRaises(bt2.TryAgain):
 299             self._graph.run()
 300
 301         interrupter2.reset()
 302
 303         with self.assertRaises(bt2._Error):
 304             self._graph.run()
 305
 306     # Test that Graph.run() raises bt2.Interrupted if the graph gets
 307     # interrupted during execution.
 308     def test_interrupt_while_running(self):
 309         class MyIter(_MyIter):
 310             def __next__(self):
 311                 return self._create_stream_beginning_message(self._stream)
 312
 313         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 314             def __init__(self, config, params, obj):
 315                 self._add_output_port('out')
 316
 317         class MySink(bt2._UserSinkComponent):
 318             def __init__(self, config, params, obj):
 319                 self._add_input_port('in')
 320
 321             def _user_consume(self):
 322                 # Pretend that somebody asynchronously interrupted the graph.
 323                 nonlocal graph
 324                 graph.interrupt()
 325                 return next(self._msg_iter)
 326
 327             def _user_graph_is_configured(self):
 328                 self._msg_iter = self._create_input_port_message_iterator(
 329                     self._input_ports['in']
 330                 )
 331
 332         graph = self._graph
 333         up = self._graph.add_component(MySource, 'down')
 334         down = self._graph.add_component(MySink, 'up')
 335         self._graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
 336
 337         with self.assertRaises(bt2.TryAgain):
 338             self._graph.run()
 339
 340     def test_run(self):
 341         class MyIter(_MyIter):
 342             def __next__(self):
 343                 if self._at == 9:
 344                     raise StopIteration
 345
 346                 if self._at == 0:
 347                     msg = self._create_stream_beginning_message(self._stream)
 348                 elif self._at == 1:
 349                     msg = self._create_packet_beginning_message(self._packet)
 350                 elif self._at == 7:
 351                     msg = self._create_packet_end_message(self._packet)
 352                 elif self._at == 8:
 353                     msg = self._create_stream_end_message(self._stream)
 354                 else:
 355                     msg = self._create_event_message(self._ec, self._packet)
 356
 357                 self._at += 1
 358                 return msg
 359
 360         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 361             def __init__(self, config, params, obj):
 362                 self._add_output_port('out')
 363
 364         class MySink(bt2._UserSinkComponent):
 365             def __init__(self, config, params, obj):
 366                 self._input_port = self._add_input_port('in')
 367                 self._at = 0
 368
 369             def _user_consume(comp_self):
 370                 msg = next(comp_self._msg_iter)
 371
 372                 if comp_self._at == 0:
 373                     self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 374                 elif comp_self._at == 1:
 375                     self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 376                 elif comp_self._at >= 2 and comp_self._at <= 6:
 377                     self.assertIs(type(msg), bt2._EventMessageConst)
 378                     self.assertEqual(msg.event.cls.name, 'salut')
 379                 elif comp_self._at == 7:
 380                     self.assertIs(type(msg), bt2._PacketEndMessageConst)
 381                 elif comp_self._at == 8:
 382                     self.assertIs(type(msg), bt2._StreamEndMessageConst)
 383
 384                 comp_self._at += 1
 385
 386             def _user_graph_is_configured(self):
 387                 self._msg_iter = self._create_input_port_message_iterator(
 388                     self._input_port
 389                 )
 390
 391         src = self._graph.add_component(MySource, 'src')
 392         sink = self._graph.add_component(MySink, 'sink')
 393         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 394         self._graph.run()
 395
 396     def test_run_once(self):
 397         class MyIter(_MyIter):
 398             pass
 399
 400         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 401             def __init__(self, config, params, obj):
 402                 self._add_output_port('out')
 403
 404         class MySink(bt2._UserSinkComponent):
 405             def __init__(self, config, params, obj):
 406                 self._input_port = self._add_input_port('in')
 407
 408             def _user_consume(comp_self):
 409                 nonlocal run_count
 410                 run_count += 1
 411                 raise bt2.TryAgain
 412
 413         run_count = 0
 414         src = self._graph.add_component(MySource, 'src')
 415         sink = self._graph.add_component(MySink, 'sink')
 416         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 417
 418         with self.assertRaises(bt2.TryAgain):
 419             self._graph.run_once()
 420
 421         self.assertEqual(run_count, 1)
 422
 423     def test_run_once_stops(self):
 424         class MyIter(_MyIter):
 425             pass
 426
 427         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 428             def __init__(self, config, params, obj):
 429                 self._add_output_port('out')
 430
 431         class MySink(bt2._UserSinkComponent):
 432             def __init__(self, config, params, obj):
 433                 self._input_port = self._add_input_port('in')
 434
 435             def _user_consume(comp_self):
 436                 raise bt2.Stop
 437
 438         src = self._graph.add_component(MySource, 'src')
 439         sink = self._graph.add_component(MySink, 'sink')
 440         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 441
 442         with self.assertRaises(bt2.Stop):
 443             self._graph.run_once()
 444
 445     def test_run_again(self):
 446         class MyIter(_MyIter):
 447             def __next__(self):
 448                 if self._at == 3:
 449                     raise bt2.TryAgain
 450
 451                 if self._at == 0:
 452                     msg = self._create_stream_beginning_message(self._stream)
 453                 elif self._at == 1:
 454                     msg = self._create_packet_beginning_message(self._packet)
 455                 elif self._at == 2:
 456                     msg = self._create_event_message(self._ec, self._packet)
 457
 458                 self._at += 1
 459                 return msg
 460
 461         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 462             def __init__(self, config, params, obj):
 463                 self._add_output_port('out')
 464
 465         class MySink(bt2._UserSinkComponent):
 466             def __init__(self, config, params, obj):
 467                 self._input_port = self._add_input_port('in')
 468                 self._at = 0
 469
 470             def _user_consume(comp_self):
 471                 msg = next(comp_self._msg_iter)
 472                 if comp_self._at == 0:
 473                     self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 474                 elif comp_self._at == 1:
 475                     self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 476                 elif comp_self._at == 2:
 477                     self.assertIs(type(msg), bt2._EventMessageConst)
 478                     raise bt2.TryAgain
 479                 else:
 480                     pass
 481
 482                 comp_self._at += 1
 483
 484             def _user_graph_is_configured(self):
 485                 self._msg_iter = self._create_input_port_message_iterator(
 486                     self._input_port
 487                 )
 488
 489         src = self._graph.add_component(MySource, 'src')
 490         sink = self._graph.add_component(MySink, 'sink')
 491         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 492
 493         with self.assertRaises(bt2.TryAgain):
 494             self._graph.run()
 495
 496     def test_run_error(self):
 497         raised_in_sink = False
 498
 499         class MyIter(_MyIter):
 500             def __next__(self):
 501                 # If this gets called after the sink raised an exception, it is
 502                 # an error.
 503                 nonlocal raised_in_sink
 504                 assert raised_in_sink is False
 505
 506                 if self._at == 0:
 507                     msg = self._create_stream_beginning_message(self._stream)
 508                 elif self._at == 1:
 509                     msg = self._create_packet_beginning_message(self._packet)
 510                 elif self._at == 2 or self._at == 3:
 511                     msg = self._create_event_message(self._ec, self._packet)
 512                 else:
 513                     raise bt2.TryAgain
 514                 self._at += 1
 515                 return msg
 516
 517         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 518             def __init__(self, config, params, obj):
 519                 self._add_output_port('out')
 520
 521         class MySink(bt2._UserSinkComponent):
 522             def __init__(self, config, params, obj):
 523                 self._input_port = self._add_input_port('in')
 524                 self._at = 0
 525
 526             def _user_consume(comp_self):
 527                 msg = next(comp_self._msg_iter)
 528                 if comp_self._at == 0:
 529                     self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 530                 elif comp_self._at == 1:
 531                     self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 532                 elif comp_self._at == 2:
 533                     self.assertIs(type(msg), bt2._EventMessageConst)
 534                 elif comp_self._at == 3:
 535                     nonlocal raised_in_sink
 536                     raised_in_sink = True
 537                     raise RuntimeError('error!')
 538
 539                 comp_self._at += 1
 540
 541             def _user_graph_is_configured(self):
 542                 self._msg_iter = self._create_input_port_message_iterator(
 543                     self._input_port
 544                 )
 545
 546         src = self._graph.add_component(MySource, 'src')
 547         sink = self._graph.add_component(MySink, 'sink')
 548         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 549
 550         with self.assertRaises(bt2._Error):
 551             self._graph.run()
 552
 553     def test_listeners(self):
 554         class MyIter(bt2._UserMessageIterator):
 555             def __next__(self):
 556                 raise bt2.Stop
 557
 558         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 559             def __init__(self, config, params, obj):
 560                 self._add_output_port('out')
 561                 self._add_output_port('zero')
 562
 563         class MySink(bt2._UserSinkComponent):
 564             def __init__(self, config, params, obj):
 565                 self._add_input_port('in')
 566
 567             def _user_consume(self):
 568                 raise bt2.Stop
 569
 570             def _user_port_connected(self, port, other_port):
 571                 self._add_input_port('taste')
 572
 573         def port_added_listener(component, port):
 574             nonlocal calls
 575             calls.append((port_added_listener, component, port))
 576
 577         def ports_connected_listener(
 578             upstream_component, upstream_port, downstream_component, downstream_port
 579         ):
 580             nonlocal calls
 581             calls.append(
 582                 (
 583                     ports_connected_listener,
 584                     upstream_component,
 585                     upstream_port,
 586                     downstream_component,
 587                     downstream_port,
 588                 )
 589             )
 590
 591         calls = []
 592         self._graph.add_port_added_listener(port_added_listener)
 593         self._graph.add_ports_connected_listener(ports_connected_listener)
 594         src = self._graph.add_component(MySource, 'src')
 595         sink = self._graph.add_component(MySink, 'sink')
 596         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 597
 598         self.assertEqual(len(calls), 5)
 599
 600         self.assertIs(calls[0][0], port_added_listener)
 601         self.assertEqual(calls[0][1].name, 'src')
 602         self.assertEqual(calls[0][2].name, 'out')
 603
 604         self.assertIs(calls[1][0], port_added_listener)
 605         self.assertEqual(calls[1][1].name, 'src')
 606         self.assertEqual(calls[1][2].name, 'zero')
 607
 608         self.assertIs(calls[2][0], port_added_listener)
 609         self.assertEqual(calls[2][1].name, 'sink')
 610         self.assertEqual(calls[2][2].name, 'in')
 611
 612         self.assertIs(calls[3][0], port_added_listener)
 613         self.assertEqual(calls[3][1].name, 'sink')
 614         self.assertEqual(calls[3][2].name, 'taste')
 615
 616         self.assertIs(calls[4][0], ports_connected_listener)
 617         self.assertEqual(calls[4][1].name, 'src')
 618         self.assertEqual(calls[4][2].name, 'out')
 619         self.assertEqual(calls[4][3].name, 'sink')
 620         self.assertEqual(calls[4][4].name, 'in')
 621
 622     def test_invalid_listeners(self):
 623         class MyIter(bt2._UserMessageIterator):
 624             def __next__(self):
 625                 raise bt2.Stop
 626
 627         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 628             def __init__(self, config, params, obj):
 629                 self._add_output_port('out')
 630                 self._add_output_port('zero')
 631
 632         class MySink(bt2._UserSinkComponent):
 633             def __init__(self, config, params, obj):
 634                 self._add_input_port('in')
 635
 636             def _user_consume(self):
 637                 raise bt2.Stop
 638
 639             def _user_port_connected(self, port, other_port):
 640                 self._add_input_port('taste')
 641
 642         with self.assertRaises(TypeError):
 643             self._graph.add_port_added_listener(1234)
 644         with self.assertRaises(TypeError):
 645             self._graph.add_ports_connected_listener(1234)
 646
 647     def test_raise_in_component_init(self):
 648         class MySink(bt2._UserSinkComponent):
 649             def __init__(self, config, params, obj):
 650                 raise ValueError('oops!')
 651
 652             def _user_consume(self):
 653                 raise bt2.Stop
 654
 655         graph = bt2.Graph()
 656
 657         with self.assertRaises(bt2._Error):
 658             graph.add_component(MySink, 'comp')
 659
 660     def test_raise_in_port_added_listener(self):
 661         class MySink(bt2._UserSinkComponent):
 662             def __init__(self, config, params, obj):
 663                 self._add_input_port('in')
 664
 665             def _user_consume(self):
 666                 raise bt2.Stop
 667
 668         def port_added_listener(component, port):
 669             raise ValueError('oh noes!')
 670
 671         graph = bt2.Graph()
 672         graph.add_port_added_listener(port_added_listener)
 673
 674         with self.assertRaises(bt2._Error):
 675             graph.add_component(MySink, 'comp')
 676
 677     def test_raise_in_ports_connected_listener(self):
 678         class MyIter(bt2._UserMessageIterator):
 679             def __next__(self):
 680                 raise bt2.Stop
 681
 682         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 683             def __init__(self, config, params, obj):
 684                 self._add_output_port('out')
 685
 686         class MySink(bt2._UserSinkComponent):
 687             def __init__(self, config, params, obj):
 688                 self._add_input_port('in')
 689
 690             def _user_consume(self):
 691                 raise bt2.Stop
 692
 693         def ports_connected_listener(
 694             upstream_component, upstream_port, downstream_component, downstream_port
 695         ):
 696             raise ValueError('oh noes!')
 697
 698         graph = bt2.Graph()
 699         graph.add_ports_connected_listener(ports_connected_listener)
 700         up = graph.add_component(MySource, 'down')
 701         down = graph.add_component(MySink, 'up')
 702
 703         with self.assertRaises(bt2._Error):
 704             graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
 705
 706
 707 if __name__ == '__main__':
 708     unittest.main()