7 @unittest.skip("this is broken")
8 class AutoPopulatePacketContextTimestampsTestCase(unittest
.TestCase
):
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
23 shutil
.rmtree(self
._trace
_path
)
25 def _get_trace_notifs(self
):
31 specs
= [bt2
.ComponentSpec('ctf', 'fs', self
._trace
_path
)]
32 notif_iter
= bt2
.TraceCollectionNotificationIterator(specs
)
33 return list(notif_iter
)
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
()
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
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
63 def _append_event_simple_ts(self
, ts
):
65 evt
.header_field
['timestamp'] = ts
66 evt
.payload_field
['value'] = 23
67 self
._stream
.append_event(evt
)
69 def _append_event_ts_in_payload(self
, header_ts
, payload_ts
):
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
)
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
80 if ts_end
is not None:
81 self
._stream
.packet_context_field
['timestamp_end'] = ts_end
83 def _get_packet_ts_begin_end(self
, notifs
):
84 ts_begin_end_list
= []
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']),
93 return ts_begin_end_list
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)
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)
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)
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)
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)
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)
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)
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)
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)
138 self
._append
_event
_simple
_ts
(2001)
139 self
._append
_event
_simple
_ts
(3500)
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)
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)
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)
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)
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)
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)
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)
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)
183 with self
.assertRaises(bt2
.Error
):
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)
191 self
._set
_ts
_begin
_end
(ts_begin
=1001)
192 self
._append
_event
_simple
_ts
(2001)
193 self
._append
_event
_simple
_ts
(3500)
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)
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)
206 self
._set
_ts
_begin
_end
(ts_begin
=1000)
207 self
._append
_event
_simple
_ts
(2001)
208 self
._append
_event
_simple
_ts
(3500)
210 with self
.assertRaises(bt2
.Error
):
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)
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)
223 def test_empty_packet_auto_ts_begin_end(self
):
224 self
._complete
_sc
_event
_simple
_ts
()
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)
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)
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)
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)
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)
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)
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)
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)
258 self
._set
_ts
_begin
_end
(ts_end
=6000)
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)