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
  44     def _check(self, trace, offset):
  45         self.assertEqual(trace['range-ns']['begin'], 13515309000000000 + offset)
  46         self.assertEqual(trace['range-ns']['end'], 13515309000000120 + offset)
  47         self.assertEqual(
  48             trace['intersection-range-ns']['begin'], 13515309000000070 + offset
  49         )
  50         self.assertEqual(
  51             trace['intersection-range-ns']['end'], 13515309000000100 + offset
  52         )
  53
  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)
  59
  60     # Test various cominations of the clock-class-offset-s and
  61     # clock-class-offset-ns parameters to babeltrace.trace-info queries.
  62
  63     # Without clock class offset
  64
  65     def test_no_clock_class_offset(self):
  66         res = bt2.QueryExecutor(
  67             self._fs, 'babeltrace.trace-info', {'inputs': self._inputs}
  68         ).query()
  69         trace = res[0]
  70         self._check(trace, 0)
  71
  72     # With clock-class-offset-s
  73
  74     def test_clock_class_offset_s(self):
  75         res = bt2.QueryExecutor(
  76             self._fs,
  77             'babeltrace.trace-info',
  78             {'inputs': self._inputs, 'clock-class-offset-s': 2},
  79         ).query()
  80         trace = res[0]
  81         self._check(trace, 2000000000)
  82
  83     # With clock-class-offset-ns
  84
  85     def test_clock_class_offset_ns(self):
  86         res = bt2.QueryExecutor(
  87             self._fs,
  88             'babeltrace.trace-info',
  89             {'inputs': self._inputs, 'clock-class-offset-ns': 2},
  90         ).query()
  91         trace = res[0]
  92         self._check(trace, 2)
  93
  94     # With both, negative
  95
  96     def test_clock_class_offset_both(self):
  97         res = bt2.QueryExecutor(
  98             self._fs,
  99             'babeltrace.trace-info',
 100             {
 101                 'inputs': self._inputs,
 102                 'clock-class-offset-s': -2,
 103                 'clock-class-offset-ns': -2,
 104             },
 105         ).query()
 106         trace = res[0]
 107         self._check(trace, -2000000002)
 108
 109     def test_clock_class_offset_s_wrong_type(self):
 110         with self.assertRaises(bt2._Error):
 111             bt2.QueryExecutor(
 112                 self._fs,
 113                 'babeltrace.trace-info',
 114                 {'inputs': self._inputs, 'clock-class-offset-s': "2"},
 115             ).query()
 116
 117     def test_clock_class_offset_s_wrong_type_none(self):
 118         with self.assertRaises(bt2._Error):
 119             bt2.QueryExecutor(
 120                 self._fs,
 121                 'babeltrace.trace-info',
 122                 {'inputs': self._inputs, 'clock-class-offset-s': None},
 123             ).query()
 124
 125     def test_clock_class_offset_ns_wrong_type(self):
 126         with self.assertRaises(bt2._Error):
 127             bt2.QueryExecutor(
 128                 self._fs,
 129                 'babeltrace.trace-info',
 130                 {'inputs': self._inputs, 'clock-class-offset-ns': "2"},
 131             ).query()
 132
 133     def test_clock_class_offset_ns_wrong_type_none(self):
 134         with self.assertRaises(bt2._Error):
 135             bt2.QueryExecutor(
 136                 self._fs,
 137                 'babeltrace.trace-info',
 138                 {'inputs': self._inputs, 'clock-class-offset-ns': None},
 139             ).query()
 140
 141
 142 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
 143     def setUp(self):
 144         ctf = bt2.find_plugin("ctf")
 145         self._fs = ctf.source_component_classes["fs"]
 146
 147     def test_trace_uuid_stream_class_id_no_stream_id(self):
 148         res = bt2.QueryExecutor(
 149             self._fs,
 150             "babeltrace.trace-info",
 151             {
 152                 "inputs": [
 153                     os.path.join(
 154                         test_ctf_traces_path, "intersection", "3eventsintersect"
 155                     )
 156                 ]
 157             },
 158         ).query()
 159
 160         os_stream_path = PurePosixPath(
 161             '/tests/data/ctf-traces/intersection/3eventsintersect/'
 162         )
 163         if os.environ['BT_OS_TYPE'] == 'mingw':
 164             os_stream_path = PureWindowsPath(os_stream_path)
 165
 166         self.assertEqual(len(res), 1)
 167         trace = res[0]
 168         streams = sorted(trace["streams"], key=sort_predictably)
 169         self.assertEqual(len(streams), 2)
 170         self.assertRegexpMatches(
 171             str(streams[0]["port-name"]),
 172             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
 173             + re.escape(str(os_stream_path / "test_stream_0"))
 174             + r"$",
 175         )
 176         self.assertRegexpMatches(
 177             str(streams[1]["port-name"]),
 178             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
 179             + re.escape(str(os_stream_path / "test_stream_1"))
 180             + r"$",
 181         )
 182
 183     def test_trace_uuid_no_stream_class_id_no_stream_id(self):
 184         res = bt2.QueryExecutor(
 185             self._fs,
 186             "babeltrace.trace-info",
 187             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 188         ).query()
 189
 190         os_stream_path = PurePosixPath(
 191             '/tests/data/ctf-traces/succeed/succeed1/dummystream'
 192         )
 193         if os.environ['BT_OS_TYPE'] == 'mingw':
 194             os_stream_path = PureWindowsPath(os_stream_path)
 195
 196         self.assertEqual(len(res), 1)
 197         trace = res[0]
 198         streams = sorted(trace["streams"], key=sort_predictably)
 199         self.assertEqual(len(streams), 1)
 200         self.assertRegexpMatches(
 201             str(streams[0]["port-name"]),
 202             r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
 203             + re.escape(str(os_stream_path))
 204             + r"$",
 205         )
 206
 207
 208 class QueryTraceInfoRangeTestCase(unittest.TestCase):
 209     def setUp(self):
 210         ctf = bt2.find_plugin("ctf")
 211         self._fs = ctf.source_component_classes["fs"]
 212
 213     def test_trace_no_range(self):
 214         # This trace has no `timestamp_begin` and `timestamp_end` in its
 215         # packet context. The `babeltrace.trace-info` query should omit
 216         # the `range-ns` fields in the `trace` and `stream` data
 217         # structures.
 218
 219         res = bt2.QueryExecutor(
 220             self._fs,
 221             "babeltrace.trace-info",
 222             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 223         ).query()
 224
 225         self.assertEqual(len(res), 1)
 226         trace = res[0]
 227         streams = trace["streams"]
 228         self.assertEqual(len(streams), 1)
 229
 230         self.assertRaises(KeyError, lambda: trace['range-ns'])
 231         self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
 232
 233
 234 if __name__ == '__main__':
 235     unittest.main()