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