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