ff8264a349fba1d32ae7edcd3c0620124db9839e
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.
21 from pathlib
import PureWindowsPath
, PurePosixPath
24 test_ctf_traces_path
= os
.environ
['BT_CTF_TRACES_PATH']
27 # Key to sort streams in a predictable order.
28 def sort_predictably(stream
):
29 return stream
['port-name']
32 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
34 ctf
= bt2
.find_plugin('ctf')
35 self
._fs
= ctf
.source_component_classes
['fs']
38 os
.path
.join(test_ctf_traces_path
, 'intersection', '3eventsintersect')
41 def _check(self
, trace
, offset
):
42 streams
= sorted(trace
['stream-infos'], key
=sort_predictably
)
43 self
.assertEqual(streams
[0]['range-ns']['begin'], 13515309000000000 + offset
)
44 self
.assertEqual(streams
[0]['range-ns']['end'], 13515309000000100 + offset
)
45 self
.assertEqual(streams
[1]['range-ns']['begin'], 13515309000000070 + offset
)
46 self
.assertEqual(streams
[1]['range-ns']['end'], 13515309000000120 + offset
)
48 # Test various cominations of the clock-class-offset-s and
49 # clock-class-offset-ns parameters to babeltrace.trace-infos queries.
51 # Without clock class offset
53 def test_no_clock_class_offset(self
):
54 res
= bt2
.QueryExecutor(
55 self
._fs
, 'babeltrace.trace-infos', {'inputs': self
._inputs
}
60 # With clock-class-offset-s
62 def test_clock_class_offset_s(self
):
63 res
= bt2
.QueryExecutor(
65 'babeltrace.trace-infos',
66 {'inputs': self
._inputs
, 'clock-class-offset-s': 2},
69 self
._check
(trace
, 2000000000)
71 # With clock-class-offset-ns
73 def test_clock_class_offset_ns(self
):
74 res
= bt2
.QueryExecutor(
76 'babeltrace.trace-infos',
77 {'inputs': self
._inputs
, 'clock-class-offset-ns': 2},
84 def test_clock_class_offset_both(self
):
85 res
= bt2
.QueryExecutor(
87 'babeltrace.trace-infos',
89 'inputs': self
._inputs
,
90 'clock-class-offset-s': -2,
91 'clock-class-offset-ns': -2,
95 self
._check
(trace
, -2000000002)
97 def test_clock_class_offset_s_wrong_type(self
):
98 with self
.assertRaises(bt2
._Error
):
101 'babeltrace.trace-infos',
102 {'inputs': self
._inputs
, 'clock-class-offset-s': "2"},
105 def test_clock_class_offset_s_wrong_type_none(self
):
106 with self
.assertRaises(bt2
._Error
):
109 'babeltrace.trace-infos',
110 {'inputs': self
._inputs
, 'clock-class-offset-s': None},
113 def test_clock_class_offset_ns_wrong_type(self
):
114 with self
.assertRaises(bt2
._Error
):
117 'babeltrace.trace-infos',
118 {'inputs': self
._inputs
, 'clock-class-offset-ns': "2"},
121 def test_clock_class_offset_ns_wrong_type_none(self
):
122 with self
.assertRaises(bt2
._Error
):
125 'babeltrace.trace-infos',
126 {'inputs': self
._inputs
, 'clock-class-offset-ns': None},
130 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
132 ctf
= bt2
.find_plugin("ctf")
133 self
._fs
= ctf
.source_component_classes
["fs"]
135 def test_trace_uuid_stream_class_id_no_stream_id(self
):
136 res
= bt2
.QueryExecutor(
138 "babeltrace.trace-infos",
142 test_ctf_traces_path
, "intersection", "3eventsintersect"
148 os_stream_path
= PurePosixPath(
149 '/tests/data/ctf-traces/intersection/3eventsintersect/'
151 if os
.environ
['BT_OS_TYPE'] == 'mingw':
152 os_stream_path
= PureWindowsPath(os_stream_path
)
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(str(os_stream_path
/ "test_stream_0"))
165 str(streams
[1]["port-name"]),
166 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
167 + re
.escape(str(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 os_stream_path
= PurePosixPath(
179 '/tests/data/ctf-traces/succeed/succeed1/dummystream'
181 if os
.environ
['BT_OS_TYPE'] == 'mingw':
182 os_stream_path
= PureWindowsPath(os_stream_path
)
184 self
.assertEqual(len(res
), 1)
186 streams
= sorted(trace
["stream-infos"], key
=sort_predictably
)
187 self
.assertEqual(len(streams
), 1)
189 str(streams
[0]["port-name"]),
190 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
191 + re
.escape(str(os_stream_path
))
196 class QueryTraceInfoRangeTestCase(unittest
.TestCase
):
198 ctf
= bt2
.find_plugin("ctf")
199 self
._fs
= ctf
.source_component_classes
["fs"]
201 def test_trace_no_range(self
):
202 # This trace has no `timestamp_begin` and `timestamp_end` in its
203 # packet context. The `babeltrace.trace-infos` query should omit
204 # the `range-ns` fields in the `trace` and `stream` data
207 res
= bt2
.QueryExecutor(
209 "babeltrace.trace-infos",
210 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
213 self
.assertEqual(len(res
), 1)
215 streams
= trace
["stream-infos"]
216 self
.assertEqual(len(streams
), 1)
218 self
.assertRaises(KeyError, lambda: trace
['range-ns'])
219 self
.assertRaises(KeyError, lambda: streams
[0]['range-ns'])
221 def test_trace_with_tracefile_rotation(self
):
222 res
= bt2
.QueryExecutor(
224 "babeltrace.trace-infos",
228 test_ctf_traces_path
,
230 "lttng-tracefile-rotation",
237 self
.assertEqual(len(res
), 1)
239 streams
= trace
["stream-infos"]
240 self
.assertEqual(len(streams
), 4)
242 # Note: the end timestamps are not the end timestamps found in the
243 # index files, because fix_index_lttng_event_after_packet_bug changes
244 # them based on the time of the last event in the stream.
246 self
.assertEqual(streams
[0]['range-ns']['begin'], 1571261795455986789)
247 self
.assertEqual(streams
[0]['range-ns']['end'], 1571261797582611840)
249 self
.assertEqual(streams
[1]['range-ns']['begin'], 1571261795456368232)
250 self
.assertEqual(streams
[1]['range-ns']['end'], 1571261797577754111)
252 self
.assertEqual(streams
[2]['range-ns']['begin'], 1571261795456748255)
253 self
.assertEqual(streams
[2]['range-ns']['end'], 1571261797577727795)
255 self
.assertEqual(streams
[3]['range-ns']['begin'], 1571261795457285142)
256 self
.assertEqual(streams
[3]['range-ns']['end'], 1571261797582522088)
259 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest
.TestCase
):
261 ctf
= bt2
.find_plugin('ctf')
262 self
._fs
= ctf
.source_component_classes
['fs']
263 self
._path
= os
.path
.join(test_ctf_traces_path
, 'succeed')
265 def _test_lttng_quirks(self
, trace_name
):
266 res
= bt2
.QueryExecutor(
268 "babeltrace.trace-infos",
269 {"inputs": [os
.path
.join(self
._path
, trace_name
)]},
272 self
.assertEqual(len(res
), 1)
275 def test_event_after_packet(self
):
276 trace
= self
._test
_lttng
_quirks
("lttng-event-after-packet")
277 streams
= trace
["stream-infos"]
278 self
.assertEqual(len(streams
), 1)
280 self
.assertEqual(streams
[0]['range-ns']['begin'], 1565957300948091100)
281 self
.assertEqual(streams
[0]['range-ns']['end'], 1565957302180016069)
283 def test_lttng_crash(self
):
284 trace
= self
._test
_lttng
_quirks
("lttng-crash")
285 streams
= trace
["stream-infos"]
286 self
.assertEqual(len(streams
), 1)
288 self
.assertEqual(streams
[0]['range-ns']['begin'], 1565891729288866738)
289 self
.assertEqual(streams
[0]['range-ns']['end'], 1565891729293526525)
292 if __name__
== '__main__':
This page took 0.035931 seconds and 4 git commands to generate.