Add facilities to test CLI and plugin regressions with expectation files
[babeltrace.git] / tests / plugins / src.ctf.fs / query / test_query_trace_info.py
CommitLineData
d907165c
SM
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
17import unittest
18import bt2
19import os
a38d7650 20import re
d907165c
SM
21
22
23test_ctf_traces_path = os.environ['TEST_CTF_TRACES_PATH']
24
25
1d4ac4b6
FD
26# Key to sort streams in a predictable order.
27def sort_predictably(stream):
28 if 'range-ns' in stream:
29 return stream['range-ns']['begin']
30 else:
31 return stream['paths'][0]
d907165c
SM
32
33
34class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
35
36 def setUp(self):
37 ctf = bt2.find_plugin('ctf')
38 self._fs = ctf.source_component_classes['fs']
39
40 self._paths = [os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')]
41 self._executor = bt2.QueryExecutor()
42
43 def _check(self, trace, offset):
44 self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
45 self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
46 self.assertEqual(trace['intersection-range-ns']['begin'], 13515309000000070 + offset)
47 self.assertEqual(trace['intersection-range-ns']['end'], 13515309000000100 + offset)
48
1d4ac4b6 49 streams = sorted(trace['streams'], key=sort_predictably)
d907165c
SM
50 self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
51 self.assertEqual(streams[0]['range-ns']['end'], 13515309000000100 + offset)
52 self.assertEqual(streams[1]['range-ns']['begin'], 13515309000000070 + offset)
53 self.assertEqual(streams[1]['range-ns']['end'], 13515309000000120 + offset)
54
55 # Test various cominations of the clock-class-offset-s and
56 # clock-class-offset-ns parameters to trace-info queries.
57
58 # Without clock class offset
59
60 def test_no_clock_class_offset(self):
61 res = self._executor.query(self._fs, 'trace-info', {
62 'paths': self._paths,
63 })
64 trace = res[0]
65 self._check(trace, 0)
66
67 # With clock-class-offset-s
68
69 def test_clock_class_offset_s(self):
70 res = self._executor.query(self._fs, 'trace-info', {
71 'paths': self._paths,
72 'clock-class-offset-s': 2,
73 })
74 trace = res[0]
75 self._check(trace, 2000000000)
76
77 # With clock-class-offset-ns
78
79 def test_clock_class_offset_ns(self):
80 res = self._executor.query(self._fs, 'trace-info', {
81 'paths': self._paths,
82 'clock-class-offset-ns': 2,
83 })
84 trace = res[0]
85 self._check(trace, 2)
86
87 # With both, negative
88
89 def test_clock_class_offset_both(self):
90 res = self._executor.query(self._fs, 'trace-info', {
91 'paths': self._paths,
92 'clock-class-offset-s': -2,
93 'clock-class-offset-ns': -2,
94 })
95 trace = res[0]
96 self._check(trace, -2000000002)
97
98 def test_clock_class_offset_s_wrong_type(self):
99 with self.assertRaises(bt2.InvalidQueryParams):
100 self._executor.query(self._fs, 'trace-info', {
101 'paths': self._paths,
102 'clock-class-offset-s': "2",
103 })
104
105 def test_clock_class_offset_s_wrong_type_none(self):
106 with self.assertRaises(bt2.InvalidQueryParams):
107 self._executor.query(self._fs, 'trace-info', {
108 'paths': self._paths,
109 'clock-class-offset-s': None,
110 })
111
112 def test_clock_class_offset_ns_wrong_type(self):
113 with self.assertRaises(bt2.InvalidQueryParams):
114 self._executor.query(self._fs, 'trace-info', {
115 'paths': self._paths,
116 'clock-class-offset-ns': "2",
117 })
118
119 def test_clock_class_offset_ns_wrong_type_none(self):
120 with self.assertRaises(bt2.InvalidQueryParams):
121 self._executor.query(self._fs, 'trace-info', {
122 'paths': self._paths,
123 'clock-class-offset-ns': None,
124 })
125
a38d7650
SM
126
127class QueryTraceInfoPortNameTestCase(unittest.TestCase):
128 def setUp(self):
129 ctf = bt2.find_plugin("ctf")
130 self._fs = ctf.source_component_classes["fs"]
131
132 self._executor = bt2.QueryExecutor()
133
134 def test_trace_uuid_stream_class_id_no_stream_id(self):
135 res = self._executor.query(
136 self._fs,
137 "trace-info",
138 {
139 "paths": [
140 os.path.join(
141 test_ctf_traces_path, "intersection", "3eventsintersect"
142 )
143 ]
144 },
145 )
146 self.assertEqual(len(res), 1)
147 trace = res[0]
1d4ac4b6 148 streams = sorted(trace["streams"], key=sort_predictably)
a38d7650
SM
149 self.assertEqual(len(streams), 2)
150 self.assertRegexpMatches(
151 str(streams[0]["port-name"]),
152 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_0$",
153 )
154 self.assertRegexpMatches(
155 str(streams[1]["port-name"]),
156 r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_1$",
157 )
158
159 def test_trace_uuid_no_stream_class_id_no_stream_id(self):
160 res = self._executor.query(
161 self._fs,
162 "trace-info",
163 {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
164 )
165 self.assertEqual(len(res), 1)
166 trace = res[0]
1d4ac4b6 167 streams = sorted(trace["streams"], key=sort_predictably)
a38d7650
SM
168 self.assertEqual(len(streams), 1)
169 self.assertRegexpMatches(
170 str(streams[0]["port-name"]),
171 r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*/tests/ctf-traces/succeed/succeed1/dummystream$",
172 )
173
174
1d4ac4b6
FD
175class QueryTraceInfoRangeTestCase(unittest.TestCase):
176 def setUp(self):
177 ctf = bt2.find_plugin("ctf")
178 self._fs = ctf.source_component_classes["fs"]
179
180 self._executor = bt2.QueryExecutor()
181
182 def test_trace_no_range(self):
183 # This trace has no `timestamp_begin` and `timestamp_end` in its packet
184 # context. The `trace-info` query should omit the `range-ns` fields in
185 # the `trace` and `stream` data structures.
186
187 res = self._executor.query(
188 self._fs,
189 "trace-info",
190 {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
191 )
192
193 self.assertEqual(len(res), 1)
194 trace = res[0]
195 streams = trace["streams"]
196 self.assertEqual(len(streams), 1)
197
198 self.assertRaises(KeyError, lambda: trace['range-ns'])
199 self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
200
201
d907165c
SM
202if __name__ == '__main__':
203 unittest.main()
This page took 0.032419 seconds and 4 git commands to generate.