tests/bindings/python/bt2/test_message_iterator.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 import sys
  22 from utils import TestOutputPortMessageIterator
  23 from bt2 import port as bt2_port
  24
  25
  26 class SimpleSink(bt2._UserSinkComponent):
  27     # Straightforward sink that creates one input port (`in`) and consumes from
  28     # it.
  29
  30     def __init__(self, params, obj):
  31         self._add_input_port('in')
  32
  33     def _user_consume(self):
  34         next(self._msg_iter)
  35
  36     def _user_graph_is_configured(self):
  37         self._msg_iter = self._create_input_port_message_iterator(
  38             self._input_ports['in']
  39         )
  40
  41
  42 def _create_graph(src_comp_cls, sink_comp_cls, flt_comp_cls=None):
  43     graph = bt2.Graph()
  44
  45     src_comp = graph.add_component(src_comp_cls, 'src')
  46     sink_comp = graph.add_component(sink_comp_cls, 'sink')
  47
  48     if flt_comp_cls is not None:
  49         flt_comp = graph.add_component(flt_comp_cls, 'flt')
  50         graph.connect_ports(src_comp.output_ports['out'], flt_comp.input_ports['in'])
  51         graph.connect_ports(flt_comp.output_ports['out'], sink_comp.input_ports['in'])
  52     else:
  53         graph.connect_ports(src_comp.output_ports['out'], sink_comp.input_ports['in'])
  54
  55     return graph
  56
  57
  58 class UserMessageIteratorTestCase(unittest.TestCase):
  59     def test_init(self):
  60         the_output_port_from_source = None
  61         the_output_port_from_iter = None
  62
  63         class MyIter(bt2._UserMessageIterator):
  64             def __init__(self, self_port_output):
  65                 nonlocal initialized
  66                 nonlocal the_output_port_from_iter
  67                 initialized = True
  68                 the_output_port_from_iter = self_port_output
  69
  70         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
  71             def __init__(self, params, obj):
  72                 nonlocal the_output_port_from_source
  73                 the_output_port_from_source = self._add_output_port('out', 'user data')
  74
  75         initialized = False
  76         graph = _create_graph(MySource, SimpleSink)
  77         graph.run()
  78         self.assertTrue(initialized)
  79         self.assertEqual(
  80             the_output_port_from_source.addr, the_output_port_from_iter.addr
  81         )
  82         self.assertEqual(the_output_port_from_iter.user_data, 'user data')
  83
  84     def test_create_from_message_iterator(self):
  85         class MySourceIter(bt2._UserMessageIterator):
  86             def __init__(self, self_port_output):
  87                 nonlocal src_iter_initialized
  88                 src_iter_initialized = True
  89
  90         class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
  91             def __init__(self, params, obj):
  92                 self._add_output_port('out')
  93
  94         class MyFilterIter(bt2._UserMessageIterator):
  95             def __init__(self, self_port_output):
  96                 nonlocal flt_iter_initialized
  97                 flt_iter_initialized = True
  98                 self._up_iter = self._create_input_port_message_iterator(
  99                     self._component._input_ports['in']
 100                 )
 101
 102             def __next__(self):
 103                 return next(self._up_iter)
 104
 105         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 106             def __init__(self, params, obj):
 107                 self._add_input_port('in')
 108                 self._add_output_port('out')
 109
 110         src_iter_initialized = False
 111         flt_iter_initialized = False
 112         graph = _create_graph(MySource, SimpleSink, MyFilter)
 113         graph.run()
 114         self.assertTrue(src_iter_initialized)
 115         self.assertTrue(flt_iter_initialized)
 116
 117     def test_create_user_error(self):
 118         # This tests both error handling by
 119         # _UserSinkComponent._create_input_port_message_iterator
 120         # and _UserMessageIterator._create_input_port_message_iterator, as they
 121         # are both used in the graph.
 122         class MySourceIter(bt2._UserMessageIterator):
 123             def __init__(self, self_port_output):
 124                 raise ValueError('Very bad error')
 125
 126         class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
 127             def __init__(self, params, obj):
 128                 self._add_output_port('out')
 129
 130         class MyFilterIter(bt2._UserMessageIterator):
 131             def __init__(self, self_port_output):
 132                 # This is expected to raise because of the error in
 133                 # MySourceIter.__init__.
 134                 self._create_input_port_message_iterator(
 135                     self._component._input_ports['in']
 136                 )
 137
 138         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 139             def __init__(self, params, obj):
 140                 self._add_input_port('in')
 141                 self._add_output_port('out')
 142
 143         graph = _create_graph(MySource, SimpleSink, MyFilter)
 144
 145         with self.assertRaises(bt2._Error) as ctx:
 146             graph.run()
 147
 148         exc = ctx.exception
 149         cause = exc[0]
 150
 151         self.assertIsInstance(cause, bt2._MessageIteratorErrorCause)
 152         self.assertEqual(cause.component_name, 'src')
 153         self.assertEqual(cause.component_output_port_name, 'out')
 154         self.assertIn('ValueError: Very bad error', cause.message)
 155
 156     def test_finalize(self):
 157         class MyIter(bt2._UserMessageIterator):
 158             def _user_finalize(self):
 159                 nonlocal finalized
 160                 finalized = True
 161
 162         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 163             def __init__(self, params, obj):
 164                 self._add_output_port('out')
 165
 166         finalized = False
 167         graph = _create_graph(MySource, SimpleSink)
 168         graph.run()
 169         del graph
 170         self.assertTrue(finalized)
 171
 172     def test_component(self):
 173         class MyIter(bt2._UserMessageIterator):
 174             def __init__(self, self_port_output):
 175                 nonlocal salut
 176                 salut = self._component._salut
 177
 178         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 179             def __init__(self, params, obj):
 180                 self._add_output_port('out')
 181                 self._salut = 23
 182
 183         salut = None
 184         graph = _create_graph(MySource, SimpleSink)
 185         graph.run()
 186         self.assertEqual(salut, 23)
 187
 188     def test_port(self):
 189         class MyIter(bt2._UserMessageIterator):
 190             def __init__(self_iter, self_port_output):
 191                 nonlocal called
 192                 called = True
 193                 port = self_iter._port
 194                 self.assertIs(type(self_port_output), bt2_port._UserComponentOutputPort)
 195                 self.assertIs(type(port), bt2_port._UserComponentOutputPort)
 196                 self.assertEqual(self_port_output.addr, port.addr)
 197
 198         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 199             def __init__(self, params, obj):
 200                 self._add_output_port('out')
 201
 202         called = False
 203         graph = _create_graph(MySource, SimpleSink)
 204         graph.run()
 205         self.assertTrue(called)
 206
 207     def test_addr(self):
 208         class MyIter(bt2._UserMessageIterator):
 209             def __init__(self, self_port_output):
 210                 nonlocal addr
 211                 addr = self.addr
 212
 213         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 214             def __init__(self, params, obj):
 215                 self._add_output_port('out')
 216
 217         addr = None
 218         graph = _create_graph(MySource, SimpleSink)
 219         graph.run()
 220         self.assertIsNotNone(addr)
 221         self.assertNotEqual(addr, 0)
 222
 223     # Test that messages returned by _UserMessageIterator.__next__ remain valid
 224     # and can be re-used.
 225     def test_reuse_message(self):
 226         class MyIter(bt2._UserMessageIterator):
 227             def __init__(self, port):
 228                 tc, sc, ec = port.user_data
 229                 trace = tc()
 230                 stream = trace.create_stream(sc)
 231                 packet = stream.create_packet()
 232
 233                 # This message will be returned twice by __next__.
 234                 event_message = self._create_event_message(ec, packet)
 235
 236                 self._msgs = [
 237                     self._create_stream_beginning_message(stream),
 238                     self._create_packet_beginning_message(packet),
 239                     event_message,
 240                     event_message,
 241                 ]
 242
 243             def __next__(self):
 244                 return self._msgs.pop(0)
 245
 246         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 247             def __init__(self, params, obj):
 248                 tc = self._create_trace_class()
 249                 sc = tc.create_stream_class(supports_packets=True)
 250                 ec = sc.create_event_class()
 251                 self._add_output_port('out', (tc, sc, ec))
 252
 253         graph = bt2.Graph()
 254         src = graph.add_component(MySource, 'src')
 255         it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
 256
 257         # Skip beginning messages.
 258         msg = next(it)
 259         self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 260         msg = next(it)
 261         self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 262
 263         msg_ev1 = next(it)
 264         msg_ev2 = next(it)
 265
 266         self.assertIs(type(msg_ev1), bt2._EventMessageConst)
 267         self.assertIs(type(msg_ev2), bt2._EventMessageConst)
 268         self.assertEqual(msg_ev1.addr, msg_ev2.addr)
 269
 270     # Try consuming many times from an iterator that always returns TryAgain.
 271     # This verifies that we are not missing an incref of Py_None, making the
 272     # refcount of Py_None reach 0.
 273     def test_try_again_many_times(self):
 274         class MyIter(bt2._UserMessageIterator):
 275             def __next__(self):
 276                 raise bt2.TryAgain
 277
 278         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 279             def __init__(self, params, obj):
 280                 self._add_output_port('out')
 281
 282         class MyFilterIter(bt2._UserMessageIterator):
 283             def __init__(self, port):
 284                 input_port = port.user_data
 285                 self._upstream_iter = self._create_input_port_message_iterator(
 286                     input_port
 287                 )
 288
 289             def __next__(self):
 290                 return next(self._upstream_iter)
 291
 292             def _user_seek_beginning(self):
 293                 self._upstream_iter.seek_beginning()
 294
 295             def _user_can_seek_beginning(self):
 296                 return self._upstream_iter.can_seek_beginning()
 297
 298         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 299             def __init__(self, params, obj):
 300                 input_port = self._add_input_port('in')
 301                 self._add_output_port('out', input_port)
 302
 303         graph = bt2.Graph()
 304         src = graph.add_component(MySource, 'src')
 305         it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
 306
 307         # Three times the initial ref count of `None` iterations should
 308         # be enough to catch the bug even if there are small differences
 309         # between configurations.
 310         none_ref_count = sys.getrefcount(None) * 3
 311
 312         for i in range(none_ref_count):
 313             with self.assertRaises(bt2.TryAgain):
 314                 next(it)
 315
 316
 317 def _setup_seek_test(sink_cls, user_seek_beginning=None, user_can_seek_beginning=None):
 318     class MySourceIter(bt2._UserMessageIterator):
 319         def __init__(self, port):
 320             tc, sc, ec = port.user_data
 321             trace = tc()
 322             stream = trace.create_stream(sc)
 323             packet = stream.create_packet()
 324
 325             self._msgs = [
 326                 self._create_stream_beginning_message(stream),
 327                 self._create_packet_beginning_message(packet),
 328                 self._create_event_message(ec, packet),
 329                 self._create_event_message(ec, packet),
 330                 self._create_packet_end_message(packet),
 331                 self._create_stream_end_message(stream),
 332             ]
 333             self._at = 0
 334
 335         def __next__(self):
 336             if self._at < len(self._msgs):
 337                 msg = self._msgs[self._at]
 338                 self._at += 1
 339                 return msg
 340             else:
 341                 raise StopIteration
 342
 343     if user_seek_beginning is not None:
 344         MySourceIter._user_seek_beginning = user_seek_beginning
 345
 346     if user_can_seek_beginning is not None:
 347         MySourceIter._user_can_seek_beginning = user_can_seek_beginning
 348
 349     class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
 350         def __init__(self, params, obj):
 351             tc = self._create_trace_class()
 352             sc = tc.create_stream_class(supports_packets=True)
 353             ec = sc.create_event_class()
 354
 355             self._add_output_port('out', (tc, sc, ec))
 356
 357     class MyFilterIter(bt2._UserMessageIterator):
 358         def __init__(self, port):
 359             self._upstream_iter = self._create_input_port_message_iterator(
 360                 self._component._input_ports['in']
 361             )
 362
 363         def __next__(self):
 364             return next(self._upstream_iter)
 365
 366         def _user_can_seek_beginning(self):
 367             return self._upstream_iter.can_seek_beginning()
 368
 369         def _user_seek_beginning(self):
 370             self._upstream_iter.seek_beginning()
 371
 372     class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 373         def __init__(self, params, obj):
 374             self._add_input_port('in')
 375             self._add_output_port('out')
 376
 377     return _create_graph(MySource, sink_cls, flt_comp_cls=MyFilter)
 378
 379
 380 class UserMessageIteratorSeekBeginningTestCase(unittest.TestCase):
 381     def test_can_seek_beginning(self):
 382         class MySink(bt2._UserSinkComponent):
 383             def __init__(self, params, obj):
 384                 self._add_input_port('in')
 385
 386             def _user_graph_is_configured(self):
 387                 self._msg_iter = self._create_input_port_message_iterator(
 388                     self._input_ports['in']
 389                 )
 390
 391             def _user_consume(self):
 392                 nonlocal can_seek_beginning
 393                 can_seek_beginning = self._msg_iter.can_seek_beginning()
 394
 395         def _user_can_seek_beginning(self):
 396             nonlocal input_port_iter_can_seek_beginning
 397             return input_port_iter_can_seek_beginning
 398
 399         graph = _setup_seek_test(
 400             MySink, user_can_seek_beginning=_user_can_seek_beginning
 401         )
 402
 403         input_port_iter_can_seek_beginning = True
 404         can_seek_beginning = None
 405         graph.run_once()
 406         self.assertTrue(can_seek_beginning)
 407
 408         input_port_iter_can_seek_beginning = False
 409         can_seek_beginning = None
 410         graph.run_once()
 411         self.assertFalse(can_seek_beginning)
 412
 413     def test_no_can_seek_beginning_with_seek_beginning(self):
 414         # Test an iterator without a _user_can_seek_beginning method, but with
 415         # a _user_seek_beginning method.
 416         class MySink(bt2._UserSinkComponent):
 417             def __init__(self, params, obj):
 418                 self._add_input_port('in')
 419
 420             def _user_graph_is_configured(self):
 421                 self._msg_iter = self._create_input_port_message_iterator(
 422                     self._input_ports['in']
 423                 )
 424
 425             def _user_consume(self):
 426                 nonlocal can_seek_beginning
 427                 can_seek_beginning = self._msg_iter.can_seek_beginning()
 428
 429         def _user_seek_beginning(self):
 430             pass
 431
 432         graph = _setup_seek_test(MySink, user_seek_beginning=_user_seek_beginning)
 433         can_seek_beginning = None
 434         graph.run_once()
 435         self.assertTrue(can_seek_beginning)
 436
 437     def test_no_can_seek_beginning(self):
 438         # Test an iterator without a _user_can_seek_beginning method, without
 439         # a _user_seek_beginning method.
 440         class MySink(bt2._UserSinkComponent):
 441             def __init__(self, params, obj):
 442                 self._add_input_port('in')
 443
 444             def _user_graph_is_configured(self):
 445                 self._msg_iter = self._create_input_port_message_iterator(
 446                     self._input_ports['in']
 447                 )
 448
 449             def _user_consume(self):
 450                 nonlocal can_seek_beginning
 451                 can_seek_beginning = self._msg_iter.can_seek_beginning()
 452
 453         graph = _setup_seek_test(MySink)
 454         can_seek_beginning = None
 455         graph.run_once()
 456         self.assertFalse(can_seek_beginning)
 457
 458     def test_can_seek_beginning_user_error(self):
 459         class MySink(bt2._UserSinkComponent):
 460             def __init__(self, params, obj):
 461                 self._add_input_port('in')
 462
 463             def _user_graph_is_configured(self):
 464                 self._msg_iter = self._create_input_port_message_iterator(
 465                     self._input_ports['in']
 466                 )
 467
 468             def _user_consume(self):
 469                 # This is expected to raise.
 470                 self._msg_iter.can_seek_beginning()
 471
 472         def _user_can_seek_beginning(self):
 473             raise ValueError('moustiquaire')
 474
 475         graph = _setup_seek_test(
 476             MySink, user_can_seek_beginning=_user_can_seek_beginning
 477         )
 478
 479         with self.assertRaises(bt2._Error) as ctx:
 480             graph.run_once()
 481
 482         cause = ctx.exception[0]
 483         self.assertIn('ValueError: moustiquaire', cause.message)
 484
 485     def test_can_seek_beginning_wrong_return_value(self):
 486         class MySink(bt2._UserSinkComponent):
 487             def __init__(self, params, obj):
 488                 self._add_input_port('in')
 489
 490             def _user_graph_is_configured(self):
 491                 self._msg_iter = self._create_input_port_message_iterator(
 492                     self._input_ports['in']
 493                 )
 494
 495             def _user_consume(self):
 496                 # This is expected to raise.
 497                 self._msg_iter.can_seek_beginning()
 498
 499         def _user_can_seek_beginning(self):
 500             return 'Amqui'
 501
 502         graph = _setup_seek_test(
 503             MySink, user_can_seek_beginning=_user_can_seek_beginning
 504         )
 505
 506         with self.assertRaises(bt2._Error) as ctx:
 507             graph.run_once()
 508
 509         cause = ctx.exception[0]
 510         self.assertIn("TypeError: 'str' is not a 'bool' object", cause.message)
 511
 512     def test_seek_beginning(self):
 513         class MySink(bt2._UserSinkComponent):
 514             def __init__(self, params, obj):
 515                 self._add_input_port('in')
 516
 517             def _user_graph_is_configured(self):
 518                 self._msg_iter = self._create_input_port_message_iterator(
 519                     self._input_ports['in']
 520                 )
 521
 522             def _user_consume(self):
 523                 nonlocal do_seek_beginning
 524                 nonlocal msg
 525
 526                 if do_seek_beginning:
 527                     self._msg_iter.seek_beginning()
 528                     return
 529
 530                 msg = next(self._msg_iter)
 531
 532         def _user_seek_beginning(self):
 533             self._at = 0
 534
 535         msg = None
 536         graph = _setup_seek_test(MySink, user_seek_beginning=_user_seek_beginning)
 537
 538         # Consume message.
 539         do_seek_beginning = False
 540         graph.run_once()
 541         self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 542
 543         # Consume message.
 544         graph.run_once()
 545         self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 546
 547         # Seek beginning.
 548         do_seek_beginning = True
 549         graph.run_once()
 550
 551         # Consume message.
 552         do_seek_beginning = False
 553         graph.run_once()
 554         self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 555
 556     def test_seek_beginning_user_error(self):
 557         class MySink(bt2._UserSinkComponent):
 558             def __init__(self, params, obj):
 559                 self._add_input_port('in')
 560
 561             def _user_graph_is_configured(self):
 562                 self._msg_iter = self._create_input_port_message_iterator(
 563                     self._input_ports['in']
 564                 )
 565
 566             def _user_consume(self):
 567                 self._msg_iter.seek_beginning()
 568
 569         def _user_seek_beginning(self):
 570             raise ValueError('ouch')
 571
 572         graph = _setup_seek_test(MySink, user_seek_beginning=_user_seek_beginning)
 573
 574         with self.assertRaises(bt2._Error):
 575             graph.run_once()
 576
 577
 578 if __name__ == '__main__':
 579     unittest.main()