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     return stream['port-name']
  30
  31
  32 class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
  33     def setUp(self):
  34         ctf = bt2.find_plugin('ctf')
  35         self._fs = ctf.source_component_classes['fs']
  36
  37         self._inputs = [
  38             os.path.join(test_ctf_traces_path, 'intersection', '3eventsintersect')
  39         ]
  40
  41     def _check(self, trace, offset):
  42         self.assertEqual(
  43             trace['intersection-range-ns']['begin'], 13515309000000070 + offset
  44         )
  45         self.assertEqual(
  46             trace['intersection-range-ns']['end'], 13515309000000100 + offset
  47         )
  48
  49         streams = sorted(trace['streams'], key=sort_predictably)
  50         self.assertEqual(streams[0]['range-ns']['begin'], 13515309000000000 + offset)
  51         self.assertEqual(streams[0]['range-ns']['end'], 13515309000000100 + offset)
  52         self.assertEqual(streams[1]['range-ns']['begin'], 13515309000000070 + offset)
  53         self.assertEqual(streams[1]['range-ns']['end'], 13515309000000120 + offset)
  54
  55     # Test various cominations of the clock-class-offset-s and
  56     # clock-class-offset-ns parameters to babeltrace.trace-info queries.
  57
  58     # Without clock class offset
  59
  60     def test_no_clock_class_offset(self):
  61         res = bt2.QueryExecutor(
  62             self._fs, 'babeltrace.trace-info', {'inputs': self._inputs}
  63         ).query()
  64         trace = res[0]
  65         self._check(trace, 0)
  66
  67     # With clock-class-offset-s
  68
  69     def test_clock_class_offset_s(self):
  70         res = bt2.QueryExecutor(
  71             self._fs,
  72             'babeltrace.trace-info',
  73             {'inputs': self._inputs, 'clock-class-offset-s': 2},
  74         ).query()
  75         trace = res[0]
  76         self._check(trace, 2000000000)
  77
  78     # With clock-class-offset-ns
  79
  80     def test_clock_class_offset_ns(self):
  81         res = bt2.QueryExecutor(
  82             self._fs,
  83             'babeltrace.trace-info',
  84             {'inputs': self._inputs, 'clock-class-offset-ns': 2},
  85         ).query()
  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 = bt2.QueryExecutor(
  93             self._fs,
  94             'babeltrace.trace-info',
  95             {
  96                 'inputs': self._inputs,
  97                 'clock-class-offset-s': -2,
  98                 'clock-class-offset-ns': -2,
  99             },
 100         ).query()
 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._Error):
 106             bt2.QueryExecutor(
 107                 self._fs,
 108                 'babeltrace.trace-info',
 109                 {'inputs': self._inputs, 'clock-class-offset-s': "2"},
 110             ).query()
 111
 112     def test_clock_class_offset_s_wrong_type_none(self):
 113         with self.assertRaises(bt2._Error):
 114             bt2.QueryExecutor(
 115                 self._fs,
 116                 'babeltrace.trace-info',
 117                 {'inputs': self._inputs, 'clock-class-offset-s': None},
 118             ).query()
 119
 120     def test_clock_class_offset_ns_wrong_type(self):
 121         with self.assertRaises(bt2._Error):
 122             bt2.QueryExecutor(
 123                 self._fs,
 124                 'babeltrace.trace-info',
 125                 {'inputs': self._inputs, 'clock-class-offset-ns': "2"},
 126             ).query()
 127
 128     def test_clock_class_offset_ns_wrong_type_none(self):
 129         with self.assertRaises(bt2._Error):
 130             bt2.QueryExecutor(
 131                 self._fs,
 132                 'babeltrace.trace-info',
 133                 {'inputs': self._inputs, 'clock-class-offset-ns': None},
 134             ).query()
 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     def test_trace_uuid_stream_class_id_no_stream_id(self):
 143         res = bt2.QueryExecutor(
 144             self._fs,
 145             "babeltrace.trace-info",
 146             {
 147                 "inputs": [
 148                     os.path.join(
 149                         test_ctf_traces_path, "intersection", "3eventsintersect"
 150                     )
 151                 ]
 152             },
 153         ).query()
 154
 155         os_stream_path = PurePosixPath(
 156             '/tests/data/ctf-traces/intersection/3eventsintersect/'
 157         )
 158         if os.environ['BT_OS_TYPE'] == 'mingw':
 159             os_stream_path = PureWindowsPath(os_stream_path)
 160
 161         self.assertEqual(len(res), 1)
 162         trace = res[0]
 163         streams = sorted(trace["streams"], key=sort_predictably)
 164         self.assertEqual(len(streams), 2)
 165         self.assertRegexpMatches(
 166             str(streams[0]["port-name"]),
 167             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
 168             + re.escape(str(os_stream_path / "test_stream_0"))
 169             + r"$",
 170         )
 171         self.assertRegexpMatches(
 172             str(streams[1]["port-name"]),
 173             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
 174             + re.escape(str(os_stream_path / "test_stream_1"))
 175             + r"$",
 176         )
 177
 178     def test_trace_uuid_no_stream_class_id_no_stream_id(self):
 179         res = bt2.QueryExecutor(
 180             self._fs,
 181             "babeltrace.trace-info",
 182             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 183         ).query()
 184
 185         os_stream_path = PurePosixPath(
 186             '/tests/data/ctf-traces/succeed/succeed1/dummystream'
 187         )
 188         if os.environ['BT_OS_TYPE'] == 'mingw':
 189             os_stream_path = PureWindowsPath(os_stream_path)
 190
 191         self.assertEqual(len(res), 1)
 192         trace = res[0]
 193         streams = sorted(trace["streams"], key=sort_predictably)
 194         self.assertEqual(len(streams), 1)
 195         self.assertRegexpMatches(
 196             str(streams[0]["port-name"]),
 197             r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
 198             + re.escape(str(os_stream_path))
 199             + r"$",
 200         )
 201
 202
 203 class QueryTraceInfoRangeTestCase(unittest.TestCase):
 204     def setUp(self):
 205         ctf = bt2.find_plugin("ctf")
 206         self._fs = ctf.source_component_classes["fs"]
 207
 208     def test_trace_no_range(self):
 209         # This trace has no `timestamp_begin` and `timestamp_end` in its
 210         # packet context. The `babeltrace.trace-info` query should omit
 211         # the `range-ns` fields in the `trace` and `stream` data
 212         # structures.
 213
 214         res = bt2.QueryExecutor(
 215             self._fs,
 216             "babeltrace.trace-info",
 217             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 218         ).query()
 219
 220         self.assertEqual(len(res), 1)
 221         trace = res[0]
 222         streams = trace["streams"]
 223         self.assertEqual(len(streams), 1)
 224
 225         self.assertRaises(KeyError, lambda: trace['range-ns'])
 226         self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
 227
 228
 229 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest.TestCase):
 230     def setUp(self):
 231         ctf = bt2.find_plugin('ctf')
 232         self._fs = ctf.source_component_classes['fs']
 233         self._path = os.path.join(test_ctf_traces_path, 'succeed')
 234
 235     def _test_lttng_quirks(self, trace_name):
 236         res = bt2.QueryExecutor(
 237             self._fs,
 238             "babeltrace.trace-info",
 239             {"inputs": [os.path.join(self._path, trace_name)]},
 240         ).query()
 241
 242         self.assertEqual(len(res), 1)
 243         return res[0]
 244
 245     def test_event_after_packet(self):
 246         trace = self._test_lttng_quirks("lttng-event-after-packet")
 247         streams = trace["streams"]
 248         self.assertEqual(len(streams), 1)
 249
 250         self.assertEqual(streams[0]['range-ns']['begin'], 1565957300948091100)
 251         self.assertEqual(streams[0]['range-ns']['end'], 1565957302180016069)
 252
 253     def test_lttng_crash(self):
 254         trace = self._test_lttng_quirks("lttng-crash")
 255         streams = trace["streams"]
 256         self.assertEqual(len(streams), 1)
 257
 258         self.assertEqual(streams[0]['range-ns']['begin'], 1565891729288866738)
 259         self.assertEqual(streams[0]['range-ns']['end'], 1565891729293526525)
 260
 261
 262 if __name__ == '__main__':
 263     unittest.main()