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()