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