1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 Simon Marchi <simon.marchi@efficios.com>
11 from test_all_ctf_versions
import test_all_ctf_versions
13 test_ctf_traces_path
= os
.environ
["BT_CTF_TRACES_PATH"]
16 # Key to sort streams in a predictable order.
17 def sort_predictably(stream
):
18 return stream
["port-name"]
21 @test_all_ctf_versions
22 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
24 ctf
= bt2
.find_plugin("ctf")
25 self
._fs
= ctf
.source_component_classes
["fs"]
32 str(self
._ctf
_version
),
38 def _check(self
, trace
, offset
):
39 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
40 self
.assertEqual(streams
[0]["range-ns"]["begin"], 13515309000000000 + offset
)
41 self
.assertEqual(streams
[0]["range-ns"]["end"], 13515309000000100 + offset
)
42 self
.assertEqual(streams
[1]["range-ns"]["begin"], 13515309000000070 + offset
)
43 self
.assertEqual(streams
[1]["range-ns"]["end"], 13515309000000120 + offset
)
45 # Test various cominations of the clock-class-offset-s and
46 # clock-class-offset-ns parameters to babeltrace.trace-infos queries.
48 # Without clock class offset
50 def test_no_clock_class_offset(self
):
51 res
= bt2
.QueryExecutor(
52 self
._fs
, "babeltrace.trace-infos", {"inputs": self
._inputs
}
57 # With clock-class-offset-s
59 def test_clock_class_offset_s(self
):
60 res
= bt2
.QueryExecutor(
62 "babeltrace.trace-infos",
63 {"inputs": self
._inputs
, "clock-class-offset-s": 2},
66 self
._check
(trace
, 2000000000)
68 # With clock-class-offset-ns
70 def test_clock_class_offset_ns(self
):
71 res
= bt2
.QueryExecutor(
73 "babeltrace.trace-infos",
74 {"inputs": self
._inputs
, "clock-class-offset-ns": 2},
81 def test_clock_class_offset_both(self
):
82 res
= bt2
.QueryExecutor(
84 "babeltrace.trace-infos",
86 "inputs": self
._inputs
,
87 "clock-class-offset-s": -2,
88 "clock-class-offset-ns": -2,
92 self
._check
(trace
, -2000000002)
94 def test_clock_class_offset_s_wrong_type(self
):
95 with self
.assertRaises(bt2
._Error
):
98 "babeltrace.trace-infos",
99 {"inputs": self
._inputs
, "clock-class-offset-s": "2"},
102 def test_clock_class_offset_s_wrong_type_none(self
):
103 with self
.assertRaises(bt2
._Error
):
106 "babeltrace.trace-infos",
107 {"inputs": self
._inputs
, "clock-class-offset-s": None},
110 def test_clock_class_offset_ns_wrong_type(self
):
111 with self
.assertRaises(bt2
._Error
):
114 "babeltrace.trace-infos",
115 {"inputs": self
._inputs
, "clock-class-offset-ns": "2"},
118 def test_clock_class_offset_ns_wrong_type_none(self
):
119 with self
.assertRaises(bt2
._Error
):
122 "babeltrace.trace-infos",
123 {"inputs": self
._inputs
, "clock-class-offset-ns": None},
127 @test_all_ctf_versions
128 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
130 ctf
= bt2
.find_plugin("ctf")
131 self
._fs
= ctf
.source_component_classes
["fs"]
133 def test_trace_uuid_stream_class_id_no_stream_id(self
):
134 res
= bt2
.QueryExecutor(
136 "babeltrace.trace-infos",
140 test_ctf_traces_path
,
141 str(self
._ctf
_version
),
150 "\\tests\\data\\ctf-traces\\{}\\intersection\\3eventsintersect\\"
151 if os
.environ
["BT_TESTS_OS_TYPE"] == "mingw"
152 else "/tests/data/ctf-traces/{}/intersection/3eventsintersect/"
153 ).format(self
._ctf
_version
)
155 self
.assertEqual(len(res
), 1)
157 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
158 self
.assertEqual(len(streams
), 2)
160 str(streams
[0]["port-name"]),
161 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
162 + re
.escape(os_stream_path
+ "test_stream_0")
166 str(streams
[1]["port-name"]),
167 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
168 + re
.escape(os_stream_path
+ "test_stream_1")
172 def test_trace_uuid_no_stream_class_id_no_stream_id(self
):
173 res
= bt2
.QueryExecutor(
175 "babeltrace.trace-infos",
178 os
.path
.join(test_ctf_traces_path
, "1", "succeed", "succeed1")
183 if os
.environ
["BT_TESTS_OS_TYPE"] == "mingw":
185 "\\tests\\data\\ctf-traces\\1\\succeed\\succeed1\\dummystream"
188 os_stream_path
= "/tests/data/ctf-traces/1/succeed/succeed1/dummystream"
190 self
.assertEqual(len(res
), 1)
192 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
193 self
.assertEqual(len(streams
), 1)
195 str(streams
[0]["port-name"]),
196 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
197 + re
.escape(os_stream_path
)
202 @test_all_ctf_versions
203 class QueryTraceInfoRangeTestCase(unittest
.TestCase
):
205 ctf
= bt2
.find_plugin("ctf")
206 self
._fs
= ctf
.source_component_classes
["fs"]
208 def test_trace_no_range(self
):
209 # This trace has no `timestamp_begin` and `timestamp_end` in its
210 # packet context. The `babeltrace.trace-infos` query should omit
211 # the `range-ns` fields in the `trace` and `stream` data
214 res
= bt2
.QueryExecutor(
216 "babeltrace.trace-infos",
220 test_ctf_traces_path
,
221 str(self
._ctf
_version
),
229 self
.assertEqual(len(res
), 1)
231 streams
= trace
["stream-infos"]
232 self
.assertEqual(len(streams
), 1)
234 self
.assertRaises(KeyError, lambda: trace
["range-ns"])
235 self
.assertRaises(KeyError, lambda: streams
[0]["range-ns"])
237 def test_trace_with_tracefile_rotation(self
):
238 res
= bt2
.QueryExecutor(
240 "babeltrace.trace-infos",
244 test_ctf_traces_path
,
245 str(self
._ctf
_version
),
247 "lttng-tracefile-rotation",
254 self
.assertEqual(len(res
), 1)
256 streams
= trace
["stream-infos"]
257 self
.assertEqual(len(streams
), 4)
259 # Note: the end timestamps are not the end timestamps found in the
260 # index files, because fix_index_lttng_event_after_packet_bug changes
261 # them based on the time of the last event in the stream.
263 self
.assertEqual(streams
[0]["range-ns"]["begin"], 1571261795455986789)
264 self
.assertEqual(streams
[0]["range-ns"]["end"], 1571261797582611840)
266 self
.assertEqual(streams
[1]["range-ns"]["begin"], 1571261795456368232)
267 self
.assertEqual(streams
[1]["range-ns"]["end"], 1571261797577754111)
269 self
.assertEqual(streams
[2]["range-ns"]["begin"], 1571261795456748255)
270 self
.assertEqual(streams
[2]["range-ns"]["end"], 1571261797577727795)
272 self
.assertEqual(streams
[3]["range-ns"]["begin"], 1571261795457285142)
273 self
.assertEqual(streams
[3]["range-ns"]["end"], 1571261797582522088)
276 @test_all_ctf_versions
277 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest
.TestCase
):
279 ctf
= bt2
.find_plugin("ctf")
280 self
._fs
= ctf
.source_component_classes
["fs"]
282 def _trace_path(self
, trace_name
):
284 test_ctf_traces_path
, str(self
._ctf
_version
), "succeed", trace_name
287 def _test_lttng_quirks(self
, trace_name
):
288 res
= bt2
.QueryExecutor(
290 "babeltrace.trace-infos",
291 {"inputs": [self
._trace
_path
(trace_name
)]},
294 self
.assertEqual(len(res
), 1)
297 def test_event_after_packet(self
):
298 trace
= self
._test
_lttng
_quirks
("lttng-event-after-packet")
299 streams
= trace
["stream-infos"]
300 self
.assertEqual(len(streams
), 1)
302 self
.assertEqual(streams
[0]["range-ns"]["begin"], 1565957300948091100)
303 self
.assertEqual(streams
[0]["range-ns"]["end"], 1565957302180016069)
305 def test_lttng_crash(self
):
306 trace
= self
._test
_lttng
_quirks
("lttng-crash")
307 streams
= trace
["stream-infos"]
308 self
.assertEqual(len(streams
), 1)
310 self
.assertEqual(streams
[0]["range-ns"]["begin"], 1565891729288866738)
311 self
.assertEqual(streams
[0]["range-ns"]["end"], 1565891729293526525)
314 if __name__
== "__main__":
This page took 0.04158 seconds and 5 git commands to generate.