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             @property
 296             def _user_can_seek_beginning(self):
 297                 return self._upstream_iter.can_seek_beginning
 298
 299         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 300             def __init__(self, params, obj):
 301                 input_port = self._add_input_port('in')
 302                 self._add_output_port('out', input_port)
 303
 304         graph = bt2.Graph()
 305         src = graph.add_component(MySource, 'src')
 306         it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
 307
 308         # Three times the initial ref count of `None` iterations should
 309         # be enough to catch the bug even if there are small differences
 310         # between configurations.
 311         none_ref_count = sys.getrefcount(None) * 3
 312
 313         for i in range(none_ref_count):
 314             with self.assertRaises(bt2.TryAgain):
 315                 next(it)
 316
 317
 318 def _setup_seek_test(sink_cls, user_seek_beginning=None, user_can_seek_beginning=None):
 319     class MySourceIter(bt2._UserMessageIterator):
 320         def __init__(self, port):
 321             tc, sc, ec = port.user_data
 322             trace = tc()
 323             stream = trace.create_stream(sc)
 324             packet = stream.create_packet()
 325
 326             self._msgs = [
 327                 self._create_stream_beginning_message(stream),
 328                 self._create_packet_beginning_message(packet),
 329                 self._create_event_message(ec, packet),
 330                 self._create_event_message(ec, packet),
 331                 self._create_packet_end_message(packet),
 332                 self._create_stream_end_message(stream),
 333             ]
 334             self._at = 0
 335
 336         def __next__(self):
 337             if self._at < len(self._msgs):
 338                 msg = self._msgs[self._at]
 339                 self._at += 1
 340                 return msg
 341             else:
 342                 raise StopIteration
 343
 344     if user_seek_beginning is not None:
 345         MySourceIter._user_seek_beginning = user_seek_beginning
 346
 347     if user_can_seek_beginning is not None:
 348         MySourceIter._user_can_seek_beginning = property(user_can_seek_beginning)
 349
 350     class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
 351         def __init__(self, params, obj):
 352             tc = self._create_trace_class()
 353             sc = tc.create_stream_class(supports_packets=True)
 354             ec = sc.create_event_class()
 355
 356             self._add_output_port('out', (tc, sc, ec))
 357
 358     class MyFilterIter(bt2._UserMessageIterator):
 359         def __init__(self, port):
 360             self._upstream_iter = self._create_input_port_message_iterator(
 361                 self._component._input_ports['in']
 362             )
 363
 364         def __next__(self):
 365             return next(self._upstream_iter)
 366
 367         @property
 368         def _user_can_seek_beginning(self):
 369             return self._upstream_iter.can_seek_beginning
 370
 371         def _user_seek_beginning(self):
 372             self._upstream_iter.seek_beginning()
 373
 374     class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 375         def __init__(self, params, obj):
 376             self._add_input_port('in')
 377             self._add_output_port('out')
 378
 379     return _create_graph(MySource, sink_cls, flt_comp_cls=MyFilter)
 380
 381
 382 class UserMessageIteratorSeekBeginningTestCase(unittest.TestCase):
 383     def test_can_seek_beginning(self):
 384         class MySink(bt2._UserSinkComponent):
 385             def __init__(self, params, obj):
 386                 self._add_input_port('in')
 387
 388             def _user_graph_is_configured(self):
 389                 self._msg_iter = self._create_input_port_message_iterator(
 390                     self._input_ports['in']
 391                 )
 392
 393             def _user_consume(self):
 394                 nonlocal can_seek_beginning
 395                 can_seek_beginning = self._msg_iter.can_seek_beginning
 396
 397         def _user_can_seek_beginning(self):
 398             nonlocal input_port_iter_can_seek_beginning
 399             return input_port_iter_can_seek_beginning
 400
 401         graph = _setup_seek_test(
 402             MySink, user_can_seek_beginning=_user_can_seek_beginning
 403         )
 404
 405         input_port_iter_can_seek_beginning = True
 406         can_seek_beginning = None
 407         graph.run_once()
 408         self.assertTrue(can_seek_beginning)
 409
 410         input_port_iter_can_seek_beginning = False
 411         can_seek_beginning = None
 412         graph.run_once()
 413         self.assertFalse(can_seek_beginning)
 414
 415     def test_no_can_seek_beginning_with_seek_beginning(self):
 416         # Test an iterator without a _user_can_seek_beginning method, but with
 417         # a _user_seek_beginning method.
 418         class MySink(bt2._UserSinkComponent):
 419             def __init__(self, params, obj):
 420                 self._add_input_port('in')
 421
 422             def _user_graph_is_configured(self):
 423                 self._msg_iter = self._create_input_port_message_iterator(
 424                     self._input_ports['in']
 425                 )
 426
 427             def _user_consume(self):
 428                 nonlocal can_seek_beginning
 429                 can_seek_beginning = self._msg_iter.can_seek_beginning
 430
 431         def _user_seek_beginning(self):
 432             pass
 433
 434         graph = _setup_seek_test(MySink, user_seek_beginning=_user_seek_beginning)
 435         can_seek_beginning = None
 436         graph.run_once()
 437         self.assertTrue(can_seek_beginning)
 438
 439     def test_no_can_seek_beginning(self):
 440         # Test an iterator without a _user_can_seek_beginning method, without
 441         # a _user_seek_beginning method.
 442         class MySink(bt2._UserSinkComponent):
 443             def __init__(self, params, obj):
 444                 self._add_input_port('in')
 445
 446             def _user_graph_is_configured(self):
 447                 self._msg_iter = self._create_input_port_message_iterator(
 448                     self._input_ports['in']
 449                 )
 450
 451             def _user_consume(self):
 452                 nonlocal can_seek_beginning
 453                 can_seek_beginning = self._msg_iter.can_seek_beginning
 454
 455         graph = _setup_seek_test(MySink)
 456         can_seek_beginning = None
 457         graph.run_once()
 458         self.assertFalse(can_seek_beginning)
 459
 460     def test_can_seek_beginning_user_error(self):
 461         class MySink(bt2._UserSinkComponent):
 462             def __init__(self, params, obj):
 463                 self._add_input_port('in')
 464
 465             def _user_graph_is_configured(self):
 466                 self._msg_iter = self._create_input_port_message_iterator(
 467                     self._input_ports['in']
 468                 )
 469
 470             def _user_consume(self):
 471                 # This is expected to raise.
 472                 self._msg_iter.can_seek_beginning
 473
 474         def _user_can_seek_beginning(self):
 475             raise ValueError('moustiquaire')
 476
 477         graph = _setup_seek_test(
 478             MySink, user_can_seek_beginning=_user_can_seek_beginning
 479         )
 480
 481         with self.assertRaises(bt2._Error) as ctx:
 482             graph.run_once()
 483
 484         cause = ctx.exception[0]
 485         self.assertIn('ValueError: moustiquaire', cause.message)
 486
 487     def test_can_seek_beginning_wrong_return_value(self):
 488         class MySink(bt2._UserSinkComponent):
 489             def __init__(self, params, obj):
 490                 self._add_input_port('in')
 491
 492             def _user_graph_is_configured(self):
 493                 self._msg_iter = self._create_input_port_message_iterator(
 494                     self._input_ports['in']
 495                 )
 496
 497             def _user_consume(self):
 498                 # This is expected to raise.
 499                 self._msg_iter.can_seek_beginning
 500
 501         def _user_can_seek_beginning(self):
 502             return 'Amqui'
 503
 504         graph = _setup_seek_test(
 505             MySink, user_can_seek_beginning=_user_can_seek_beginning
 506         )
 507
 508         with self.assertRaises(bt2._Error) as ctx:
 509             graph.run_once()
 510
 511         cause = ctx.exception[0]
 512         self.assertIn("TypeError: 'str' is not a 'bool' object", cause.message)
 513
 514     def test_seek_beginning(self):
 515         class MySink(bt2._UserSinkComponent):
 516             def __init__(self, params, obj):
 517                 self._add_input_port('in')
 518
 519             def _user_graph_is_configured(self):
 520                 self._msg_iter = self._create_input_port_message_iterator(
 521                     self._input_ports['in']
 522                 )
 523
 524             def _user_consume(self):
 525                 nonlocal do_seek_beginning
 526                 nonlocal msg
 527
 528                 if do_seek_beginning:
 529                     self._msg_iter.seek_beginning()
 530                     return
 531
 532                 msg = next(self._msg_iter)
 533
 534         def _user_seek_beginning(self):
 535             self._at = 0
 536
 537         msg = None
 538         graph = _setup_seek_test(MySink, user_seek_beginning=_user_seek_beginning)
 539
 540         # Consume message.
 541         do_seek_beginning = False
 542         graph.run_once()
 543         self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 544
 545         # Consume message.
 546         graph.run_once()
 547         self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 548
 549         # Seek beginning.
 550         do_seek_beginning = True
 551         graph.run_once()
 552
 553         # Consume message.
 554         do_seek_beginning = False
 555         graph.run_once()
 556         self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 557
 558     def test_seek_beginning_user_error(self):
 559         class MySink(bt2._UserSinkComponent):
 560             def __init__(self, params, obj):
 561                 self._add_input_port('in')
 562
 563             def _user_graph_is_configured(self):
 564                 self._msg_iter = self._create_input_port_message_iterator(
 565                     self._input_ports['in']
 566                 )
 567
 568             def _user_consume(self):
 569                 self._msg_iter.seek_beginning()
 570
 571         def _user_seek_beginning(self):
 572             raise ValueError('ouch')
 573
 574         graph = _setup_seek_test(MySink, user_seek_beginning=_user_seek_beginning)
 575
 576         with self.assertRaises(bt2._Error):
 577             graph.run_once()
 578
 579
 580 if __name__ == '__main__':
 581     unittest.main()