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.
23 test_ctf_traces_path
= os
.environ
['BT_CTF_TRACES_PATH']
26 # Key to sort streams in a predictable order.
27 def sort_predictably(stream
):
28 if 'range-ns' in stream
:
29 return stream
['range-ns']['begin']
31 return stream
['paths'][0]
34 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
36 ctf
= bt2
.find_plugin('ctf')
37 self
._fs
= ctf
.source_component_classes
['fs']
40 os
.path
.join(test_ctf_traces_path
, 'intersection', '3eventsintersect')
42 self
._executor
= bt2
.QueryExecutor()
44 def _check(self
, trace
, offset
):
45 self
.assertEqual(trace
['range-ns']['begin'], 13515309000000000 + offset
)
46 self
.assertEqual(trace
['range-ns']['end'], 13515309000000120 + offset
)
48 trace
['intersection-range-ns']['begin'], 13515309000000070 + offset
51 trace
['intersection-range-ns']['end'], 13515309000000100 + offset
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
)
60 # Test various cominations of the clock-class-offset-s and
61 # clock-class-offset-ns parameters to trace-info queries.
63 # Without clock class offset
65 def test_no_clock_class_offset(self
):
66 res
= self
._executor
.query(self
._fs
, 'trace-info', {'paths': self
._paths
})
70 # With clock-class-offset-s
72 def test_clock_class_offset_s(self
):
73 res
= self
._executor
.query(
74 self
._fs
, 'trace-info', {'paths': self
._paths
, 'clock-class-offset-s': 2}
77 self
._check
(trace
, 2000000000)
79 # With clock-class-offset-ns
81 def test_clock_class_offset_ns(self
):
82 res
= self
._executor
.query(
83 self
._fs
, 'trace-info', {'paths': self
._paths
, 'clock-class-offset-ns': 2}
90 def test_clock_class_offset_both(self
):
91 res
= self
._executor
.query(
96 'clock-class-offset-s': -2,
97 'clock-class-offset-ns': -2,
101 self
._check
(trace
, -2000000002)
103 def test_clock_class_offset_s_wrong_type(self
):
104 with self
.assertRaises(bt2
.InvalidParams
):
105 self
._executor
.query(
108 {'paths': self
._paths
, 'clock-class-offset-s': "2"},
111 def test_clock_class_offset_s_wrong_type_none(self
):
112 with self
.assertRaises(bt2
.InvalidParams
):
113 self
._executor
.query(
116 {'paths': self
._paths
, 'clock-class-offset-s': None},
119 def test_clock_class_offset_ns_wrong_type(self
):
120 with self
.assertRaises(bt2
.InvalidParams
):
121 self
._executor
.query(
124 {'paths': self
._paths
, 'clock-class-offset-ns': "2"},
127 def test_clock_class_offset_ns_wrong_type_none(self
):
128 with self
.assertRaises(bt2
.InvalidParams
):
129 self
._executor
.query(
132 {'paths': self
._paths
, 'clock-class-offset-ns': None},
136 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
138 ctf
= bt2
.find_plugin("ctf")
139 self
._fs
= ctf
.source_component_classes
["fs"]
141 self
._executor
= bt2
.QueryExecutor()
143 def test_trace_uuid_stream_class_id_no_stream_id(self
):
144 res
= self
._executor
.query(
150 test_ctf_traces_path
, "intersection", "3eventsintersect"
155 self
.assertEqual(len(res
), 1)
157 streams
= sorted(trace
["streams"], key
=sort_predictably
)
158 self
.assertEqual(len(streams
), 2)
159 self
.assertRegexpMatches(
160 str(streams
[0]["port-name"]),
161 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/data/ctf-traces/intersection/3eventsintersect/test_stream_0$",
163 self
.assertRegexpMatches(
164 str(streams
[1]["port-name"]),
165 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/data/ctf-traces/intersection/3eventsintersect/test_stream_1$",
168 def test_trace_uuid_no_stream_class_id_no_stream_id(self
):
169 res
= self
._executor
.query(
172 {"paths": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
174 self
.assertEqual(len(res
), 1)
176 streams
= sorted(trace
["streams"], key
=sort_predictably
)
177 self
.assertEqual(len(streams
), 1)
178 self
.assertRegexpMatches(
179 str(streams
[0]["port-name"]),
180 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*/tests/data/ctf-traces/succeed/succeed1/dummystream$",
184 class QueryTraceInfoRangeTestCase(unittest
.TestCase
):
186 ctf
= bt2
.find_plugin("ctf")
187 self
._fs
= ctf
.source_component_classes
["fs"]
189 self
._executor
= bt2
.QueryExecutor()
191 def test_trace_no_range(self
):
192 # This trace has no `timestamp_begin` and `timestamp_end` in its packet
193 # context. The `trace-info` query should omit the `range-ns` fields in
194 # the `trace` and `stream` data structures.
196 res
= self
._executor
.query(
199 {"paths": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
202 self
.assertEqual(len(res
), 1)
204 streams
= trace
["streams"]
205 self
.assertEqual(len(streams
), 1)
207 self
.assertRaises(KeyError, lambda: trace
['range-ns'])
208 self
.assertRaises(KeyError, lambda: streams
[0]['range-ns'])
211 if __name__
== '__main__':
This page took 0.036044 seconds and 5 git commands to generate.