tests/plugins/src.ctf.fs/query/test_query_trace_info.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 Simon Marchi <simon.marchi@efficios.com>
   4 #
   5
   6 import unittest
   7 import bt2
   8 import os
   9 import re
  10
  11
  12 test_ctf_traces_path = os.environ["BT_CTF_TRACES_PATH"]
  13
  14
  15 # Key to sort streams in a predictable order.
  16 def sort_predictably(stream):
  17     return stream["port-name"]
  18
  19
  20 class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
  21     def setUp(self):
  22         ctf = bt2.find_plugin("ctf")
  23         self._fs = ctf.source_component_classes["fs"]
  24
  25         self._inputs = [
  26             os.path.join(test_ctf_traces_path, "intersection", "3eventsintersect")
  27         ]
  28
  29     def _check(self, trace, offset):
  30         streams = sorted(trace["stream-infos"], key=sort_predictably)
  31         self.assertEqual(streams[0]["range-ns"]["begin"], 13515309000000000 + offset)
  32         self.assertEqual(streams[0]["range-ns"]["end"], 13515309000000100 + offset)
  33         self.assertEqual(streams[1]["range-ns"]["begin"], 13515309000000070 + offset)
  34         self.assertEqual(streams[1]["range-ns"]["end"], 13515309000000120 + offset)
  35
  36     # Test various cominations of the clock-class-offset-s and
  37     # clock-class-offset-ns parameters to babeltrace.trace-infos queries.
  38
  39     # Without clock class offset
  40
  41     def test_no_clock_class_offset(self):
  42         res = bt2.QueryExecutor(
  43             self._fs, "babeltrace.trace-infos", {"inputs": self._inputs}
  44         ).query()
  45         trace = res[0]
  46         self._check(trace, 0)
  47
  48     # With clock-class-offset-s
  49
  50     def test_clock_class_offset_s(self):
  51         res = bt2.QueryExecutor(
  52             self._fs,
  53             "babeltrace.trace-infos",
  54             {"inputs": self._inputs, "clock-class-offset-s": 2},
  55         ).query()
  56         trace = res[0]
  57         self._check(trace, 2000000000)
  58
  59     # With clock-class-offset-ns
  60
  61     def test_clock_class_offset_ns(self):
  62         res = bt2.QueryExecutor(
  63             self._fs,
  64             "babeltrace.trace-infos",
  65             {"inputs": self._inputs, "clock-class-offset-ns": 2},
  66         ).query()
  67         trace = res[0]
  68         self._check(trace, 2)
  69
  70     # With both, negative
  71
  72     def test_clock_class_offset_both(self):
  73         res = bt2.QueryExecutor(
  74             self._fs,
  75             "babeltrace.trace-infos",
  76             {
  77                 "inputs": self._inputs,
  78                 "clock-class-offset-s": -2,
  79                 "clock-class-offset-ns": -2,
  80             },
  81         ).query()
  82         trace = res[0]
  83         self._check(trace, -2000000002)
  84
  85     def test_clock_class_offset_s_wrong_type(self):
  86         with self.assertRaises(bt2._Error):
  87             bt2.QueryExecutor(
  88                 self._fs,
  89                 "babeltrace.trace-infos",
  90                 {"inputs": self._inputs, "clock-class-offset-s": "2"},
  91             ).query()
  92
  93     def test_clock_class_offset_s_wrong_type_none(self):
  94         with self.assertRaises(bt2._Error):
  95             bt2.QueryExecutor(
  96                 self._fs,
  97                 "babeltrace.trace-infos",
  98                 {"inputs": self._inputs, "clock-class-offset-s": None},
  99             ).query()
 100
 101     def test_clock_class_offset_ns_wrong_type(self):
 102         with self.assertRaises(bt2._Error):
 103             bt2.QueryExecutor(
 104                 self._fs,
 105                 "babeltrace.trace-infos",
 106                 {"inputs": self._inputs, "clock-class-offset-ns": "2"},
 107             ).query()
 108
 109     def test_clock_class_offset_ns_wrong_type_none(self):
 110         with self.assertRaises(bt2._Error):
 111             bt2.QueryExecutor(
 112                 self._fs,
 113                 "babeltrace.trace-infos",
 114                 {"inputs": self._inputs, "clock-class-offset-ns": None},
 115             ).query()
 116
 117
 118 class QueryTraceInfoPortNameTestCase(unittest.TestCase):
 119     def setUp(self):
 120         ctf = bt2.find_plugin("ctf")
 121         self._fs = ctf.source_component_classes["fs"]
 122
 123     def test_trace_uuid_stream_class_id_no_stream_id(self):
 124         res = bt2.QueryExecutor(
 125             self._fs,
 126             "babeltrace.trace-infos",
 127             {
 128                 "inputs": [
 129                     os.path.join(
 130                         test_ctf_traces_path, "intersection", "3eventsintersect"
 131                     )
 132                 ]
 133             },
 134         ).query()
 135
 136         if os.environ["BT_TESTS_OS_TYPE"] == "mingw":
 137             os_stream_path = (
 138                 "\\tests\\data\\ctf-traces\\intersection\\3eventsintersect\\"
 139             )
 140         else:
 141             os_stream_path = "/tests/data/ctf-traces/intersection/3eventsintersect/"
 142
 143         self.assertEqual(len(res), 1)
 144         trace = res[0]
 145         streams = sorted(trace["stream-infos"], key=sort_predictably)
 146         self.assertEqual(len(streams), 2)
 147         self.assertRegex(
 148             str(streams[0]["port-name"]),
 149             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
 150             + re.escape(os_stream_path + "test_stream_0")
 151             + r"$",
 152         )
 153         self.assertRegex(
 154             str(streams[1]["port-name"]),
 155             r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
 156             + re.escape(os_stream_path + "test_stream_1")
 157             + r"$",
 158         )
 159
 160     def test_trace_uuid_no_stream_class_id_no_stream_id(self):
 161         res = bt2.QueryExecutor(
 162             self._fs,
 163             "babeltrace.trace-infos",
 164             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 165         ).query()
 166
 167         if os.environ["BT_TESTS_OS_TYPE"] == "mingw":
 168             os_stream_path = "\\tests\\data\\ctf-traces\\succeed\\succeed1\\dummystream"
 169         else:
 170             os_stream_path = "/tests/data/ctf-traces/succeed/succeed1/dummystream"
 171
 172         self.assertEqual(len(res), 1)
 173         trace = res[0]
 174         streams = sorted(trace["stream-infos"], key=sort_predictably)
 175         self.assertEqual(len(streams), 1)
 176         self.assertRegex(
 177             str(streams[0]["port-name"]),
 178             r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
 179             + re.escape(os_stream_path)
 180             + r"$",
 181         )
 182
 183
 184 class QueryTraceInfoRangeTestCase(unittest.TestCase):
 185     def setUp(self):
 186         ctf = bt2.find_plugin("ctf")
 187         self._fs = ctf.source_component_classes["fs"]
 188
 189     def test_trace_no_range(self):
 190         # This trace has no `timestamp_begin` and `timestamp_end` in its
 191         # packet context. The `babeltrace.trace-infos` query should omit
 192         # the `range-ns` fields in the `trace` and `stream` data
 193         # structures.
 194
 195         res = bt2.QueryExecutor(
 196             self._fs,
 197             "babeltrace.trace-infos",
 198             {"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
 199         ).query()
 200
 201         self.assertEqual(len(res), 1)
 202         trace = res[0]
 203         streams = trace["stream-infos"]
 204         self.assertEqual(len(streams), 1)
 205
 206         self.assertRaises(KeyError, lambda: trace["range-ns"])
 207         self.assertRaises(KeyError, lambda: streams[0]["range-ns"])
 208
 209     def test_trace_with_tracefile_rotation(self):
 210         res = bt2.QueryExecutor(
 211             self._fs,
 212             "babeltrace.trace-infos",
 213             {
 214                 "inputs": [
 215                     os.path.join(
 216                         test_ctf_traces_path,
 217                         "succeed",
 218                         "lttng-tracefile-rotation",
 219                         "kernel",
 220                     )
 221                 ]
 222             },
 223         ).query()
 224
 225         self.assertEqual(len(res), 1)
 226         trace = res[0]
 227         streams = trace["stream-infos"]
 228         self.assertEqual(len(streams), 4)
 229
 230         # Note: the end timestamps are not the end timestamps found in the
 231         # index files, because fix_index_lttng_event_after_packet_bug changes
 232         # them based on the time of the last event in the stream.
 233
 234         self.assertEqual(streams[0]["range-ns"]["begin"], 1571261795455986789)
 235         self.assertEqual(streams[0]["range-ns"]["end"], 1571261797582611840)
 236
 237         self.assertEqual(streams[1]["range-ns"]["begin"], 1571261795456368232)
 238         self.assertEqual(streams[1]["range-ns"]["end"], 1571261797577754111)
 239
 240         self.assertEqual(streams[2]["range-ns"]["begin"], 1571261795456748255)
 241         self.assertEqual(streams[2]["range-ns"]["end"], 1571261797577727795)
 242
 243         self.assertEqual(streams[3]["range-ns"]["begin"], 1571261795457285142)
 244         self.assertEqual(streams[3]["range-ns"]["end"], 1571261797582522088)
 245
 246
 247 class QueryTraceInfoPacketTimestampQuirksTestCase(unittest.TestCase):
 248     def setUp(self):
 249         ctf = bt2.find_plugin("ctf")
 250         self._fs = ctf.source_component_classes["fs"]
 251         self._path = os.path.join(test_ctf_traces_path, "succeed")
 252
 253     def _test_lttng_quirks(self, trace_name):
 254         res = bt2.QueryExecutor(
 255             self._fs,
 256             "babeltrace.trace-infos",
 257             {"inputs": [os.path.join(self._path, trace_name)]},
 258         ).query()
 259
 260         self.assertEqual(len(res), 1)
 261         return res[0]
 262
 263     def test_event_after_packet(self):
 264         trace = self._test_lttng_quirks("lttng-event-after-packet")
 265         streams = trace["stream-infos"]
 266         self.assertEqual(len(streams), 1)
 267
 268         self.assertEqual(streams[0]["range-ns"]["begin"], 1565957300948091100)
 269         self.assertEqual(streams[0]["range-ns"]["end"], 1565957302180016069)
 270
 271     def test_lttng_crash(self):
 272         trace = self._test_lttng_quirks("lttng-crash")
 273         streams = trace["stream-infos"]
 274         self.assertEqual(len(streams), 1)
 275
 276         self.assertEqual(streams[0]["range-ns"]["begin"], 1565891729288866738)
 277         self.assertEqual(streams[0]["range-ns"]["end"], 1565891729293526525)
 278
 279
 280 if __name__ == "__main__":
 281     unittest.main()