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