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