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