Add tests: CTF writer: `timestamp_begin`/`timestamp_end` autopopulation
[babeltrace.git] / tests / lib / ctf-writer / test_auto_populate.py
1 import tempfile
2 import unittest
3 import bt2
4 import shutil
5
6
7 class AutoPopulatePacketContextTimestampsTestCase(unittest.TestCase):
8 def setUp(self):
9 self._trace_path = tempfile.mkdtemp()
10 self._writer = bt2.CtfWriter(self._trace_path)
11 self._cc = bt2.ClockClass('default', int(1e9))
12 self._writer.trace.add_clock_class(self._cc)
13 self._sc = bt2.StreamClass()
14 pcft = bt2.StructureFieldType()
15 pcft.append_field('packet_size', bt2.IntegerFieldType(32))
16 pcft.append_field('content_size', bt2.IntegerFieldType(32))
17 pcft.append_field('timestamp_begin', bt2.IntegerFieldType(64, mapped_clock_class=self._cc))
18 pcft.append_field('timestamp_end', bt2.IntegerFieldType(64, mapped_clock_class=self._cc))
19 self._sc.packet_context_field_type = pcft
20
21 def tearDown(self):
22 shutil.rmtree(self._trace_path)
23
24 def _get_trace_notifs(self):
25 del self._cc
26 del self._sc
27 del self._ec
28 del self._stream
29 del self._writer
30 specs = [bt2.ComponentSpec('ctf', 'fs', self._trace_path)]
31 notif_iter = bt2.TraceCollectionNotificationIterator(specs)
32 return list(notif_iter)
33
34 def _complete_sc(self):
35 self._sc.add_event_class(self._ec)
36 self._writer.trace.add_stream_class(self._sc)
37 self._stream = self._sc()
38
39 def _complete_sc_event_simple_ts(self):
40 ehft = bt2.StructureFieldType()
41 ehft.append_field('id', bt2.IntegerFieldType(32))
42 ehft.append_field('timestamp', bt2.IntegerFieldType(16, mapped_clock_class=self._cc))
43 self._sc.event_header_field_type = ehft
44 self._ec = bt2.EventClass('evt')
45 payloadft = bt2.StructureFieldType()
46 payloadft.append_field('value', bt2.IntegerFieldType(32))
47 self._ec.payload_field_type = payloadft
48 self._complete_sc()
49
50 def _complete_sc_event_ts_in_payload(self):
51 ehft = bt2.StructureFieldType()
52 ehft.append_field('id', bt2.IntegerFieldType(32))
53 ehft.append_field('timestamp', bt2.IntegerFieldType(16, mapped_clock_class=self._cc))
54 self._sc.event_header_field_type = ehft
55 self._ec = bt2.EventClass('evt')
56 payloadft = bt2.StructureFieldType()
57 payloadft.append_field('value', bt2.IntegerFieldType(32))
58 payloadft.append_field('a_ts', bt2.IntegerFieldType(8, mapped_clock_class=self._cc))
59 self._ec.payload_field_type = payloadft
60 self._complete_sc()
61
62 def _append_event_simple_ts(self, ts):
63 evt = self._ec()
64 evt.header_field['timestamp'] = ts
65 evt.payload_field['value'] = 23
66 self._stream.append_event(evt)
67
68 def _append_event_ts_in_payload(self, header_ts, payload_ts):
69 evt = self._ec()
70 evt.header_field['timestamp'] = header_ts
71 evt.payload_field['value'] = 23
72 evt.payload_field['a_ts'] = payload_ts
73 self._stream.append_event(evt)
74
75 def _set_ts_begin_end(self, ts_begin=None, ts_end=None):
76 if ts_begin is not None:
77 self._stream.packet_context_field['timestamp_begin'] = ts_begin
78
79 if ts_end is not None:
80 self._stream.packet_context_field['timestamp_end'] = ts_end
81
82 def _get_packet_ts_begin_end(self, notifs):
83 ts_begin_end_list = []
84
85 for notif in notifs:
86 if type(notif) is bt2.PacketBeginningNotification:
87 ts_begin_end_list.append((
88 int(notif.packet.context_field['timestamp_begin']),
89 int(notif.packet.context_field['timestamp_end']),
90 ))
91
92 return ts_begin_end_list
93
94 def test_ts_inc(self):
95 self._complete_sc_event_simple_ts()
96 self._append_event_simple_ts(12)
97 self._append_event_simple_ts(144)
98 self._stream.flush()
99 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
100 self.assertEqual(ts_begin_end_list[0][0], 0)
101 self.assertEqual(ts_begin_end_list[0][1], 144)
102
103 def test_ts_equal(self):
104 self._complete_sc_event_simple_ts()
105 self._append_event_simple_ts(12)
106 self._append_event_simple_ts(144)
107 self._append_event_simple_ts(144)
108 self._stream.flush()
109 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
110 self.assertEqual(ts_begin_end_list[0][0], 0)
111 self.assertEqual(ts_begin_end_list[0][1], 144)
112
113 def test_ts_wraps(self):
114 self._complete_sc_event_simple_ts()
115 self._append_event_simple_ts(12)
116 self._append_event_simple_ts(144)
117 self._append_event_simple_ts(11)
118 self._stream.flush()
119 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
120 self.assertEqual(ts_begin_end_list[0][0], 0)
121 self.assertEqual(ts_begin_end_list[0][1], 65547)
122
123 def test_ts_begin_from_0(self):
124 self._complete_sc_event_simple_ts()
125 self._append_event_simple_ts(12)
126 self._append_event_simple_ts(144)
127 self._stream.flush()
128 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
129 self.assertEqual(ts_begin_end_list[0][0], 0)
130 self.assertEqual(ts_begin_end_list[0][1], 144)
131
132 def test_ts_begin_from_last_ts_end(self):
133 self._complete_sc_event_simple_ts()
134 self._append_event_simple_ts(12)
135 self._append_event_simple_ts(144)
136 self._stream.flush()
137 self._append_event_simple_ts(2001)
138 self._append_event_simple_ts(3500)
139 self._stream.flush()
140 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
141 self.assertEqual(ts_begin_end_list[0][0], 0)
142 self.assertEqual(ts_begin_end_list[0][1], 144)
143 self.assertEqual(ts_begin_end_list[1][0], 144)
144 self.assertEqual(ts_begin_end_list[1][1], 3500)
145
146 def test_ts_begin_from_provided(self):
147 self._complete_sc_event_simple_ts()
148 self._set_ts_begin_end(ts_begin=17)
149 self._append_event_simple_ts(11)
150 self._append_event_simple_ts(15)
151 self._stream.flush()
152 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
153 self.assertEqual(ts_begin_end_list[0][0], 17)
154 self.assertEqual(ts_begin_end_list[0][1], 65551)
155
156 def test_ts_end_from_provided(self):
157 self._complete_sc_event_simple_ts()
158 self._set_ts_begin_end(ts_end=1001)
159 self._append_event_simple_ts(11)
160 self._append_event_simple_ts(15)
161 self._stream.flush()
162 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
163 self.assertEqual(ts_begin_end_list[0][0], 0)
164 self.assertEqual(ts_begin_end_list[0][1], 1001)
165
166 def test_ts_end_from_provided_equal(self):
167 self._complete_sc_event_simple_ts()
168 self._set_ts_begin_end(ts_end=1001)
169 self._append_event_simple_ts(11)
170 self._append_event_simple_ts(1001)
171 self._stream.flush()
172 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
173 self.assertEqual(ts_begin_end_list[0][0], 0)
174 self.assertEqual(ts_begin_end_list[0][1], 1001)
175
176 def test_ts_end_from_provided_too_small(self):
177 self._complete_sc_event_simple_ts()
178 self._set_ts_begin_end(ts_end=1001)
179 self._append_event_simple_ts(11)
180 self._append_event_simple_ts(1002)
181
182 with self.assertRaises(bt2.Error):
183 self._stream.flush()
184
185 def test_ts_begin_provided_equal_last_ts_end_two_packets(self):
186 self._complete_sc_event_simple_ts()
187 self._append_event_simple_ts(12)
188 self._append_event_simple_ts(1001)
189 self._stream.flush()
190 self._set_ts_begin_end(ts_begin=1001)
191 self._append_event_simple_ts(2001)
192 self._append_event_simple_ts(3500)
193 self._stream.flush()
194 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
195 self.assertEqual(ts_begin_end_list[0][0], 0)
196 self.assertEqual(ts_begin_end_list[0][1], 1001)
197 self.assertEqual(ts_begin_end_list[1][0], 1001)
198 self.assertEqual(ts_begin_end_list[1][1], 3500)
199
200 def test_ts_begin_provided_less_than_last_ts_end_two_packets(self):
201 self._complete_sc_event_simple_ts()
202 self._append_event_simple_ts(12)
203 self._append_event_simple_ts(1001)
204 self._stream.flush()
205 self._set_ts_begin_end(ts_begin=1000)
206 self._append_event_simple_ts(2001)
207 self._append_event_simple_ts(3500)
208
209 with self.assertRaises(bt2.Error):
210 self._stream.flush()
211
212 def test_ts_end_from_value_event_payload(self):
213 self._complete_sc_event_ts_in_payload()
214 self._append_event_ts_in_payload(11, 15)
215 self._append_event_ts_in_payload(5, 10)
216 self._append_event_ts_in_payload(18, 10)
217 self._stream.flush()
218 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
219 self.assertEqual(ts_begin_end_list[0][0], 0)
220 self.assertEqual(ts_begin_end_list[0][1], 65802)
221
222 def test_empty_packet_auto_ts_begin_end(self):
223 self._complete_sc_event_simple_ts()
224 self._stream.flush()
225 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
226 self.assertEqual(ts_begin_end_list[0][0], 0)
227 self.assertEqual(ts_begin_end_list[0][1], 0)
228
229 def test_empty_packet_provided_ts_begin(self):
230 self._complete_sc_event_simple_ts()
231 self._set_ts_begin_end(ts_begin=1001)
232 self._stream.flush()
233 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
234 self.assertEqual(ts_begin_end_list[0][0], 1001)
235 self.assertEqual(ts_begin_end_list[0][1], 1001)
236
237 def test_empty_packet_provided_ts_end(self):
238 self._complete_sc_event_simple_ts()
239 self._set_ts_begin_end(ts_end=1001)
240 self._stream.flush()
241 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
242 self.assertEqual(ts_begin_end_list[0][0], 0)
243 self.assertEqual(ts_begin_end_list[0][1], 1001)
244
245 def test_empty_packet_provided_ts_begin_end(self):
246 self._complete_sc_event_simple_ts()
247 self._set_ts_begin_end(1001, 3003)
248 self._stream.flush()
249 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
250 self.assertEqual(ts_begin_end_list[0][0], 1001)
251 self.assertEqual(ts_begin_end_list[0][1], 3003)
252
253 def test_empty_packet_ts_begin_from_last_ts_end(self):
254 self._complete_sc_event_simple_ts()
255 self._set_ts_begin_end(1001, 3003)
256 self._stream.flush()
257 self._set_ts_begin_end(ts_end=6000)
258 self._stream.flush()
259 ts_begin_end_list = self._get_packet_ts_begin_end(self._get_trace_notifs())
260 self.assertEqual(ts_begin_end_list[0][0], 1001)
261 self.assertEqual(ts_begin_end_list[0][1], 3003)
262 self.assertEqual(ts_begin_end_list[1][0], 3003)
263 self.assertEqual(ts_begin_end_list[1][1], 6000)
This page took 0.037706 seconds and 4 git commands to generate.