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