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