lib: make can_seek_ns_from_origin logic use `can_seek_forward` property of iterator
[babeltrace.git] / tests / data / plugins / flt.utils.trimmer / bt_plugin_trimmer_test.py
1 import bt2
2
3
4 class TheIteratorOfAllEvil(bt2._UserMessageIterator):
5 def __init__(self, config, port):
6 tc, sc, ec1, ec2, params = port.user_data
7 trace = tc()
8 stream = trace.create_stream(sc)
9
10 # Test with and without packets, once packets are optional.
11 packet = stream.create_packet()
12
13 if params['with-stream-msgs-cs']:
14 sb_msg = self._create_stream_beginning_message(stream, 100)
15 else:
16 sb_msg = self._create_stream_beginning_message(stream)
17
18 ev_msg1 = self._create_event_message(ec1, packet, 300)
19 ev_msg2 = self._create_event_message(ec2, packet, 400)
20
21 if params['with-stream-msgs-cs']:
22 se_msg = self._create_stream_end_message(stream, 1000)
23 else:
24 se_msg = self._create_stream_end_message(stream)
25
26 self._msgs = [
27 sb_msg,
28 self._create_packet_beginning_message(packet, 200),
29 ev_msg1,
30 ev_msg2,
31 self._create_packet_end_message(packet, 900),
32 se_msg,
33 ]
34 self._at = 0
35 config.can_seek_forward = True
36
37 def _user_seek_beginning(self):
38 self._at = 0
39
40 def __next__(self):
41 if self._at < len(self._msgs):
42 msg = self._msgs[self._at]
43 self._at += 1
44 return msg
45 else:
46 raise StopIteration
47
48
49 @bt2.plugin_component_class
50 class TheSourceOfAllEvil(
51 bt2._UserSourceComponent, message_iterator_class=TheIteratorOfAllEvil
52 ):
53 def __init__(self, config, params, obj):
54 tc = self._create_trace_class()
55
56 # Use a clock class with an offset, so we can test with --begin or --end
57 # smaller than this offset (in other words, a time that it's not
58 # possible to represent with this clock class).
59 cc = self._create_clock_class(frequency=1, offset=bt2.ClockClassOffset(10000))
60 sc = tc.create_stream_class(
61 default_clock_class=cc,
62 supports_packets=True,
63 packets_have_beginning_default_clock_snapshot=True,
64 packets_have_end_default_clock_snapshot=True,
65 )
66 ec1 = sc.create_event_class(name='event 1')
67 ec2 = sc.create_event_class(name='event 2')
68 self._add_output_port('out', (tc, sc, ec1, ec2, params))
69
70
71 bt2.register_plugin(__name__, 'test-trimmer')
This page took 0.033095 seconds and 4 git commands to generate.