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