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