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
  24
  25 class UserMessageIteratorTestCase(unittest.TestCase):
  26     @staticmethod
  27     def _create_graph(src_comp_cls, flt_comp_cls=None):
  28         class MySink(bt2._UserSinkComponent):
  29             def __init__(self, params, obj):
  30                 self._add_input_port('in')
  31
  32             def _user_consume(self):
  33                 next(self._msg_iter)
  34
  35             def _user_graph_is_configured(self):
  36                 self._msg_iter = self._create_input_port_message_iterator(
  37                     self._input_ports['in']
  38                 )
  39
  40         graph = bt2.Graph()
  41         src_comp = graph.add_component(src_comp_cls, 'src')
  42
  43         if flt_comp_cls is not None:
  44             flt_comp = graph.add_component(flt_comp_cls, 'flt')
  45
  46         sink_comp = graph.add_component(MySink, 'sink')
  47
  48         if flt_comp_cls is not None:
  49             assert flt_comp is not None
  50             graph.connect_ports(
  51                 src_comp.output_ports['out'], flt_comp.input_ports['in']
  52             )
  53             out_port = flt_comp.output_ports['out']
  54         else:
  55             out_port = src_comp.output_ports['out']
  56
  57         graph.connect_ports(out_port, sink_comp.input_ports['in'])
  58         return graph
  59
  60     def test_init(self):
  61         the_output_port_from_source = None
  62         the_output_port_from_iter = None
  63
  64         class MyIter(bt2._UserMessageIterator):
  65             def __init__(self, self_port_output):
  66                 nonlocal initialized
  67                 nonlocal the_output_port_from_iter
  68                 initialized = True
  69                 the_output_port_from_iter = self_port_output
  70
  71         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
  72             def __init__(self, params, obj):
  73                 nonlocal the_output_port_from_source
  74                 the_output_port_from_source = self._add_output_port('out', 'user data')
  75
  76         initialized = False
  77         graph = self._create_graph(MySource)
  78         graph.run()
  79         self.assertTrue(initialized)
  80         self.assertEqual(
  81             the_output_port_from_source.addr, the_output_port_from_iter.addr
  82         )
  83         self.assertEqual(the_output_port_from_iter.user_data, 'user data')
  84
  85     def test_create_from_message_iterator(self):
  86         class MySourceIter(bt2._UserMessageIterator):
  87             def __init__(self, self_port_output):
  88                 nonlocal src_iter_initialized
  89                 src_iter_initialized = True
  90
  91         class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
  92             def __init__(self, params, obj):
  93                 self._add_output_port('out')
  94
  95         class MyFilterIter(bt2._UserMessageIterator):
  96             def __init__(self, self_port_output):
  97                 nonlocal flt_iter_initialized
  98                 flt_iter_initialized = True
  99                 self._up_iter = self._create_input_port_message_iterator(
 100                     self._component._input_ports['in']
 101                 )
 102
 103             def __next__(self):
 104                 return next(self._up_iter)
 105
 106         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 107             def __init__(self, params, obj):
 108                 self._add_input_port('in')
 109                 self._add_output_port('out')
 110
 111         src_iter_initialized = False
 112         flt_iter_initialized = False
 113         graph = self._create_graph(MySource, MyFilter)
 114         graph.run()
 115         self.assertTrue(src_iter_initialized)
 116         self.assertTrue(flt_iter_initialized)
 117
 118     def test_create_user_error(self):
 119         # This tests both error handling by
 120         # _UserSinkComponent._create_input_port_message_iterator
 121         # and _UserMessageIterator._create_input_port_message_iterator, as they
 122         # are both used in the graph.
 123         class MySourceIter(bt2._UserMessageIterator):
 124             def __init__(self, self_port_output):
 125                 raise ValueError('Very bad error')
 126
 127         class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
 128             def __init__(self, params, obj):
 129                 self._add_output_port('out')
 130
 131         class MyFilterIter(bt2._UserMessageIterator):
 132             def __init__(self, self_port_output):
 133                 # This is expected to raise because of the error in
 134                 # MySourceIter.__init__.
 135                 self._create_input_port_message_iterator(
 136                     self._component._input_ports['in']
 137                 )
 138
 139         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 140             def __init__(self, params, obj):
 141                 self._add_input_port('in')
 142                 self._add_output_port('out')
 143
 144         graph = self._create_graph(MySource, MyFilter)
 145
 146         with self.assertRaises(bt2._Error) as ctx:
 147             graph.run()
 148
 149         exc = ctx.exception
 150         cause = exc[0]
 151
 152         self.assertIsInstance(cause, bt2._MessageIteratorErrorCause)
 153         self.assertEqual(cause.component_name, 'src')
 154         self.assertEqual(cause.component_output_port_name, 'out')
 155         self.assertIn('ValueError: Very bad error', cause.message)
 156
 157     def test_finalize(self):
 158         class MyIter(bt2._UserMessageIterator):
 159             def _user_finalize(self):
 160                 nonlocal finalized
 161                 finalized = True
 162
 163         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 164             def __init__(self, params, obj):
 165                 self._add_output_port('out')
 166
 167         finalized = False
 168         graph = self._create_graph(MySource)
 169         graph.run()
 170         del graph
 171         self.assertTrue(finalized)
 172
 173     def test_component(self):
 174         class MyIter(bt2._UserMessageIterator):
 175             def __init__(self, self_port_output):
 176                 nonlocal salut
 177                 salut = self._component._salut
 178
 179         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 180             def __init__(self, params, obj):
 181                 self._add_output_port('out')
 182                 self._salut = 23
 183
 184         salut = None
 185         graph = self._create_graph(MySource)
 186         graph.run()
 187         self.assertEqual(salut, 23)
 188
 189     def test_addr(self):
 190         class MyIter(bt2._UserMessageIterator):
 191             def __init__(self, self_port_output):
 192                 nonlocal addr
 193                 addr = self.addr
 194
 195         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 196             def __init__(self, params, obj):
 197                 self._add_output_port('out')
 198
 199         addr = None
 200         graph = self._create_graph(MySource)
 201         graph.run()
 202         self.assertIsNotNone(addr)
 203         self.assertNotEqual(addr, 0)
 204
 205     # Test that messages returned by _UserMessageIterator.__next__ remain valid
 206     # and can be re-used.
 207     def test_reuse_message(self):
 208         class MyIter(bt2._UserMessageIterator):
 209             def __init__(self, port):
 210                 tc, sc, ec = port.user_data
 211                 trace = tc()
 212                 stream = trace.create_stream(sc)
 213                 packet = stream.create_packet()
 214
 215                 # This message will be returned twice by __next__.
 216                 event_message = self._create_event_message(ec, packet)
 217
 218                 self._msgs = [
 219                     self._create_stream_beginning_message(stream),
 220                     self._create_packet_beginning_message(packet),
 221                     event_message,
 222                     event_message,
 223                 ]
 224
 225             def __next__(self):
 226                 return self._msgs.pop(0)
 227
 228         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 229             def __init__(self, params, obj):
 230                 tc = self._create_trace_class()
 231                 sc = tc.create_stream_class(supports_packets=True)
 232                 ec = sc.create_event_class()
 233                 self._add_output_port('out', (tc, sc, ec))
 234
 235         graph = bt2.Graph()
 236         src = graph.add_component(MySource, 'src')
 237         it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
 238
 239         # Skip beginning messages.
 240         msg = next(it)
 241         self.assertIsInstance(msg, bt2._StreamBeginningMessage)
 242         msg = next(it)
 243         self.assertIsInstance(msg, bt2._PacketBeginningMessage)
 244
 245         msg_ev1 = next(it)
 246         msg_ev2 = next(it)
 247
 248         self.assertIsInstance(msg_ev1, bt2._EventMessage)
 249         self.assertIsInstance(msg_ev2, bt2._EventMessage)
 250         self.assertEqual(msg_ev1.addr, msg_ev2.addr)
 251
 252     @staticmethod
 253     def _setup_seek_beginning_test(sink_cls):
 254         # Use a source, a filter and an output port iterator.  This allows us
 255         # to test calling `seek_beginning` on both a _OutputPortMessageIterator
 256         # and a _UserComponentInputPortMessageIterator, on top of checking that
 257         # _UserMessageIterator._seek_beginning is properly called.
 258
 259         class MySourceIter(bt2._UserMessageIterator):
 260             def __init__(self, port):
 261                 tc, sc, ec = port.user_data
 262                 trace = tc()
 263                 stream = trace.create_stream(sc)
 264                 packet = stream.create_packet()
 265
 266                 self._msgs = [
 267                     self._create_stream_beginning_message(stream),
 268                     self._create_packet_beginning_message(packet),
 269                     self._create_event_message(ec, packet),
 270                     self._create_event_message(ec, packet),
 271                     self._create_packet_end_message(packet),
 272                     self._create_stream_end_message(stream),
 273                 ]
 274                 self._at = 0
 275
 276             def _user_seek_beginning(self):
 277                 self._at = 0
 278
 279             def __next__(self):
 280                 if self._at < len(self._msgs):
 281                     msg = self._msgs[self._at]
 282                     self._at += 1
 283                     return msg
 284                 else:
 285                     raise StopIteration
 286
 287         class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
 288             def __init__(self, params, obj):
 289                 tc = self._create_trace_class()
 290                 sc = tc.create_stream_class(supports_packets=True)
 291                 ec = sc.create_event_class()
 292
 293                 self._add_output_port('out', (tc, sc, ec))
 294
 295         class MyFilterIter(bt2._UserMessageIterator):
 296             def __init__(self, port):
 297                 input_port = port.user_data
 298                 self._upstream_iter = self._create_input_port_message_iterator(
 299                     input_port
 300                 )
 301
 302             def __next__(self):
 303                 return next(self._upstream_iter)
 304
 305             def _user_seek_beginning(self):
 306                 self._upstream_iter.seek_beginning()
 307
 308             @property
 309             def _user_can_seek_beginning(self):
 310                 return self._upstream_iter.can_seek_beginning
 311
 312         class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
 313             def __init__(self, params, obj):
 314                 input_port = self._add_input_port('in')
 315                 self._add_output_port('out', input_port)
 316
 317         graph = bt2.Graph()
 318         src = graph.add_component(MySource, 'src')
 319         flt = graph.add_component(MyFilter, 'flt')
 320         sink = graph.add_component(sink_cls, 'sink')
 321         graph.connect_ports(src.output_ports['out'], flt.input_ports['in'])
 322         graph.connect_ports(flt.output_ports['out'], sink.input_ports['in'])
 323         return MySourceIter, graph
 324
 325     def test_can_seek_beginning(self):
 326         class MySink(bt2._UserSinkComponent):
 327             def __init__(self, params, obj):
 328                 self._add_input_port('in')
 329
 330             def _user_graph_is_configured(self):
 331                 self._msg_iter = self._create_input_port_message_iterator(
 332                     self._input_ports['in']
 333                 )
 334
 335             def _user_consume(self):
 336                 nonlocal can_seek_beginning
 337                 can_seek_beginning = self._msg_iter.can_seek_beginning
 338
 339         MySourceIter, graph = self._setup_seek_beginning_test(MySink)
 340
 341         def _user_can_seek_beginning(self):
 342             nonlocal input_port_iter_can_seek_beginning
 343             return input_port_iter_can_seek_beginning
 344
 345         MySourceIter._user_can_seek_beginning = property(_user_can_seek_beginning)
 346
 347         input_port_iter_can_seek_beginning = True
 348         can_seek_beginning = None
 349         graph.run_once()
 350         self.assertTrue(can_seek_beginning)
 351
 352         input_port_iter_can_seek_beginning = False
 353         can_seek_beginning = None
 354         graph.run_once()
 355         self.assertFalse(can_seek_beginning)
 356
 357         # Once can_seek_beginning returns an error, verify that it raises when
 358         # _can_seek_beginning has/returns the wrong type.
 359
 360         # Remove the _can_seek_beginning method, we now rely on the presence of
 361         # a _seek_beginning method to know whether the iterator can seek to
 362         # beginning or not.
 363         del MySourceIter._user_can_seek_beginning
 364         can_seek_beginning = None
 365         graph.run_once()
 366         self.assertTrue(can_seek_beginning)
 367
 368         del MySourceIter._user_seek_beginning
 369         can_seek_beginning = None
 370         graph.run_once()
 371         self.assertFalse(can_seek_beginning)
 372
 373     def test_seek_beginning(self):
 374         class MySink(bt2._UserSinkComponent):
 375             def __init__(self, params, obj):
 376                 self._add_input_port('in')
 377
 378             def _user_graph_is_configured(self):
 379                 self._msg_iter = self._create_input_port_message_iterator(
 380                     self._input_ports['in']
 381                 )
 382
 383             def _user_consume(self):
 384                 nonlocal do_seek_beginning
 385                 nonlocal msg
 386
 387                 if do_seek_beginning:
 388                     self._msg_iter.seek_beginning()
 389                     return
 390
 391                 msg = next(self._msg_iter)
 392
 393         do_seek_beginning = False
 394         msg = None
 395         MySourceIter, graph = self._setup_seek_beginning_test(MySink)
 396         graph.run_once()
 397         self.assertIsInstance(msg, bt2._StreamBeginningMessage)
 398         graph.run_once()
 399         self.assertIsInstance(msg, bt2._PacketBeginningMessage)
 400         do_seek_beginning = True
 401         graph.run_once()
 402         do_seek_beginning = False
 403         graph.run_once()
 404         self.assertIsInstance(msg, bt2._StreamBeginningMessage)
 405
 406     def test_seek_beginning_user_error(self):
 407         class MySink(bt2._UserSinkComponent):
 408             def __init__(self, params, obj):
 409                 self._add_input_port('in')
 410
 411             def _user_graph_is_configured(self):
 412                 self._msg_iter = self._create_input_port_message_iterator(
 413                     self._input_ports['in']
 414                 )
 415
 416             def _user_consume(self):
 417                 self._msg_iter.seek_beginning()
 418
 419         MySourceIter, graph = self._setup_seek_beginning_test(MySink)
 420
 421         def _user_seek_beginning_error(self):
 422             raise ValueError('ouch')
 423
 424         MySourceIter._user_seek_beginning = _user_seek_beginning_error
 425
 426         with self.assertRaises(bt2._Error):
 427             graph.run_once()
 428
 429     # Try consuming many times from an iterator that always returns TryAgain.
 430     # This verifies that we are not missing an incref of Py_None, making the
 431     # refcount of Py_None reach 0.
 432     def test_try_again_many_times(self):
 433         class MyIter(bt2._UserMessageIterator):
 434             def __next__(self):
 435                 raise bt2.TryAgain
 436
 437         class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
 438             def __init__(self, params, obj):
 439                 self._add_output_port('out')
 440
 441         graph = bt2.Graph()
 442         src = graph.add_component(MySource, 'src')
 443         it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
 444
 445         # Three times the initial ref count of `None` iterations should
 446         # be enough to catch the bug even if there are small differences
 447         # between configurations.
 448         none_ref_count = sys.getrefcount(None) * 3
 449
 450         for i in range(none_ref_count):
 451             with self.assertRaises(bt2.TryAgain):
 452                 next(it)
 453
 454
 455 if __name__ == '__main__':
 456     unittest.main()