`test_query_trace_info.py`: adapt regex to NT path style
[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._paths = [
41 os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')
42 ]
43 self._executor = bt2.QueryExecutor()
44
45 def _check(self, trace, offset):
46 self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
47 self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
48 self.assertEqual(
49 trace['intersection-range-ns']['begin'], 13515309000000070 + offset
50 )
51 self.assertEqual(
52 trace['intersection-range-ns']['end'], 13515309000000100 + offset
53 )
54
55 streams = sorted(trace['streams'], key=sort_predictably)
56 self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
57 self.assertEqual(streams[0]['range-ns']['end'], 13515309000000100 + offset)
58 self.assertEqual(streams[1]['range-ns']['begin'], 13515309000000070 + offset)
59 self.assertEqual(streams[1]['range-ns']['end'], 13515309000000120 + offset)
60
61 # Test various cominations of the clock-class-offset-s and
62 # clock-class-offset-ns parameters to trace-info queries.
63
64 # Without clock class offset
65
66 def test_no_clock_class_offset(self):
67 res = self._executor.query(self._fs, 'trace-info', {'paths': self._paths})
68 trace = res[0]
69 self._check(trace, 0)
70
71 # With clock-class-offset-s
72
73 def test_clock_class_offset_s(self):
74 res = self._executor.query(
75 self._fs, 'trace-info', {'paths': self._paths, 'clock-class-offset-s': 2}
76 )
77 trace = res[0]
78 self._check(trace, 2000000000)
79
80 # With clock-class-offset-ns
81
82 def test_clock_class_offset_ns(self):
83 res = self._executor.query(
84 self._fs, 'trace-info', {'paths': self._paths, 'clock-class-offset-ns': 2}
85 )
86 trace = res[0]
87 self._check(trace, 2)
88
89 # With both, negative
90
91 def test_clock_class_offset_both(self):
92 res = self._executor.query(
93 self._fs,
94 'trace-info',
95 {
96 'paths': self._paths,
97 'clock-class-offset-s': -2,
98 'clock-class-offset-ns': -2,
99 },
100 )
101 trace = res[0]
102 self._check(trace, -2000000002)
103
104 def test_clock_class_offset_s_wrong_type(self):
105 with self.assertRaises(bt2.InvalidParams):
106 self._executor.query(
107 self._fs,
108 'trace-info',
109 {'paths': self._paths, 'clock-class-offset-s': "2"},
110 )
111
112 def test_clock_class_offset_s_wrong_type_none(self):
113 with self.assertRaises(bt2.InvalidParams):
114 self._executor.query(
115 self._fs,
116 'trace-info',
117 {'paths': self._paths, 'clock-class-offset-s': None},
118 )
119
120 def test_clock_class_offset_ns_wrong_type(self):
121 with self.assertRaises(bt2.InvalidParams):
122 self._executor.query(
123 self._fs,
124 'trace-info',
125 {'paths': self._paths, 'clock-class-offset-ns': "2"},
126 )
127
128 def test_clock_class_offset_ns_wrong_type_none(self):
129 with self.assertRaises(bt2.InvalidParams):
130 self._executor.query(
131 self._fs,
132 'trace-info',
133 {'paths': self._paths, 'clock-class-offset-ns': None},
134 )
135
136
137 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
138 def setUp(self):
139 ctf = bt2.find_plugin("ctf")
140 self._fs = ctf.source_component_classes["fs"]
141
142 self._executor = bt2.QueryExecutor()
143
144 def test_trace_uuid_stream_class_id_no_stream_id(self):
145 res = self._executor.query(
146 self._fs,
147 "trace-info",
148 {
149 "paths": [
150 os.path.join(
151 test_ctf_traces_path, "intersection", "3eventsintersect"
152 )
153 ]
154 },
155 )
156
157 os_stream_path = PurePosixPath(
158 '/tests/data/ctf-traces/intersection/3eventsintersect/'
159 )
160 if os.environ['BT_OS_TYPE'] == 'mingw':
161 os_stream_path = PureWindowsPath(os_stream_path)
162
163 self.assertEqual(len(res), 1)
164 trace = res[0]
165 streams = sorted(trace["streams"], key=sort_predictably)
166 self.assertEqual(len(streams), 2)
167 self.assertRegexpMatches(
168 str(streams[0]["port-name"]),
169 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
170 + re.escape(str(os_stream_path / "test_stream_0"))
171 + r"$",
172 )
173 self.assertRegexpMatches(
174 str(streams[1]["port-name"]),
175 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
176 + re.escape(str(os_stream_path / "test_stream_1"))
177 + r"$",
178 )
179
180 def test_trace_uuid_no_stream_class_id_no_stream_id(self):
181 res = self._executor.query(
182 self._fs,
183 "trace-info",
184 {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
185 )
186
187 os_stream_path = PurePosixPath(
188 '/tests/data/ctf-traces/succeed/succeed1/dummystream'
189 )
190 if os.environ['BT_OS_TYPE'] == 'mingw':
191 os_stream_path = PureWindowsPath(os_stream_path)
192
193 self.assertEqual(len(res), 1)
194 trace = res[0]
195 streams = sorted(trace["streams"], key=sort_predictably)
196 self.assertEqual(len(streams), 1)
197 self.assertRegexpMatches(
198 str(streams[0]["port-name"]),
199 r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
200 + re.escape(str(os_stream_path))
201 + r"$",
202 )
203
204
205 class QueryTraceInfoRangeTestCase(unittest.TestCase):
206 def setUp(self):
207 ctf = bt2.find_plugin("ctf")
208 self._fs = ctf.source_component_classes["fs"]
209
210 self._executor = bt2.QueryExecutor()
211
212 def test_trace_no_range(self):
213 # This trace has no `timestamp_begin` and `timestamp_end` in its packet
214 # context. The `trace-info` query should omit the `range-ns` fields in
215 # the `trace` and `stream` data structures.
216
217 res = self._executor.query(
218 self._fs,
219 "trace-info",
220 {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
221 )
222
223 self.assertEqual(len(res), 1)
224 trace = res[0]
225 streams = trace["streams"]
226 self.assertEqual(len(streams), 1)
227
228 self.assertRaises(KeyError, lambda: trace['range-ns'])
229 self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
230
231
232 if __name__ == '__main__':
233 unittest.main()
This page took 0.034213 seconds and 4 git commands to generate.