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