e0e716b21b73a2fda3b9392bd8d4fb18ab0e2818
[babeltrace.git] / tests / plugins / src.ctf.fs / query / test_query_trace_info.py
1 # Copyright (C) 2019 Simon Marchi <simon.marchi@efficios.com>
2 #
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
6 # of the License.
7 #
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.
12 #
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.
16
17 import unittest
18 import bt2
19 import os
20 import re
21
22
23 test_ctf_traces_path = os.environ['BT_CTF_TRACES_PATH']
24
25
26 # Key to sort streams in a predictable order.
27 def sort_predictably(stream):
28 return stream['port-name']
29
30
31 class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
32 def setUp(self):
33 ctf = bt2.find_plugin('ctf')
34 self._fs = ctf.source_component_classes['fs']
35
36 self._inputs = [
37 os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')
38 ]
39
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)
46
47 # Test various cominations of the clock-class-offset-s and
48 # clock-class-offset-ns parameters to babeltrace.trace-infos queries.
49
50 # Without clock class offset
51
52 def test_no_clock_class_offset(self):
53 res = bt2.QueryExecutor(
54 self._fs, 'babeltrace.trace-infos', {'inputs': self._inputs}
55 ).query()
56 trace = res[0]
57 self._check(trace, 0)
58
59 # With clock-class-offset-s
60
61 def test_clock_class_offset_s(self):
62 res = bt2.QueryExecutor(
63 self._fs,
64 'babeltrace.trace-infos',
65 {'inputs': self._inputs, 'clock-class-offset-s': 2},
66 ).query()
67 trace = res[0]
68 self._check(trace, 2000000000)
69
70 # With clock-class-offset-ns
71
72 def test_clock_class_offset_ns(self):
73 res = bt2.QueryExecutor(
74 self._fs,
75 'babeltrace.trace-infos',
76 {'inputs': self._inputs, 'clock-class-offset-ns': 2},
77 ).query()
78 trace = res[0]
79 self._check(trace, 2)
80
81 # With both, negative
82
83 def test_clock_class_offset_both(self):
84 res = bt2.QueryExecutor(
85 self._fs,
86 'babeltrace.trace-infos',
87 {
88 'inputs': self._inputs,
89 'clock-class-offset-s': -2,
90 'clock-class-offset-ns': -2,
91 },
92 ).query()
93 trace = res[0]
94 self._check(trace, -2000000002)
95
96 def test_clock_class_offset_s_wrong_type(self):
97 with self.assertRaises(bt2._Error):
98 bt2.QueryExecutor(
99 self._fs,
100 'babeltrace.trace-infos',
101 {'inputs': self._inputs, 'clock-class-offset-s': "2"},
102 ).query()
103
104 def test_clock_class_offset_s_wrong_type_none(self):
105 with self.assertRaises(bt2._Error):
106 bt2.QueryExecutor(
107 self._fs,
108 'babeltrace.trace-infos',
109 {'inputs': self._inputs, 'clock-class-offset-s': None},
110 ).query()
111
112 def test_clock_class_offset_ns_wrong_type(self):
113 with self.assertRaises(bt2._Error):
114 bt2.QueryExecutor(
115 self._fs,
116 'babeltrace.trace-infos',
117 {'inputs': self._inputs, 'clock-class-offset-ns': "2"},
118 ).query()
119
120 def test_clock_class_offset_ns_wrong_type_none(self):
121 with self.assertRaises(bt2._Error):
122 bt2.QueryExecutor(
123 self._fs,
124 'babeltrace.trace-infos',
125 {'inputs': self._inputs, 'clock-class-offset-ns': None},
126 ).query()
127
128
129 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
130 def setUp(self):
131 ctf = bt2.find_plugin("ctf")
132 self._fs = ctf.source_component_classes["fs"]
133
134 def test_trace_uuid_stream_class_id_no_stream_id(self):
135 res = bt2.QueryExecutor(
136 self._fs,
137 "babeltrace.trace-infos",
138 {
139 "inputs": [
140 os.path.join(
141 test_ctf_traces_path, "intersection", "3eventsintersect"
142 )
143 ]
144 },
145 ).query()
146
147 if os.environ['BT_OS_TYPE'] == 'mingw':
148 os_stream_path = (
149 '\\tests\\data\\ctf-traces\\intersection\\3eventsintersect\\'
150 )
151 else:
152 os_stream_path = '/tests/data/ctf-traces/intersection/3eventsintersect/'
153
154 self.assertEqual(len(res), 1)
155 trace = res[0]
156 streams = sorted(trace["stream-infos"], key=sort_predictably)
157 self.assertEqual(len(streams), 2)
158 self.assertRegex(
159 str(streams[0]["port-name"]),
160 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
161 + re.escape(os_stream_path + "test_stream_0")
162 + r"$",
163 )
164 self.assertRegex(
165 str(streams[1]["port-name"]),
166 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
167 + re.escape(os_stream_path + "test_stream_1")
168 + r"$",
169 )
170
171 def test_trace_uuid_no_stream_class_id_no_stream_id(self):
172 res = bt2.QueryExecutor(
173 self._fs,
174 "babeltrace.trace-infos",
175 {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
176 ).query()
177
178 if os.environ['BT_OS_TYPE'] == 'mingw':
179 os_stream_path = '\\tests\\data\\ctf-traces\\succeed\\succeed1\\dummystream'
180 else:
181 os_stream_path = '/tests/data/ctf-traces/succeed/succeed1/dummystream'
182
183 self.assertEqual(len(res), 1)
184 trace = res[0]
185 streams = sorted(trace["stream-infos"], key=sort_predictably)
186 self.assertEqual(len(streams), 1)
187 self.assertRegex(
188 str(streams[0]["port-name"]),
189 r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
190 + re.escape(os_stream_path)
191 + r"$",
192 )
193
194
195 class QueryTraceInfoRangeTestCase(unittest.TestCase):
196 def setUp(self):
197 ctf = bt2.find_plugin("ctf")
198 self._fs = ctf.source_component_classes["fs"]
199
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
204 # structures.
205
206 res = bt2.QueryExecutor(
207 self._fs,
208 "babeltrace.trace-infos",
209 {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
210 ).query()
211
212 self.assertEqual(len(res), 1)
213 trace = res[0]
214 streams = trace["stream-infos"]
215 self.assertEqual(len(streams), 1)
216
217 self.assertRaises(KeyError, lambda: trace['range-ns'])
218 self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
219
220 def test_trace_with_tracefile_rotation(self):
221 res = bt2.QueryExecutor(
222 self._fs,
223 "babeltrace.trace-infos",
224 {
225 "inputs": [
226 os.path.join(
227 test_ctf_traces_path,
228 "succeed",
229 "lttng-tracefile-rotation",
230 "kernel",
231 )
232 ]
233 },
234 ).query()
235
236 self.assertEqual(len(res), 1)
237 trace = res[0]
238 streams = trace["stream-infos"]
239 self.assertEqual(len(streams), 4)
240
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.
244
245 self.assertEqual(streams[0]['range-ns']['begin'], 1571261795455986789)
246 self.assertEqual(streams[0]['range-ns']['end'], 1571261797582611840)
247
248 self.assertEqual(streams[1]['range-ns']['begin'], 1571261795456368232)
249 self.assertEqual(streams[1]['range-ns']['end'], 1571261797577754111)
250
251 self.assertEqual(streams[2]['range-ns']['begin'], 1571261795456748255)
252 self.assertEqual(streams[2]['range-ns']['end'], 1571261797577727795)
253
254 self.assertEqual(streams[3]['range-ns']['begin'], 1571261795457285142)
255 self.assertEqual(streams[3]['range-ns']['end'], 1571261797582522088)
256
257
258 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest.TestCase):
259 def setUp(self):
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')
263
264 def _test_lttng_quirks(self, trace_name):
265 res = bt2.QueryExecutor(
266 self._fs,
267 "babeltrace.trace-infos",
268 {"inputs": [os.path.join(self._path, trace_name)]},
269 ).query()
270
271 self.assertEqual(len(res), 1)
272 return res[0]
273
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)
278
279 self.assertEqual(streams[0]['range-ns']['begin'], 1565957300948091100)
280 self.assertEqual(streams[0]['range-ns']['end'], 1565957302180016069)
281
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)
286
287 self.assertEqual(streams[0]['range-ns']['begin'], 1565891729288866738)
288 self.assertEqual(streams[0]['range-ns']['end'], 1565891729293526525)
289
290
291 if __name__ == '__main__':
292 unittest.main()
This page took 0.048218 seconds and 3 git commands to generate.