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._inputs = [
  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 babeltrace.trace-info queries.
  63
  64     # Without clock class offset
  65
  66     def test_no_clock_class_offset(self):
  67         res = self._executor.query(
  68             self._fs, 'babeltrace.trace-info', {'inputs': self._inputs}
  69         )
  70         trace = res[0]
  71         self._check(trace, 0)
  72
  73     # With clock-class-offset-s
  74
  75     def test_clock_class_offset_s(self):
  76         res = self._executor.query(
  77             self._fs,
  78             'babeltrace.trace-info',
  79             {'inputs': self._inputs, 'clock-class-offset-s': 2},
  80         )
  81         trace = res[0]
  82         self._check(trace, 2000000000)
  83
  84     # With clock-class-offset-ns
  85
  86     def test_clock_class_offset_ns(self):
  87         res = self._executor.query(
  88             self._fs,
  89             'babeltrace.trace-info',
  90             {'inputs': self._inputs, 'clock-class-offset-ns': 2},
  91         )
  92         trace = res[0]
  93         self._check(trace, 2)
  94
  95     # With both, negative
  96
  97     def test_clock_class_offset_both(self):
  98         res = self._executor.query(
  99             self._fs,
 100             'babeltrace.trace-info',
 101             {
 102                 'inputs': self._inputs,
 103                 'clock-class-offset-s': -2,
 104                 'clock-class-offset-ns': -2,
 105             },
 106         )
 107         trace = res[0]
 108         self._check(trace, -2000000002)
 109
 110     def test_clock_class_offset_s_wrong_type(self):
 111         with self.assertRaises(bt2._Error):
 112             self._executor.query(
 113                 self._fs,
 114                 'babeltrace.trace-info',
 115                 {'inputs': self._inputs, 'clock-class-offset-s': "2"},
 116             )
 117
 118     def test_clock_class_offset_s_wrong_type_none(self):
 119         with self.assertRaises(bt2._Error):
 120             self._executor.query(
 121                 self._fs,
 122                 'babeltrace.trace-info',
 123                 {'inputs': self._inputs, 'clock-class-offset-s': None},
 124             )
 125
 126     def test_clock_class_offset_ns_wrong_type(self):
 127         with self.assertRaises(bt2._Error):
 128             self._executor.query(
 129                 self._fs,
 130                 'babeltrace.trace-info',
 131                 {'inputs': self._inputs, 'clock-class-offset-ns': "2"},
 132             )
 133
 134     def test_clock_class_offset_ns_wrong_type_none(self):
 135         with self.assertRaises(bt2._Error):
 136             self._executor.query(
 137                 self._fs,
 138                 'babeltrace.trace-info',
 139                 {'inputs': self._inputs, 'clock-class-offset-ns': None},
 140             )
 141
 142
 143 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
 144     def setUp(self):
 145         ctf = bt2.find_plugin("ctf")
 146         self._fs = ctf.source_component_classes["fs"]
 147
 148         self._executor = bt2.QueryExecutor()
 149
 150     def test_trace_uuid_stream_class_id_no_stream_id(self):
 151         res = self._executor.query(
 152             self._fs,
 153             "babeltrace.trace-info",
 154             {
 155                 "inputs": [
 156                     os.path.join(
 157                         test_ctf_traces_path, "intersection", "3eventsintersect"
 158                     )
 159                 ]
 160             },
 161         )
 162
 163         os_stream_path = PurePosixPath(
 164             '/tests/data/ctf-traces/intersection/3eventsintersect/'
 165         )
 166         if os.environ['BT_OS_TYPE'] == 'mingw':
 167             os_stream_path = PureWindowsPath(os_stream_path)
 168
 169         self.assertEqual(len(res), 1)
 170         trace = res[0]
 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"))
 177             + r"$",
 178         )
 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"))
 183             + r"$",
 184         )
 185
 186     def test_trace_uuid_no_stream_class_id_no_stream_id(self):
 187         res = self._executor.query(
 188             self._fs,
 189             "babeltrace.trace-info",
 190             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 191         )
 192
 193         os_stream_path = PurePosixPath(
 194             '/tests/data/ctf-traces/succeed/succeed1/dummystream'
 195         )
 196         if os.environ['BT_OS_TYPE'] == 'mingw':
 197             os_stream_path = PureWindowsPath(os_stream_path)
 198
 199         self.assertEqual(len(res), 1)
 200         trace = res[0]
 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))
 207             + r"$",
 208         )
 209
 210
 211 class QueryTraceInfoRangeTestCase(unittest.TestCase):
 212     def setUp(self):
 213         ctf = bt2.find_plugin("ctf")
 214         self._fs = ctf.source_component_classes["fs"]
 215
 216         self._executor = bt2.QueryExecutor()
 217
 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
 222         # structures.
 223
 224         res = self._executor.query(
 225             self._fs,
 226             "babeltrace.trace-info",
 227             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 228         )
 229
 230         self.assertEqual(len(res), 1)
 231         trace = res[0]
 232         streams = trace["streams"]
 233         self.assertEqual(len(streams), 1)
 234
 235         self.assertRaises(KeyError, lambda: trace['range-ns'])
 236         self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
 237
 238
 239 if __name__ == '__main__':
 240     unittest.main()