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_cancel(self):
 200         self.assertFalse(self._graph.is_canceled)
 201         self._graph.cancel()
 202         self.assertTrue(self._graph.is_canceled)
 203
 204     # Test that Graph.run() raises bt2.Canceled if the graph gets canceled
 205     # during execution.
 206     def test_cancel_while_running(self):
 207         class MyIter(_MyIter):
 208             def __next__(self):
 209                 return self._create_stream_beginning_message(self._stream)
 210
 211         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 212             def __init__(self, params):
 213                 self._add_output_port('out')
 214
 215         class MySink(bt2._UserSinkComponent):
 216             def __init__(self, params):
 217                 self._add_input_port('in')
 218
 219             def _consume(self):
 220                 # Pretend that somebody asynchronously cancelled the graph.
 221                 nonlocal graph
 222                 graph.cancel()
 223
 224                 return next(self._msg_iter)
 225
 226             def _graph_is_configured(self):
 227                 self._msg_iter = self._create_input_port_message_iterator(
 228                     self._input_ports['in']
 229                 )
 230
 231         graph = bt2.Graph()
 232         up = graph.add_component(MySource, 'down')
 233         down = graph.add_component(MySink, 'up')
 234         graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
 235         with self.assertRaises(bt2.Canceled):
 236             graph.run()
 237
 238     def test_run(self):
 239         class MyIter(_MyIter):
 240             def __next__(self):
 241                 if self._at == 9:
 242                     raise StopIteration
 243
 244                 if self._at == 0:
 245                     msg = self._create_stream_beginning_message(self._stream)
 246                 elif self._at == 1:
 247                     msg = self._create_packet_beginning_message(self._packet)
 248                 elif self._at == 7:
 249                     msg = self._create_packet_end_message(self._packet)
 250                 elif self._at == 8:
 251                     msg = self._create_stream_end_message(self._stream)
 252                 else:
 253                     msg = self._create_event_message(self._ec, self._packet)
 254
 255                 self._at += 1
 256                 return msg
 257
 258         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 259             def __init__(self, params):
 260                 self._add_output_port('out')
 261
 262         class MySink(bt2._UserSinkComponent):
 263             def __init__(self, params):
 264                 self._input_port = self._add_input_port('in')
 265                 self._at = 0
 266
 267             def _consume(comp_self):
 268                 msg = next(comp_self._msg_iter)
 269
 270                 if comp_self._at == 0:
 271                     self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
 272                 elif comp_self._at == 1:
 273                     self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
 274                 elif comp_self._at >= 2 and comp_self._at <= 6:
 275                     self.assertIsInstance(msg, bt2.message._EventMessage)
 276                     self.assertEqual(msg.event.cls.name, 'salut')
 277                 elif comp_self._at == 7:
 278                     self.assertIsInstance(msg, bt2.message._PacketEndMessage)
 279                 elif comp_self._at == 8:
 280                     self.assertIsInstance(msg, bt2.message._StreamEndMessage)
 281
 282                 comp_self._at += 1
 283
 284             def _graph_is_configured(self):
 285                 self._msg_iter = self._create_input_port_message_iterator(
 286                     self._input_port
 287                 )
 288
 289         src = self._graph.add_component(MySource, 'src')
 290         sink = self._graph.add_component(MySink, 'sink')
 291         conn = self._graph.connect_ports(
 292             src.output_ports['out'], sink.input_ports['in']
 293         )
 294         self._graph.run()
 295
 296     def test_run_again(self):
 297         class MyIter(_MyIter):
 298             def __next__(self):
 299                 if self._at == 3:
 300                     raise bt2.TryAgain
 301
 302                 if self._at == 0:
 303                     msg = self._create_stream_beginning_message(self._stream)
 304                 elif self._at == 1:
 305                     msg = self._create_packet_beginning_message(self._packet)
 306                 elif self._at == 2:
 307                     msg = self._create_event_message(self._ec, self._packet)
 308
 309                 self._at += 1
 310                 return msg
 311
 312         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 313             def __init__(self, params):
 314                 self._add_output_port('out')
 315
 316         class MySink(bt2._UserSinkComponent):
 317             def __init__(self, params):
 318                 self._input_port = self._add_input_port('in')
 319                 self._at = 0
 320
 321             def _consume(comp_self):
 322                 msg = next(comp_self._msg_iter)
 323                 if comp_self._at == 0:
 324                     self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
 325                 elif comp_self._at == 1:
 326                     self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
 327                 elif comp_self._at == 2:
 328                     self.assertIsInstance(msg, bt2.message._EventMessage)
 329                     raise bt2.TryAgain
 330                 else:
 331                     pass
 332
 333                 comp_self._at += 1
 334
 335             def _graph_is_configured(self):
 336                 self._msg_iter = self._create_input_port_message_iterator(
 337                     self._input_port
 338                 )
 339
 340         src = self._graph.add_component(MySource, 'src')
 341         sink = self._graph.add_component(MySink, 'sink')
 342         conn = self._graph.connect_ports(
 343             src.output_ports['out'], sink.input_ports['in']
 344         )
 345
 346         with self.assertRaises(bt2.TryAgain):
 347             self._graph.run()
 348
 349     def test_run_error(self):
 350         raised_in_sink = False
 351
 352         class MyIter(_MyIter):
 353             def __next__(self):
 354                 # If this gets called after the sink raised an exception, it is
 355                 # an error.
 356                 nonlocal raised_in_sink
 357                 assert raised_in_sink is False
 358
 359                 if self._at == 0:
 360                     msg = self._create_stream_beginning_message(self._stream)
 361                 elif self._at == 1:
 362                     msg = self._create_packet_beginning_message(self._packet)
 363                 elif self._at == 2 or self._at == 3:
 364                     msg = self._create_event_message(self._ec, self._packet)
 365                 else:
 366                     raise bt2.TryAgain
 367                 self._at += 1
 368                 return msg
 369
 370         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 371             def __init__(self, params):
 372                 self._add_output_port('out')
 373
 374         class MySink(bt2._UserSinkComponent):
 375             def __init__(self, params):
 376                 self._input_port = self._add_input_port('in')
 377                 self._at = 0
 378
 379             def _consume(comp_self):
 380                 msg = next(comp_self._msg_iter)
 381                 if comp_self._at == 0:
 382                     self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
 383                 elif comp_self._at == 1:
 384                     self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
 385                 elif comp_self._at == 2:
 386                     self.assertIsInstance(msg, bt2.message._EventMessage)
 387                 elif comp_self._at == 3:
 388                     nonlocal raised_in_sink
 389                     raised_in_sink = True
 390                     raise RuntimeError('error!')
 391
 392                 comp_self._at += 1
 393
 394             def _graph_is_configured(self):
 395                 self._msg_iter = self._create_input_port_message_iterator(
 396                     self._input_port
 397                 )
 398
 399         src = self._graph.add_component(MySource, 'src')
 400         sink = self._graph.add_component(MySink, 'sink')
 401         conn = self._graph.connect_ports(
 402             src.output_ports['out'], sink.input_ports['in']
 403         )
 404
 405         with self.assertRaises(bt2._Error):
 406             self._graph.run()
 407
 408     def test_listeners(self):
 409         class MyIter(bt2._UserMessageIterator):
 410             def __next__(self):
 411                 raise bt2.Stop
 412
 413         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 414             def __init__(self, params):
 415                 self._add_output_port('out')
 416                 self._add_output_port('zero')
 417
 418         class MySink(bt2._UserSinkComponent):
 419             def __init__(self, params):
 420                 self._add_input_port('in')
 421
 422             def _consume(self):
 423                 raise bt2.Stop
 424
 425             def _graph_is_configured(self):
 426                 pass
 427
 428             def _port_connected(self, port, other_port):
 429                 self._add_input_port('taste')
 430
 431         def port_added_listener(component, port):
 432             nonlocal calls
 433             calls.append((port_added_listener, component, port))
 434
 435         def ports_connected_listener(
 436             upstream_component, upstream_port, downstream_component, downstream_port
 437         ):
 438             nonlocal calls
 439             calls.append(
 440                 (
 441                     ports_connected_listener,
 442                     upstream_component,
 443                     upstream_port,
 444                     downstream_component,
 445                     downstream_port,
 446                 )
 447             )
 448
 449         calls = []
 450         self._graph.add_port_added_listener(port_added_listener)
 451         self._graph.add_ports_connected_listener(ports_connected_listener)
 452         src = self._graph.add_component(MySource, 'src')
 453         sink = self._graph.add_component(MySink, 'sink')
 454         self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
 455
 456         self.assertEqual(len(calls), 5)
 457
 458         self.assertIs(calls[0][0], port_added_listener)
 459         self.assertEqual(calls[0][1].name, 'src')
 460         self.assertEqual(calls[0][2].name, 'out')
 461
 462         self.assertIs(calls[1][0], port_added_listener)
 463         self.assertEqual(calls[1][1].name, 'src')
 464         self.assertEqual(calls[1][2].name, 'zero')
 465
 466         self.assertIs(calls[2][0], port_added_listener)
 467         self.assertEqual(calls[2][1].name, 'sink')
 468         self.assertEqual(calls[2][2].name, 'in')
 469
 470         self.assertIs(calls[3][0], port_added_listener)
 471         self.assertEqual(calls[3][1].name, 'sink')
 472         self.assertEqual(calls[3][2].name, 'taste')
 473
 474         self.assertIs(calls[4][0], ports_connected_listener)
 475         self.assertEqual(calls[4][1].name, 'src')
 476         self.assertEqual(calls[4][2].name, 'out')
 477         self.assertEqual(calls[4][3].name, 'sink')
 478         self.assertEqual(calls[4][4].name, 'in')
 479
 480     def test_invalid_listeners(self):
 481         class MyIter(bt2._UserMessageIterator):
 482             def __next__(self):
 483                 raise bt2.Stop
 484
 485         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 486             def __init__(self, params):
 487                 self._add_output_port('out')
 488                 self._add_output_port('zero')
 489
 490         class MySink(bt2._UserSinkComponent):
 491             def __init__(self, params):
 492                 self._add_input_port('in')
 493
 494             def _consume(self):
 495                 raise bt2.Stop
 496
 497             def _graph_is_configured(self):
 498                 pass
 499
 500             def _port_connected(self, port, other_port):
 501                 self._add_input_port('taste')
 502
 503         with self.assertRaises(TypeError):
 504             self._graph.add_port_added_listener(1234)
 505         with self.assertRaises(TypeError):
 506             self._graph.add_ports_connected_listener(1234)
 507
 508     def test_raise_in_component_init(self):
 509         class MySink(bt2._UserSinkComponent):
 510             def __init__(self, params):
 511                 raise ValueError('oops!')
 512
 513             def _consume(self):
 514                 raise bt2.Stop
 515
 516             def _graph_is_configured(self):
 517                 pass
 518
 519         graph = bt2.Graph()
 520
 521         with self.assertRaises(bt2._Error):
 522             graph.add_component(MySink, 'comp')
 523
 524     def test_raise_in_port_added_listener(self):
 525         class MySink(bt2._UserSinkComponent):
 526             def __init__(self, params):
 527                 self._add_input_port('in')
 528
 529             def _consume(self):
 530                 raise bt2.Stop
 531
 532             def _graph_is_configured(self):
 533                 pass
 534
 535         def port_added_listener(component, port):
 536             raise ValueError('oh noes!')
 537
 538         graph = bt2.Graph()
 539         graph.add_port_added_listener(port_added_listener)
 540
 541         with self.assertRaises(bt2._Error):
 542             graph.add_component(MySink, 'comp')
 543
 544     def test_raise_in_ports_connected_listener(self):
 545         class MyIter(bt2._UserMessageIterator):
 546             def __next__(self):
 547                 raise bt2.Stop
 548
 549         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 550             def __init__(self, params):
 551                 self._add_output_port('out')
 552
 553         class MySink(bt2._UserSinkComponent):
 554             def __init__(self, params):
 555                 self._add_input_port('in')
 556
 557             def _consume(self):
 558                 raise bt2.Stop
 559
 560             def _graph_is_configured(self):
 561                 pass
 562
 563         def ports_connected_listener(
 564             upstream_component, upstream_port, downstream_component, downstream_port
 565         ):
 566             raise ValueError('oh noes!')
 567
 568         graph = bt2.Graph()
 569         graph.add_ports_connected_listener(ports_connected_listener)
 570         up = graph.add_component(MySource, 'down')
 571         down = graph.add_component(MySink, 'up')
 572
 573         with self.assertRaises(bt2._Error):
 574             graph.connect_ports(up.output_ports['out'], down.input_ports['in'])