7 class AutoPopulatePacketContextTimestampsTestCase(unittest
.TestCase
):
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
22 shutil
.rmtree(self
._trace
_path
)
24 def _get_trace_notifs(self
):
30 specs
= [bt2
.ComponentSpec('ctf', 'fs', self
._trace
_path
)]
31 notif_iter
= bt2
.TraceCollectionNotificationIterator(specs
)
32 return list(notif_iter
)
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
()
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
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
62 def _append_event_simple_ts(self
, ts
):
64 evt
.header_field
['timestamp'] = ts
65 evt
.payload_field
['value'] = 23
66 self
._stream
.append_event(evt
)
68 def _append_event_ts_in_payload(self
, header_ts
, payload_ts
):
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
)
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
79 if ts_end
is not None:
80 self
._stream
.packet_context_field
['timestamp_end'] = ts_end
82 def _get_packet_ts_begin_end(self
, notifs
):
83 ts_begin_end_list
= []
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']),
92 return ts_begin_end_list
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)
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)
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)
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)
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)
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)
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)
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)
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)
137 self
._append
_event
_simple
_ts
(2001)
138 self
._append
_event
_simple
_ts
(3500)
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)
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)
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)
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)
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)
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)
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)
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)
182 with self
.assertRaises(bt2
.Error
):
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)
190 self
._set
_ts
_begin
_end
(ts_begin
=1001)
191 self
._append
_event
_simple
_ts
(2001)
192 self
._append
_event
_simple
_ts
(3500)
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)
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)
205 self
._set
_ts
_begin
_end
(ts_begin
=1000)
206 self
._append
_event
_simple
_ts
(2001)
207 self
._append
_event
_simple
_ts
(3500)
209 with self
.assertRaises(bt2
.Error
):
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)
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)
222 def test_empty_packet_auto_ts_begin_end(self
):
223 self
._complete
_sc
_event
_simple
_ts
()
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)
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)
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)
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)
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)
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)
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)
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)
257 self
._set
_ts
_begin
_end
(ts_end
=6000)
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)