tests/data/plugins/flt.utils.trimmer/bt_plugin_trimmer_test.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import bt2
   7
   8
   9 class TheIteratorOfAllEvil(bt2._UserMessageIterator):
  10     def __init__(self, config, port):
  11         tc, sc, ec1, ec2, params = port.user_data
  12         trace = tc()
  13         stream = trace.create_stream(sc)
  14
  15         if params['with-packet-msgs']:
  16             packet = stream.create_packet()
  17
  18         if params['with-stream-msgs-cs']:
  19             sb_msg = self._create_stream_beginning_message(stream, 100)
  20         else:
  21             sb_msg = self._create_stream_beginning_message(stream)
  22
  23         parent = packet if params['with-packet-msgs'] else stream
  24
  25         ev_msg1 = self._create_event_message(ec1, parent, 300)
  26         ev_msg2 = self._create_event_message(ec2, parent, 400)
  27
  28         if params['with-stream-msgs-cs']:
  29             se_msg = self._create_stream_end_message(stream, 1000)
  30         else:
  31             se_msg = self._create_stream_end_message(stream)
  32
  33         self._msgs = []
  34
  35         self._msgs.append(sb_msg)
  36
  37         if params['with-packet-msgs']:
  38             self._msgs.append(self._create_packet_beginning_message(packet, 200))
  39
  40         self._msgs.append(ev_msg1)
  41         self._msgs.append(ev_msg2)
  42
  43         if params['with-packet-msgs']:
  44             self._msgs.append(self._create_packet_end_message(packet, 900))
  45
  46         self._msgs.append(se_msg)
  47
  48         self._at = 0
  49         config.can_seek_forward = True
  50
  51     def _user_seek_beginning(self):
  52         self._at = 0
  53
  54     def __next__(self):
  55         if self._at < len(self._msgs):
  56             msg = self._msgs[self._at]
  57             self._at += 1
  58             return msg
  59         else:
  60             raise StopIteration
  61
  62
  63 @bt2.plugin_component_class
  64 class TheSourceOfAllEvil(
  65     bt2._UserSourceComponent, message_iterator_class=TheIteratorOfAllEvil
  66 ):
  67     def __init__(self, config, params, obj):
  68         tc = self._create_trace_class()
  69
  70         with_packets = bool(params['with-packet-msgs'])
  71
  72         # Use a clock class with an offset, so we can test with --begin or --end
  73         # smaller than this offset (in other words, a time that it's not
  74         # possible to represent with this clock class).
  75         cc = self._create_clock_class(frequency=1, offset=bt2.ClockClassOffset(10000))
  76         sc = tc.create_stream_class(
  77             default_clock_class=cc,
  78             supports_packets=with_packets,
  79             packets_have_beginning_default_clock_snapshot=with_packets,
  80             packets_have_end_default_clock_snapshot=with_packets,
  81         )
  82         ec1 = sc.create_event_class(name='event 1')
  83         ec2 = sc.create_event_class(name='event 2')
  84         self._add_output_port('out', (tc, sc, ec1, ec2, params))
  85
  86
  87 bt2.register_plugin(__name__, 'test-trimmer')