053c0e1e43c20b4a7f8892d96100bb09c8903ccd
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
['TEST_CTF_TRACES_PATH']
26 # Key to streams by their being timestamp. Used to get the list of streams in
27 # a predictable order.
29 def sort_by_begin(stream
):
30 return stream
['range-ns']['begin']
33 class QueryTraceInfoClockOffsetTestCase(unittest
.TestCase
):
36 ctf
= bt2
.find_plugin('ctf')
37 self
._fs
= ctf
.source_component_classes
['fs']
39 self
._paths
= [os
.path
.join(test_ctf_traces_path
, 'intersection', '3eventsintersect')]
40 self
._executor
= bt2
.QueryExecutor()
42 def _check(self
, trace
, offset
):
43 self
.assertEqual(trace
['range-ns']['begin'], 13515309000000000 + offset
)
44 self
.assertEqual(trace
['range-ns']['end'], 13515309000000120 + offset
)
45 self
.assertEqual(trace
['intersection-range-ns']['begin'], 13515309000000070 + offset
)
46 self
.assertEqual(trace
['intersection-range-ns']['end'], 13515309000000100 + offset
)
48 streams
= sorted(trace
['streams'], key
=sort_by_begin
)
49 self
.assertEqual(streams
[0]['range-ns']['begin'], 13515309000000000 + offset
)
50 self
.assertEqual(streams
[0]['range-ns']['end'], 13515309000000100 + offset
)
51 self
.assertEqual(streams
[1]['range-ns']['begin'], 13515309000000070 + offset
)
52 self
.assertEqual(streams
[1]['range-ns']['end'], 13515309000000120 + offset
)
54 # Test various cominations of the clock-class-offset-s and
55 # clock-class-offset-ns parameters to trace-info queries.
57 # Without clock class offset
59 def test_no_clock_class_offset(self
):
60 res
= self
._executor
.query(self
._fs
, 'trace-info', {
66 # With clock-class-offset-s
68 def test_clock_class_offset_s(self
):
69 res
= self
._executor
.query(self
._fs
, 'trace-info', {
71 'clock-class-offset-s': 2,
74 self
._check
(trace
, 2000000000)
76 # With clock-class-offset-ns
78 def test_clock_class_offset_ns(self
):
79 res
= self
._executor
.query(self
._fs
, 'trace-info', {
81 'clock-class-offset-ns': 2,
88 def test_clock_class_offset_both(self
):
89 res
= self
._executor
.query(self
._fs
, 'trace-info', {
91 'clock-class-offset-s': -2,
92 'clock-class-offset-ns': -2,
95 self
._check
(trace
, -2000000002)
97 def test_clock_class_offset_s_wrong_type(self
):
98 with self
.assertRaises(bt2
.InvalidQueryParams
):
99 self
._executor
.query(self
._fs
, 'trace-info', {
100 'paths': self
._paths
,
101 'clock-class-offset-s': "2",
104 def test_clock_class_offset_s_wrong_type_none(self
):
105 with self
.assertRaises(bt2
.InvalidQueryParams
):
106 self
._executor
.query(self
._fs
, 'trace-info', {
107 'paths': self
._paths
,
108 'clock-class-offset-s': None,
111 def test_clock_class_offset_ns_wrong_type(self
):
112 with self
.assertRaises(bt2
.InvalidQueryParams
):
113 self
._executor
.query(self
._fs
, 'trace-info', {
114 'paths': self
._paths
,
115 'clock-class-offset-ns': "2",
118 def test_clock_class_offset_ns_wrong_type_none(self
):
119 with self
.assertRaises(bt2
.InvalidQueryParams
):
120 self
._executor
.query(self
._fs
, 'trace-info', {
121 'paths': self
._paths
,
122 'clock-class-offset-ns': None,
126 class QueryTraceInfoPortNameTestCase(unittest
.TestCase
):
128 ctf
= bt2
.find_plugin("ctf")
129 self
._fs
= ctf
.source_component_classes
["fs"]
131 self
._executor
= bt2
.QueryExecutor()
133 def test_trace_uuid_stream_class_id_no_stream_id(self
):
134 res
= self
._executor
.query(
140 test_ctf_traces_path
, "intersection", "3eventsintersect"
145 self
.assertEqual(len(res
), 1)
147 streams
= sorted(trace
["streams"], key
=sort_by_begin
)
148 self
.assertEqual(len(streams
), 2)
149 self
.assertRegexpMatches(
150 str(streams
[0]["port-name"]),
151 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_0$",
153 self
.assertRegexpMatches(
154 str(streams
[1]["port-name"]),
155 r
"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_1$",
158 def test_trace_uuid_no_stream_class_id_no_stream_id(self
):
159 res
= self
._executor
.query(
162 {"paths": [os
.path
.join(test_ctf_traces_path
, "succeed", "succeed1")]},
164 self
.assertEqual(len(res
), 1)
166 streams
= sorted(trace
["streams"], key
=sort_by_begin
)
167 self
.assertEqual(len(streams
), 1)
168 self
.assertRegexpMatches(
169 str(streams
[0]["port-name"]),
170 r
"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*/tests/ctf-traces/succeed/succeed1/dummystream$",
174 if __name__
== '__main__':
This page took 0.032569 seconds and 3 git commands to generate.