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