8d8cda783a58e7762fb61e478b44550d6436b25c
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 if 'range-ns' in stream
:
30 return stream
['range-ns']['begin']
32 return stream
['paths'][0]
35 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
37 ctf
= bt2
.find_plugin('ctf')
38 self
._fs
= ctf
.source_component_classes
['fs']
41 os
.path
.join(test_ctf_traces_path
, 'intersection', '3eventsintersect')
43 self
._executor
= bt2
.QueryExecutor()
45 def _check(self
, trace
, offset
):
46 self
.assertEqual(trace
['range-ns']['begin'], 13515309000000000 + offset
)
47 self
.assertEqual(trace
['range-ns']['end'], 13515309000000120 + offset
)
49 trace
['intersection-range-ns']['begin'], 13515309000000070 + offset
52 trace
['intersection-range-ns']['end'], 13515309000000100 + offset
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
)
61 # Test various cominations of the clock-class-offset-s and
62 # clock-class-offset-ns parameters to babeltrace.trace-info queries.
64 # Without clock class offset
66 def test_no_clock_class_offset(self
):
67 res
= self
._executor
.query(
68 self
._fs
, 'babeltrace.trace-info', {'inputs': self
._inputs
}
73 # With clock-class-offset-s
75 def test_clock_class_offset_s(self
):
76 res
= self
._executor
.query(
78 'babeltrace.trace-info',
79 {'inputs': self
._inputs
, 'clock-class-offset-s': 2},
82 self
._check
(trace
, 2000000000)
84 # With clock-class-offset-ns
86 def test_clock_class_offset_ns(self
):
87 res
= self
._executor
.query(
89 'babeltrace.trace-info',
90 {'inputs': self
._inputs
, 'clock-class-offset-ns': 2},
97 def test_clock_class_offset_both(self
):
98 res
= self
._executor
.query(
100 'babeltrace.trace-info',
102 'inputs': self
._inputs
,
103 'clock-class-offset-s': -2,
104 'clock-class-offset-ns': -2,
108 self
._check
(trace
, -2000000002)
110 def test_clock_class_offset_s_wrong_type(self
):
111 with self
.assertRaises(bt2
._Error
):
112 self
._executor
.query(
114 'babeltrace.trace-info',
115 {'inputs': self
._inputs
, 'clock-class-offset-s': "2"},
118 def test_clock_class_offset_s_wrong_type_none(self
):
119 with self
.assertRaises(bt2
._Error
):
120 self
._executor
.query(
122 'babeltrace.trace-info',
123 {'inputs': self
._inputs
, 'clock-class-offset-s': None},
126 def test_clock_class_offset_ns_wrong_type(self
):
127 with self
.assertRaises(bt2
._Error
):
128 self
._executor
.query(
130 'babeltrace.trace-info',
131 {'inputs': self
._inputs
, 'clock-class-offset-ns': "2"},
134 def test_clock_class_offset_ns_wrong_type_none(self
):
135 with self
.assertRaises(bt2
._Error
):
136 self
._executor
.query(
138 'babeltrace.trace-info',
139 {'inputs': self
._inputs
, 'clock-class-offset-ns': None},
143 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
145 ctf
= bt2
.find_plugin("ctf")
146 self
._fs
= ctf
.source_component_classes
["fs"]
148 self
._executor
= bt2
.QueryExecutor()
150 def test_trace_uuid_stream_class_id_no_stream_id(self
):
151 res
= self
._executor
.query(
153 "babeltrace.trace-info",
157 test_ctf_traces_path
, "intersection", "3eventsintersect"
163 os_stream_path
= PurePosixPath(
164 '/tests/data/ctf-traces/intersection/3eventsintersect/'
166 if os
.environ
['BT_OS_TYPE'] == 'mingw':
167 os_stream_path
= PureWindowsPath(os_stream_path
)
169 self
.assertEqual(len(res
), 1)
171 streams
= sorted(trace
["streams"], key
=sort_predictably
)
172 self
.assertEqual(len(streams
), 2)
173 self
.assertRegexpMatches(
174 str(streams
[0]["port-name"]),
175 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
176 + re
.escape(str(os_stream_path
/ "test_stream_0"))
179 self
.assertRegexpMatches(
180 str(streams
[1]["port-name"]),
181 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
182 + re
.escape(str(os_stream_path
/ "test_stream_1"))
186 def test_trace_uuid_no_stream_class_id_no_stream_id(self
):
187 res
= self
._executor
.query(
189 "babeltrace.trace-info",
190 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
193 os_stream_path
= PurePosixPath(
194 '/tests/data/ctf-traces/succeed/succeed1/dummystream'
196 if os
.environ
['BT_OS_TYPE'] == 'mingw':
197 os_stream_path
= PureWindowsPath(os_stream_path
)
199 self
.assertEqual(len(res
), 1)
201 streams
= sorted(trace
["streams"], key
=sort_predictably
)
202 self
.assertEqual(len(streams
), 1)
203 self
.assertRegexpMatches(
204 str(streams
[0]["port-name"]),
205 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
206 + re
.escape(str(os_stream_path
))
211 class QueryTraceInfoRangeTestCase(unittest
.TestCase
):
213 ctf
= bt2
.find_plugin("ctf")
214 self
._fs
= ctf
.source_component_classes
["fs"]
216 self
._executor
= bt2
.QueryExecutor()
218 def test_trace_no_range(self
):
219 # This trace has no `timestamp_begin` and `timestamp_end` in its
220 # packet context. The `babeltrace.trace-info` query should omit
221 # the `range-ns` fields in the `trace` and `stream` data
224 res
= self
._executor
.query(
226 "babeltrace.trace-info",
227 {"inputs": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
230 self
.assertEqual(len(res
), 1)
232 streams
= trace
["streams"]
233 self
.assertEqual(len(streams
), 1)
235 self
.assertRaises(KeyError, lambda: trace
['range-ns'])
236 self
.assertRaises(KeyError, lambda: streams
[0]['range-ns'])
239 if __name__
== '__main__':
This page took 0.035845 seconds and 4 git commands to generate.