e0e716b21b73a2fda3b9392bd8d4fb18ab0e2818
1 # Copyright (C) 2019 Simon Marchi <simon.marchi@efficios.com>
3 # This program is free software; you can redistribute it and/or
4 # modify it under the terms of the GNU General Public License
5 # as published by the Free Software Foundation; only version 2
8 # This program is distributed in the hope that it will be useful,
9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 # GNU General Public License for more details.
13 # You should have received a copy of the GNU General Public License
14 # along with this program; if not, write to the Free Software
15 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
23 test_ctf_traces_path
= os
.environ
['BT_CTF_TRACES_PATH']
26 # Key to sort streams in a predictable order.
27 def sort_predictably(stream
):
28 return stream
['port-name']
31 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
33 ctf
= bt2
.find_plugin('ctf')
34 self
._fs
= ctf
.source_component_classes
['fs']
37 os
.path
.join(test_ctf_traces_path
, 'intersection', '3eventsintersect')
40 def _check(self
, trace
, offset
):
41 streams
= sorted(trace
['stream-infos'], key
=sort_predictably
)
42 self
.assertEqual(streams
[0]['range-ns']['begin'], 13515309000000000 + offset
)
43 self
.assertEqual(streams
[0]['range-ns']['end'], 13515309000000100 + offset
)
44 self
.assertEqual(streams
[1]['range-ns']['begin'], 13515309000000070 + offset
)
45 self
.assertEqual(streams
[1]['range-ns']['end'], 13515309000000120 + offset
)
47 # Test various cominations of the clock-class-offset-s and
48 # clock-class-offset-ns parameters to babeltrace.trace-infos queries.
50 # Without clock class offset
52 def test_no_clock_class_offset(self
):
53 res
= bt2
.QueryExecutor(
54 self
._fs
, 'babeltrace.trace-infos', {'inputs': self
._inputs
}
59 # With clock-class-offset-s
61 def test_clock_class_offset_s(self
):
62 res
= bt2
.QueryExecutor(
64 'babeltrace.trace-infos',
65 {'inputs': self
._inputs
, 'clock-class-offset-s': 2},
68 self
._check
(trace
, 2000000000)
70 # With clock-class-offset-ns
72 def test_clock_class_offset_ns(self
):
73 res
= bt2
.QueryExecutor(
75 'babeltrace.trace-infos',
76 {'inputs': self
._inputs
, 'clock-class-offset-ns': 2},
83 def test_clock_class_offset_both(self
):
84 res
= bt2
.QueryExecutor(
86 'babeltrace.trace-infos',
88 'inputs': self
._inputs
,
89 'clock-class-offset-s': -2,
90 'clock-class-offset-ns': -2,
94 self
._check
(trace
, -2000000002)
96 def test_clock_class_offset_s_wrong_type(self
):
97 with self
.assertRaises(bt2
._Error
):
100 'babeltrace.trace-infos',
101 {'inputs': self
._inputs
, 'clock-class-offset-s': "2"},
104 def test_clock_class_offset_s_wrong_type_none(self
):
105 with self
.assertRaises(bt2
._Error
):
108 'babeltrace.trace-infos',
109 {'inputs': self
._inputs
, 'clock-class-offset-s': None},
112 def test_clock_class_offset_ns_wrong_type(self
):
113 with self
.assertRaises(bt2
._Error
):
116 'babeltrace.trace-infos',
117 {'inputs': self
._inputs
, 'clock-class-offset-ns': "2"},
120 def test_clock_class_offset_ns_wrong_type_none(self
):
121 with self
.assertRaises(bt2
._Error
):
124 'babeltrace.trace-infos',
125 {'inputs': self
._inputs
, 'clock-class-offset-ns': None},
129 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
131 ctf
= bt2
.find_plugin("ctf")
132 self
._fs
= ctf
.source_component_classes
["fs"]
134 def test_trace_uuid_stream_class_id_no_stream_id(self
):
135 res
= bt2
.QueryExecutor(
137 "babeltrace.trace-infos",
141 test_ctf_traces_path
, "intersection", "3eventsintersect"
147 if os
.environ
['BT_OS_TYPE'] == 'mingw':
149 '\\tests\\data\\ctf-traces\\intersection\\3eventsintersect\\'
152 os_stream_path
= '/tests/data/ctf-traces/intersection/3eventsintersect/'
154 self
.assertEqual(len(res
), 1)
156 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
157 self
.assertEqual(len(streams
), 2)
159 str(streams
[0]["port-name"]),
160 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
161 + re
.escape(os_stream_path
+ "test_stream_0")
165 str(streams
[1]["port-name"]),
166 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
167 + re
.escape(os_stream_path
+ "test_stream_1")
171 def test_trace_uuid_no_stream_class_id_no_stream_id(self
):
172 res
= bt2
.QueryExecutor(
174 "babeltrace.trace-infos",
175 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
178 if os
.environ
['BT_OS_TYPE'] == 'mingw':
179 os_stream_path
= '\\tests\\data\\ctf-traces\\succeed\\succeed1\\dummystream'
181 os_stream_path
= '/tests/data/ctf-traces/succeed/succeed1/dummystream'
183 self
.assertEqual(len(res
), 1)
185 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
186 self
.assertEqual(len(streams
), 1)
188 str(streams
[0]["port-name"]),
189 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
190 + re
.escape(os_stream_path
)
195 class QueryTraceInfoRangeTestCase(unittest
.TestCase
):
197 ctf
= bt2
.find_plugin("ctf")
198 self
._fs
= ctf
.source_component_classes
["fs"]
200 def test_trace_no_range(self
):
201 # This trace has no `timestamp_begin` and `timestamp_end` in its
202 # packet context. The `babeltrace.trace-infos` query should omit
203 # the `range-ns` fields in the `trace` and `stream` data
206 res
= bt2
.QueryExecutor(
208 "babeltrace.trace-infos",
209 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
212 self
.assertEqual(len(res
), 1)
214 streams
= trace
["stream-infos"]
215 self
.assertEqual(len(streams
), 1)
217 self
.assertRaises(KeyError, lambda: trace
['range-ns'])
218 self
.assertRaises(KeyError, lambda: streams
[0]['range-ns'])
220 def test_trace_with_tracefile_rotation(self
):
221 res
= bt2
.QueryExecutor(
223 "babeltrace.trace-infos",
227 test_ctf_traces_path
,
229 "lttng-tracefile-rotation",
236 self
.assertEqual(len(res
), 1)
238 streams
= trace
["stream-infos"]
239 self
.assertEqual(len(streams
), 4)
241 # Note: the end timestamps are not the end timestamps found in the
242 # index files, because fix_index_lttng_event_after_packet_bug changes
243 # them based on the time of the last event in the stream.
245 self
.assertEqual(streams
[0]['range-ns']['begin'], 1571261795455986789)
246 self
.assertEqual(streams
[0]['range-ns']['end'], 1571261797582611840)
248 self
.assertEqual(streams
[1]['range-ns']['begin'], 1571261795456368232)
249 self
.assertEqual(streams
[1]['range-ns']['end'], 1571261797577754111)
251 self
.assertEqual(streams
[2]['range-ns']['begin'], 1571261795456748255)
252 self
.assertEqual(streams
[2]['range-ns']['end'], 1571261797577727795)
254 self
.assertEqual(streams
[3]['range-ns']['begin'], 1571261795457285142)
255 self
.assertEqual(streams
[3]['range-ns']['end'], 1571261797582522088)
258 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest
.TestCase
):
260 ctf
= bt2
.find_plugin('ctf')
261 self
._fs
= ctf
.source_component_classes
['fs']
262 self
._path
= os
.path
.join(test_ctf_traces_path
, 'succeed')
264 def _test_lttng_quirks(self
, trace_name
):
265 res
= bt2
.QueryExecutor(
267 "babeltrace.trace-infos",
268 {"inputs": [os
.path
.join(self
._path
, trace_name
)]},
271 self
.assertEqual(len(res
), 1)
274 def test_event_after_packet(self
):
275 trace
= self
._test
_lttng
_quirks
("lttng-event-after-packet")
276 streams
= trace
["stream-infos"]
277 self
.assertEqual(len(streams
), 1)
279 self
.assertEqual(streams
[0]['range-ns']['begin'], 1565957300948091100)
280 self
.assertEqual(streams
[0]['range-ns']['end'], 1565957302180016069)
282 def test_lttng_crash(self
):
283 trace
= self
._test
_lttng
_quirks
("lttng-crash")
284 streams
= trace
["stream-infos"]
285 self
.assertEqual(len(streams
), 1)
287 self
.assertEqual(streams
[0]['range-ns']['begin'], 1565891729288866738)
288 self
.assertEqual(streams
[0]['range-ns']['end'], 1565891729293526525)
291 if __name__
== '__main__':
This page took 0.048218 seconds and 3 git commands to generate.