5ecec1356d16640dfddcdb48d06a5b3059a5510a
[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 from pathlib import PureWindowsPath, PurePosixPath
22
23
24 test_ctf_traces_path = os.environ['BT_CTF_TRACES_PATH']
25
26
27 # Key to sort streams in a predictable order.
28 def sort_predictably(stream):
29 if 'range-ns' in stream:
30 return stream['range-ns']['begin']
31 else:
32 return stream['paths'][0]
33
34
35 class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
36 def setUp(self):
37 ctf = bt2.find_plugin('ctf')
38 self._fs = ctf.source_component_classes['fs']
39
40 self._inputs = [
41 os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')
42 ]
43
44 def _check(self, trace, offset):
45 self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
46 self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
47 self.assertEqual(
48 trace['intersection-range-ns']['begin'], 13515309000000070 + offset
49 )
50 self.assertEqual(
51 trace['intersection-range-ns']['end'], 13515309000000100 + offset
52 )
53
54 streams = sorted(trace['streams'], key=sort_predictably)
55 self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
56 self.assertEqual(streams[0]['range-ns']['end'], 13515309000000100 + offset)
57 self.assertEqual(streams[1]['range-ns']['begin'], 13515309000000070 + offset)
58 self.assertEqual(streams[1]['range-ns']['end'], 13515309000000120 + offset)
59
60 # Test various cominations of the clock-class-offset-s and
61 # clock-class-offset-ns parameters to babeltrace.trace-info queries.
62
63 # Without clock class offset
64
65 def test_no_clock_class_offset(self):
66 res = bt2.QueryExecutor(
67 self._fs, 'babeltrace.trace-info', {'inputs': self._inputs}
68 ).query()
69 trace = res[0]
70 self._check(trace, 0)
71
72 # With clock-class-offset-s
73
74 def test_clock_class_offset_s(self):
75 res = bt2.QueryExecutor(
76 self._fs,
77 'babeltrace.trace-info',
78 {'inputs': self._inputs, 'clock-class-offset-s': 2},
79 ).query()
80 trace = res[0]
81 self._check(trace, 2000000000)
82
83 # With clock-class-offset-ns
84
85 def test_clock_class_offset_ns(self):
86 res = bt2.QueryExecutor(
87 self._fs,
88 'babeltrace.trace-info',
89 {'inputs': self._inputs, 'clock-class-offset-ns': 2},
90 ).query()
91 trace = res[0]
92 self._check(trace, 2)
93
94 # With both, negative
95
96 def test_clock_class_offset_both(self):
97 res = bt2.QueryExecutor(
98 self._fs,
99 'babeltrace.trace-info',
100 {
101 'inputs': self._inputs,
102 'clock-class-offset-s': -2,
103 'clock-class-offset-ns': -2,
104 },
105 ).query()
106 trace = res[0]
107 self._check(trace, -2000000002)
108
109 def test_clock_class_offset_s_wrong_type(self):
110 with self.assertRaises(bt2._Error):
111 bt2.QueryExecutor(
112 self._fs,
113 'babeltrace.trace-info',
114 {'inputs': self._inputs, 'clock-class-offset-s': "2"},
115 ).query()
116
117 def test_clock_class_offset_s_wrong_type_none(self):
118 with self.assertRaises(bt2._Error):
119 bt2.QueryExecutor(
120 self._fs,
121 'babeltrace.trace-info',
122 {'inputs': self._inputs, 'clock-class-offset-s': None},
123 ).query()
124
125 def test_clock_class_offset_ns_wrong_type(self):
126 with self.assertRaises(bt2._Error):
127 bt2.QueryExecutor(
128 self._fs,
129 'babeltrace.trace-info',
130 {'inputs': self._inputs, 'clock-class-offset-ns': "2"},
131 ).query()
132
133 def test_clock_class_offset_ns_wrong_type_none(self):
134 with self.assertRaises(bt2._Error):
135 bt2.QueryExecutor(
136 self._fs,
137 'babeltrace.trace-info',
138 {'inputs': self._inputs, 'clock-class-offset-ns': None},
139 ).query()
140
141
142 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
143 def setUp(self):
144 ctf = bt2.find_plugin("ctf")
145 self._fs = ctf.source_component_classes["fs"]
146
147 def test_trace_uuid_stream_class_id_no_stream_id(self):
148 res = bt2.QueryExecutor(
149 self._fs,
150 "babeltrace.trace-info",
151 {
152 "inputs": [
153 os.path.join(
154 test_ctf_traces_path, "intersection", "3eventsintersect"
155 )
156 ]
157 },
158 ).query()
159
160 os_stream_path = PurePosixPath(
161 '/tests/data/ctf-traces/intersection/3eventsintersect/'
162 )
163 if os.environ['BT_OS_TYPE'] == 'mingw':
164 os_stream_path = PureWindowsPath(os_stream_path)
165
166 self.assertEqual(len(res), 1)
167 trace = res[0]
168 streams = sorted(trace["streams"], key=sort_predictably)
169 self.assertEqual(len(streams), 2)
170 self.assertRegexpMatches(
171 str(streams[0]["port-name"]),
172 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
173 + re.escape(str(os_stream_path / "test_stream_0"))
174 + r"$",
175 )
176 self.assertRegexpMatches(
177 str(streams[1]["port-name"]),
178 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
179 + re.escape(str(os_stream_path / "test_stream_1"))
180 + r"$",
181 )
182
183 def test_trace_uuid_no_stream_class_id_no_stream_id(self):
184 res = bt2.QueryExecutor(
185 self._fs,
186 "babeltrace.trace-info",
187 {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
188 ).query()
189
190 os_stream_path = PurePosixPath(
191 '/tests/data/ctf-traces/succeed/succeed1/dummystream'
192 )
193 if os.environ['BT_OS_TYPE'] == 'mingw':
194 os_stream_path = PureWindowsPath(os_stream_path)
195
196 self.assertEqual(len(res), 1)
197 trace = res[0]
198 streams = sorted(trace["streams"], key=sort_predictably)
199 self.assertEqual(len(streams), 1)
200 self.assertRegexpMatches(
201 str(streams[0]["port-name"]),
202 r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
203 + re.escape(str(os_stream_path))
204 + r"$",
205 )
206
207
208 class QueryTraceInfoRangeTestCase(unittest.TestCase):
209 def setUp(self):
210 ctf = bt2.find_plugin("ctf")
211 self._fs = ctf.source_component_classes["fs"]
212
213 def test_trace_no_range(self):
214 # This trace has no `timestamp_begin` and `timestamp_end` in its
215 # packet context. The `babeltrace.trace-info` query should omit
216 # the `range-ns` fields in the `trace` and `stream` data
217 # structures.
218
219 res = bt2.QueryExecutor(
220 self._fs,
221 "babeltrace.trace-info",
222 {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
223 ).query()
224
225 self.assertEqual(len(res), 1)
226 trace = res[0]
227 streams = trace["streams"]
228 self.assertEqual(len(streams), 1)
229
230 self.assertRaises(KeyError, lambda: trace['range-ns'])
231 self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
232
233
234 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest.TestCase):
235 def setUp(self):
236 ctf = bt2.find_plugin('ctf')
237 self._fs = ctf.source_component_classes['fs']
238 self._path = os.path.join(test_ctf_traces_path, 'succeed')
239
240 def _test_lttng_quirks(self, trace_name):
241 res = bt2.QueryExecutor(
242 self._fs,
243 "babeltrace.trace-info",
244 {"inputs": [os.path.join(self._path, trace_name)]},
245 ).query()
246
247 self.assertEqual(len(res), 1)
248 return res[0]
249
250 def test_event_after_packet(self):
251 trace = self._test_lttng_quirks("lttng-event-after-packet")
252 streams = trace["streams"]
253 self.assertEqual(len(streams), 1)
254
255 self.assertEqual(streams[0]['range-ns']['begin'], 1565957300948091100)
256 self.assertEqual(streams[0]['range-ns']['end'], 1565957302180016069)
257
258 def test_lttng_crash(self):
259 trace = self._test_lttng_quirks("lttng-crash")
260 streams = trace["streams"]
261 self.assertEqual(len(streams), 1)
262
263 self.assertEqual(streams[0]['range-ns']['begin'], 1565891729288866738)
264 self.assertEqual(streams[0]['range-ns']['end'], 1565891729293526525)
265
266
267 if __name__ == '__main__':
268 unittest.main()
This page took 0.043482 seconds and 3 git commands to generate.