1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 Simon Marchi <simon.marchi@efficios.com>
12 test_ctf_traces_path
= os
.environ
['BT_CTF_TRACES_PATH']
15 # Key to sort streams in a predictable order.
16 def sort_predictably(stream
):
17 return stream
['port-name']
20 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
22 ctf
= bt2
.find_plugin('ctf')
23 self
._fs
= ctf
.source_component_classes
['fs']
26 os
.path
.join(test_ctf_traces_path
, 'intersection', '3eventsintersect')
29 def _check(self
, trace
, offset
):
30 streams
= sorted(trace
['stream-infos'], key
=sort_predictably
)
31 self
.assertEqual(streams
[0]['range-ns']['begin'], 13515309000000000 + offset
)
32 self
.assertEqual(streams
[0]['range-ns']['end'], 13515309000000100 + offset
)
33 self
.assertEqual(streams
[1]['range-ns']['begin'], 13515309000000070 + offset
)
34 self
.assertEqual(streams
[1]['range-ns']['end'], 13515309000000120 + offset
)
36 # Test various cominations of the clock-class-offset-s and
37 # clock-class-offset-ns parameters to babeltrace.trace-infos queries.
39 # Without clock class offset
41 def test_no_clock_class_offset(self
):
42 res
= bt2
.QueryExecutor(
43 self
._fs
, 'babeltrace.trace-infos', {'inputs': self
._inputs
}
48 # With clock-class-offset-s
50 def test_clock_class_offset_s(self
):
51 res
= bt2
.QueryExecutor(
53 'babeltrace.trace-infos',
54 {'inputs': self
._inputs
, 'clock-class-offset-s': 2},
57 self
._check
(trace
, 2000000000)
59 # With clock-class-offset-ns
61 def test_clock_class_offset_ns(self
):
62 res
= bt2
.QueryExecutor(
64 'babeltrace.trace-infos',
65 {'inputs': self
._inputs
, 'clock-class-offset-ns': 2},
72 def test_clock_class_offset_both(self
):
73 res
= bt2
.QueryExecutor(
75 'babeltrace.trace-infos',
77 'inputs': self
._inputs
,
78 'clock-class-offset-s': -2,
79 'clock-class-offset-ns': -2,
83 self
._check
(trace
, -2000000002)
85 def test_clock_class_offset_s_wrong_type(self
):
86 with self
.assertRaises(bt2
._Error
):
89 'babeltrace.trace-infos',
90 {'inputs': self
._inputs
, 'clock-class-offset-s': "2"},
93 def test_clock_class_offset_s_wrong_type_none(self
):
94 with self
.assertRaises(bt2
._Error
):
97 'babeltrace.trace-infos',
98 {'inputs': self
._inputs
, 'clock-class-offset-s': None},
101 def test_clock_class_offset_ns_wrong_type(self
):
102 with self
.assertRaises(bt2
._Error
):
105 'babeltrace.trace-infos',
106 {'inputs': self
._inputs
, 'clock-class-offset-ns': "2"},
109 def test_clock_class_offset_ns_wrong_type_none(self
):
110 with self
.assertRaises(bt2
._Error
):
113 'babeltrace.trace-infos',
114 {'inputs': self
._inputs
, 'clock-class-offset-ns': None},
118 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
120 ctf
= bt2
.find_plugin("ctf")
121 self
._fs
= ctf
.source_component_classes
["fs"]
123 def test_trace_uuid_stream_class_id_no_stream_id(self
):
124 res
= bt2
.QueryExecutor(
126 "babeltrace.trace-infos",
130 test_ctf_traces_path
, "intersection", "3eventsintersect"
136 if os
.environ
['BT_OS_TYPE'] == 'mingw':
138 '\\tests\\data\\ctf-traces\\intersection\\3eventsintersect\\'
141 os_stream_path
= '/tests/data/ctf-traces/intersection/3eventsintersect/'
143 self
.assertEqual(len(res
), 1)
145 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
146 self
.assertEqual(len(streams
), 2)
148 str(streams
[0]["port-name"]),
149 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
150 + re
.escape(os_stream_path
+ "test_stream_0")
154 str(streams
[1]["port-name"]),
155 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
156 + re
.escape(os_stream_path
+ "test_stream_1")
160 def test_trace_uuid_no_stream_class_id_no_stream_id(self
):
161 res
= bt2
.QueryExecutor(
163 "babeltrace.trace-infos",
164 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
167 if os
.environ
['BT_OS_TYPE'] == 'mingw':
168 os_stream_path
= '\\tests\\data\\ctf-traces\\succeed\\succeed1\\dummystream'
170 os_stream_path
= '/tests/data/ctf-traces/succeed/succeed1/dummystream'
172 self
.assertEqual(len(res
), 1)
174 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
175 self
.assertEqual(len(streams
), 1)
177 str(streams
[0]["port-name"]),
178 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
179 + re
.escape(os_stream_path
)
184 class QueryTraceInfoRangeTestCase(unittest
.TestCase
):
186 ctf
= bt2
.find_plugin("ctf")
187 self
._fs
= ctf
.source_component_classes
["fs"]
189 def test_trace_no_range(self
):
190 # This trace has no `timestamp_begin` and `timestamp_end` in its
191 # packet context. The `babeltrace.trace-infos` query should omit
192 # the `range-ns` fields in the `trace` and `stream` data
195 res
= bt2
.QueryExecutor(
197 "babeltrace.trace-infos",
198 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
201 self
.assertEqual(len(res
), 1)
203 streams
= trace
["stream-infos"]
204 self
.assertEqual(len(streams
), 1)
206 self
.assertRaises(KeyError, lambda: trace
['range-ns'])
207 self
.assertRaises(KeyError, lambda: streams
[0]['range-ns'])
209 def test_trace_with_tracefile_rotation(self
):
210 res
= bt2
.QueryExecutor(
212 "babeltrace.trace-infos",
216 test_ctf_traces_path
,
218 "lttng-tracefile-rotation",
225 self
.assertEqual(len(res
), 1)
227 streams
= trace
["stream-infos"]
228 self
.assertEqual(len(streams
), 4)
230 # Note: the end timestamps are not the end timestamps found in the
231 # index files, because fix_index_lttng_event_after_packet_bug changes
232 # them based on the time of the last event in the stream.
234 self
.assertEqual(streams
[0]['range-ns']['begin'], 1571261795455986789)
235 self
.assertEqual(streams
[0]['range-ns']['end'], 1571261797582611840)
237 self
.assertEqual(streams
[1]['range-ns']['begin'], 1571261795456368232)
238 self
.assertEqual(streams
[1]['range-ns']['end'], 1571261797577754111)
240 self
.assertEqual(streams
[2]['range-ns']['begin'], 1571261795456748255)
241 self
.assertEqual(streams
[2]['range-ns']['end'], 1571261797577727795)
243 self
.assertEqual(streams
[3]['range-ns']['begin'], 1571261795457285142)
244 self
.assertEqual(streams
[3]['range-ns']['end'], 1571261797582522088)
247 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest
.TestCase
):
249 ctf
= bt2
.find_plugin('ctf')
250 self
._fs
= ctf
.source_component_classes
['fs']
251 self
._path
= os
.path
.join(test_ctf_traces_path
, 'succeed')
253 def _test_lttng_quirks(self
, trace_name
):
254 res
= bt2
.QueryExecutor(
256 "babeltrace.trace-infos",
257 {"inputs": [os
.path
.join(self
._path
, trace_name
)]},
260 self
.assertEqual(len(res
), 1)
263 def test_event_after_packet(self
):
264 trace
= self
._test
_lttng
_quirks
("lttng-event-after-packet")
265 streams
= trace
["stream-infos"]
266 self
.assertEqual(len(streams
), 1)
268 self
.assertEqual(streams
[0]['range-ns']['begin'], 1565957300948091100)
269 self
.assertEqual(streams
[0]['range-ns']['end'], 1565957302180016069)
271 def test_lttng_crash(self
):
272 trace
= self
._test
_lttng
_quirks
("lttng-crash")
273 streams
= trace
["stream-infos"]
274 self
.assertEqual(len(streams
), 1)
276 self
.assertEqual(streams
[0]['range-ns']['begin'], 1565891729288866738)
277 self
.assertEqual(streams
[0]['range-ns']['end'], 1565891729293526525)
280 if __name__
== '__main__':
This page took 0.04937 seconds and 4 git commands to generate.