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_TESTS_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_TESTS_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.040813 seconds and 4 git commands to generate.