tests/bindings/python/bt2/test_trace_collection_message_iterator.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import os
   7 import os.path
   8 import datetime
   9 import unittest
  10
  11 import bt2
  12
  13 _BT_TESTS_DATADIR = os.environ["BT_TESTS_DATADIR"]
  14 _BT_CTF_TRACES_PATH = os.environ["BT_CTF_TRACES_PATH"]
  15 _3EVENTS_INTERSECT_TRACE_PATH = os.path.join(
  16     _BT_CTF_TRACES_PATH, "intersection", "3eventsintersect"
  17 )
  18 _NOINTERSECT_TRACE_PATH = os.path.join(
  19     _BT_CTF_TRACES_PATH, "intersection", "nointersect"
  20 )
  21 _SEQUENCE_TRACE_PATH = os.path.join(_BT_CTF_TRACES_PATH, "succeed", "sequence")
  22 _AUTO_SOURCE_DISCOVERY_GROUPING_PATH = os.path.join(
  23     _BT_TESTS_DATADIR, "auto-source-discovery", "grouping"
  24 )
  25 _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH = os.path.join(
  26     _BT_TESTS_DATADIR, "auto-source-discovery", "params-log-level"
  27 )
  28
  29 _METADATA_SYNTAX_ERROR_TRACE_PATH = os.path.join(
  30     _BT_CTF_TRACES_PATH, "fail", "metadata-syntax-error"
  31 )
  32 _BT_ENABLE_PYTHON_PLUGINS = os.environ["BT_TESTS_ENABLE_PYTHON_PLUGINS"] == "1"
  33
  34
  35 class _SomeSource(
  36     bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  37 ):
  38     pass
  39
  40
  41 class _SomeFilter(
  42     bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
  43 ):
  44     pass
  45
  46
  47 class _SomeSink(bt2._UserSinkComponent):
  48     def _user_consume(self):
  49         pass
  50
  51
  52 class ComponentSpecTestCase(unittest.TestCase):
  53     def setUp(self):
  54         # A source CC from a plugin.
  55         self._dmesg_cc = bt2.find_plugin("text").source_component_classes["dmesg"]
  56         assert self._dmesg_cc is not None
  57
  58         # A filter CC from a plugin.
  59         self._muxer_cc = bt2.find_plugin("utils").filter_component_classes["muxer"]
  60         assert self._muxer_cc is not None
  61
  62         # A sink CC from a plugin.
  63         self._pretty_cc = bt2.find_plugin("text").sink_component_classes["pretty"]
  64         assert self._pretty_cc is not None
  65
  66     def test_create_source_from_name(self):
  67         spec = bt2.ComponentSpec.from_named_plugin_and_component_class("text", "dmesg")
  68         self.assertEqual(spec.component_class.name, "dmesg")
  69
  70     def test_create_source_from_plugin(self):
  71         spec = bt2.ComponentSpec(self._dmesg_cc)
  72         self.assertEqual(spec.component_class.name, "dmesg")
  73
  74     def test_create_source_from_user(self):
  75         spec = bt2.ComponentSpec(_SomeSource)
  76         self.assertEqual(spec.component_class.name, "_SomeSource")
  77
  78     def test_create_filter_from_name(self):
  79         spec = bt2.ComponentSpec.from_named_plugin_and_component_class("utils", "muxer")
  80         self.assertEqual(spec.component_class.name, "muxer")
  81
  82     def test_create_filter_from_object(self):
  83         spec = bt2.ComponentSpec(self._muxer_cc)
  84         self.assertEqual(spec.component_class.name, "muxer")
  85
  86     def test_create_sink_from_name(self):
  87         with self.assertRaisesRegex(
  88             KeyError,
  89             "source or filter component class `pretty` not found in plugin `text`",
  90         ):
  91             bt2.ComponentSpec.from_named_plugin_and_component_class("text", "pretty")
  92
  93     def test_create_sink_from_object(self):
  94         with self.assertRaisesRegex(
  95             TypeError,
  96             "'_SinkComponentClassConst' is not a source or filter component class",
  97         ):
  98             bt2.ComponentSpec(self._pretty_cc)
  99
 100     def test_create_from_object_with_params(self):
 101         spec = bt2.ComponentSpec(self._dmesg_cc, {"salut": 23})
 102         self.assertEqual(spec.params["salut"], 23)
 103
 104     def test_create_from_name_with_params(self):
 105         spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 106             "text", "dmesg", {"salut": 23}
 107         )
 108         self.assertEqual(spec.params["salut"], 23)
 109
 110     def test_create_from_object_with_path_params(self):
 111         spec = spec = bt2.ComponentSpec(self._dmesg_cc, "a path")
 112         self.assertEqual(spec.params["inputs"], ["a path"])
 113
 114     def test_create_from_name_with_path_params(self):
 115         spec = spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 116             "text", "dmesg", "a path"
 117         )
 118         self.assertEqual(spec.params["inputs"], ["a path"])
 119
 120     def test_create_wrong_comp_class_type(self):
 121         with self.assertRaisesRegex(
 122             TypeError, "'int' is not a source or filter component class"
 123         ):
 124             bt2.ComponentSpec(18)
 125
 126     def test_create_from_name_wrong_plugin_name_type(self):
 127         with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
 128             bt2.ComponentSpec.from_named_plugin_and_component_class(23, "compcls")
 129
 130     def test_create_from_name_non_existent_plugin(self):
 131         with self.assertRaisesRegex(
 132             ValueError, "no such plugin: this_plugin_does_not_exist"
 133         ):
 134             bt2.ComponentSpec.from_named_plugin_and_component_class(
 135                 "this_plugin_does_not_exist", "compcls"
 136             )
 137
 138     def test_create_from_name_wrong_component_class_name_type(self):
 139         with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
 140             bt2.ComponentSpec.from_named_plugin_and_component_class("utils", 190)
 141
 142     def test_create_wrong_params_type(self):
 143         with self.assertRaisesRegex(
 144             TypeError, "cannot create value object from 'datetime' object"
 145         ):
 146             bt2.ComponentSpec(self._dmesg_cc, params=datetime.datetime.now())
 147
 148     def test_create_from_name_wrong_params_type(self):
 149         with self.assertRaisesRegex(
 150             TypeError, "cannot create value object from 'datetime' object"
 151         ):
 152             bt2.ComponentSpec.from_named_plugin_and_component_class(
 153                 "text", "dmesg", datetime.datetime.now()
 154             )
 155
 156     def test_create_wrong_log_level_type(self):
 157         with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
 158             bt2.ComponentSpec(self._dmesg_cc, logging_level="banane")
 159
 160     def test_create_from_name_wrong_log_level_type(self):
 161         with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
 162             bt2.ComponentSpec.from_named_plugin_and_component_class(
 163                 "text", "dmesg", logging_level="banane"
 164             )
 165
 166
 167 # Return a map, msg type -> number of messages of this type.
 168
 169
 170 def _count_msgs_by_type(msgs):
 171     res = {}
 172
 173     for msg in msgs:
 174         t = type(msg)
 175         n = res.get(t, 0)
 176         res[t] = n + 1
 177
 178     return res
 179
 180
 181 class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
 182     def test_create_wrong_stream_intersection_mode_type(self):
 183         specs = [
 184             bt2.ComponentSpec.from_named_plugin_and_component_class(
 185                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 186             )
 187         ]
 188
 189         with self.assertRaises(TypeError):
 190             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
 191
 192     def test_create_wrong_begin_type(self):
 193         specs = [
 194             bt2.ComponentSpec.from_named_plugin_and_component_class(
 195                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 196             )
 197         ]
 198
 199         with self.assertRaises(TypeError):
 200             bt2.TraceCollectionMessageIterator(specs, begin="hi")
 201
 202     def test_create_wrong_end_type(self):
 203         specs = [
 204             bt2.ComponentSpec.from_named_plugin_and_component_class(
 205                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 206             )
 207         ]
 208
 209         with self.assertRaises(TypeError):
 210             bt2.TraceCollectionMessageIterator(specs, begin="lel")
 211
 212     def test_create_begin_s(self):
 213         specs = [
 214             bt2.ComponentSpec.from_named_plugin_and_component_class(
 215                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 216             )
 217         ]
 218         bt2.TraceCollectionMessageIterator(specs, begin=19457.918232)
 219
 220     def test_create_end_s(self):
 221         specs = [
 222             bt2.ComponentSpec.from_named_plugin_and_component_class(
 223                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 224             )
 225         ]
 226         bt2.TraceCollectionMessageIterator(specs, end=123.12312)
 227
 228     def test_create_begin_datetime(self):
 229         specs = [
 230             bt2.ComponentSpec.from_named_plugin_and_component_class(
 231                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 232             )
 233         ]
 234         bt2.TraceCollectionMessageIterator(specs, begin=datetime.datetime.now())
 235
 236     def test_create_end_datetime(self):
 237         specs = [
 238             bt2.ComponentSpec.from_named_plugin_and_component_class(
 239                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 240             )
 241         ]
 242         bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
 243
 244     def test_iter_no_intersection(self):
 245         specs = [
 246             bt2.ComponentSpec.from_named_plugin_and_component_class(
 247                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 248             )
 249         ]
 250         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 251         msgs = list(msg_iter)
 252         self.assertEqual(len(msgs), 28)
 253         hist = _count_msgs_by_type(msgs)
 254         self.assertEqual(hist[bt2._EventMessageConst], 8)
 255
 256     # Same as the above, but we pass a single spec instead of a spec list.
 257     def test_iter_specs_not_list(self):
 258         spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 259             "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 260         )
 261         msg_iter = bt2.TraceCollectionMessageIterator(spec)
 262         msgs = list(msg_iter)
 263         self.assertEqual(len(msgs), 28)
 264         hist = _count_msgs_by_type(msgs)
 265         self.assertEqual(hist[bt2._EventMessageConst], 8)
 266
 267     def test_iter_custom_filter(self):
 268         src_spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 269             "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 270         )
 271         flt_spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 272             "utils", "trimmer", {"end": "13515309.000000075"}
 273         )
 274         msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
 275         hist = _count_msgs_by_type(msg_iter)
 276         self.assertEqual(hist[bt2._EventMessageConst], 5)
 277
 278     def test_iter_intersection(self):
 279         specs = [
 280             bt2.ComponentSpec.from_named_plugin_and_component_class(
 281                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 282             )
 283         ]
 284         msg_iter = bt2.TraceCollectionMessageIterator(
 285             specs, stream_intersection_mode=True
 286         )
 287         msgs = list(msg_iter)
 288         self.assertEqual(len(msgs), 15)
 289         hist = _count_msgs_by_type(msgs)
 290         self.assertEqual(hist[bt2._EventMessageConst], 3)
 291
 292     def test_iter_intersection_params(self):
 293         # Check that all params used to create the source component are passed
 294         # to the `babeltrace.trace-infos` query.
 295         specs = [
 296             bt2.ComponentSpec.from_named_plugin_and_component_class(
 297                 "ctf",
 298                 "fs",
 299                 {
 300                     "inputs": [_3EVENTS_INTERSECT_TRACE_PATH],
 301                     "clock-class-offset-s": 1000,
 302                 },
 303             )
 304         ]
 305
 306         msg_iter = bt2.TraceCollectionMessageIterator(
 307             specs, stream_intersection_mode=True
 308         )
 309
 310         event_msgs = [x for x in msg_iter if type(x) is bt2._EventMessageConst]
 311         self.assertEqual(len(event_msgs), 3)
 312         self.assertEqual(
 313             event_msgs[0].default_clock_snapshot.ns_from_origin, 13516309000000071
 314         )
 315         self.assertEqual(
 316             event_msgs[1].default_clock_snapshot.ns_from_origin, 13516309000000072
 317         )
 318         self.assertEqual(
 319             event_msgs[2].default_clock_snapshot.ns_from_origin, 13516309000000082
 320         )
 321
 322     def test_iter_no_intersection_two_traces(self):
 323         spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 324             "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 325         )
 326         specs = [spec, spec]
 327         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 328         msgs = list(msg_iter)
 329         self.assertEqual(len(msgs), 56)
 330         hist = _count_msgs_by_type(msgs)
 331         self.assertEqual(hist[bt2._EventMessageConst], 16)
 332
 333     def test_iter_no_intersection_begin(self):
 334         specs = [
 335             bt2.ComponentSpec.from_named_plugin_and_component_class(
 336                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 337             )
 338         ]
 339         msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
 340         hist = _count_msgs_by_type(msg_iter)
 341         self.assertEqual(hist[bt2._EventMessageConst], 6)
 342
 343     def test_iter_no_intersection_end(self):
 344         specs = [
 345             bt2.ComponentSpec.from_named_plugin_and_component_class(
 346                 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
 347             )
 348         ]
 349         msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
 350         hist = _count_msgs_by_type(msg_iter)
 351         self.assertEqual(hist[bt2._EventMessageConst], 5)
 352
 353     def test_iter_auto_source_component_spec(self):
 354         specs = [bt2.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH)]
 355         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 356         msgs = list(msg_iter)
 357         self.assertEqual(len(msgs), 28)
 358         hist = _count_msgs_by_type(msgs)
 359         self.assertEqual(hist[bt2._EventMessageConst], 8)
 360
 361     def test_iter_auto_source_component_spec_list_of_strings(self):
 362         msg_iter = bt2.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH])
 363         msgs = list(msg_iter)
 364         self.assertEqual(len(msgs), 28)
 365         hist = _count_msgs_by_type(msgs)
 366         self.assertEqual(hist[bt2._EventMessageConst], 8)
 367
 368     def test_iter_auto_source_component_spec_string(self):
 369         msg_iter = bt2.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH)
 370         msgs = list(msg_iter)
 371         self.assertEqual(len(msgs), 28)
 372         hist = _count_msgs_by_type(msgs)
 373         self.assertEqual(hist[bt2._EventMessageConst], 8)
 374
 375     def test_iter_mixed_inputs(self):
 376         msg_iter = bt2.TraceCollectionMessageIterator(
 377             [
 378                 _3EVENTS_INTERSECT_TRACE_PATH,
 379                 bt2.AutoSourceComponentSpec(_SEQUENCE_TRACE_PATH),
 380                 bt2.ComponentSpec.from_named_plugin_and_component_class(
 381                     "ctf", "fs", _NOINTERSECT_TRACE_PATH
 382                 ),
 383             ]
 384         )
 385         msgs = list(msg_iter)
 386         self.assertEqual(len(msgs), 76)
 387         hist = _count_msgs_by_type(msgs)
 388         self.assertEqual(hist[bt2._EventMessageConst], 24)
 389
 390     def test_auto_source_component_non_existent(self):
 391         with self.assertRaisesRegex(
 392             RuntimeError,
 393             "Some auto source component specs did not produce any component",
 394         ):
 395             # Test with one path known to contain a trace and one path known
 396             # to not contain any trace.
 397             bt2.TraceCollectionMessageIterator(
 398                 [_SEQUENCE_TRACE_PATH, "/this/path/better/not/exist"]
 399             )
 400
 401
 402 class _TestAutoDiscoverSourceComponentSpecs(unittest.TestCase):
 403     def setUp(self):
 404         self._saved_babeltrace_plugin_path = os.environ["BABELTRACE_PLUGIN_PATH"]
 405         os.environ["BABELTRACE_PLUGIN_PATH"] += os.pathsep + self._plugin_path
 406
 407     def tearDown(self):
 408         os.environ["BABELTRACE_PLUGIN_PATH"] = self._saved_babeltrace_plugin_path
 409
 410
 411 @unittest.skipUnless(
 412     _BT_ENABLE_PYTHON_PLUGINS,
 413     "Support for Python plugins is disabled",
 414 )
 415 class TestAutoDiscoverSourceComponentSpecsGrouping(
 416     _TestAutoDiscoverSourceComponentSpecs
 417 ):
 418     _plugin_path = _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
 419
 420     def test_grouping(self):
 421         specs = [
 422             bt2.AutoSourceComponentSpec("ABCDE"),
 423             bt2.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH),
 424         ]
 425         it = bt2.TraceCollectionMessageIterator(specs)
 426         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 427
 428         self.assertEqual(len(msgs), 8)
 429
 430         self.assertEqual(msgs[0].stream.name, "TestSourceABCDE: ABCDE")
 431         self.assertEqual(msgs[1].stream.name, "TestSourceExt: aaa1, aaa2, aaa3")
 432         self.assertEqual(msgs[2].stream.name, "TestSourceExt: bbb1, bbb2")
 433         self.assertEqual(msgs[3].stream.name, "TestSourceExt: ccc1")
 434         self.assertEqual(msgs[4].stream.name, "TestSourceExt: ccc2")
 435         self.assertEqual(msgs[5].stream.name, "TestSourceExt: ccc3")
 436         self.assertEqual(msgs[6].stream.name, "TestSourceExt: ccc4")
 437         self.assertEqual(msgs[7].stream.name, "TestSourceSomeDir: some-dir")
 438
 439
 440 @unittest.skipUnless(
 441     _BT_ENABLE_PYTHON_PLUGINS,
 442     "Support for Python plugins is disabled",
 443 )
 444 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
 445     _TestAutoDiscoverSourceComponentSpecs
 446 ):
 447     _plugin_path = _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
 448
 449     _dir_a = os.path.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH, "dir-a")
 450     _dir_b = os.path.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH, "dir-b")
 451     _dir_ab = os.path.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH, "dir-ab")
 452
 453     def _test_two_comps_from_one_spec(self, params, obj=None, logging_level=None):
 454         specs = [
 455             bt2.AutoSourceComponentSpec(
 456                 self._dir_ab, params=params, obj=obj, logging_level=logging_level
 457             )
 458         ]
 459         it = bt2.TraceCollectionMessageIterator(specs)
 460         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 461
 462         self.assertEqual(len(msgs), 2)
 463
 464         return msgs
 465
 466     def test_params_two_comps_from_one_spec(self):
 467         msgs = self._test_two_comps_from_one_spec(
 468             params={"test-allo": "madame", "what": "test-params"}
 469         )
 470
 471         self.assertEqual(msgs[0].stream.name, "TestSourceA: ('test-allo', 'madame')")
 472         self.assertEqual(msgs[1].stream.name, "TestSourceB: ('test-allo', 'madame')")
 473
 474     def test_obj_two_comps_from_one_spec(self):
 475         msgs = self._test_two_comps_from_one_spec(
 476             params={"what": "python-obj"}, obj="deore"
 477         )
 478
 479         self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
 480         self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
 481
 482     def test_log_level_two_comps_from_one_spec(self):
 483         msgs = self._test_two_comps_from_one_spec(
 484             params={"what": "log-level"}, logging_level=bt2.LoggingLevel.DEBUG
 485         )
 486
 487         self.assertEqual(
 488             msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.DEBUG)
 489         )
 490         self.assertEqual(
 491             msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.DEBUG)
 492         )
 493
 494     def _test_two_comps_from_two_specs(
 495         self,
 496         params_a=None,
 497         params_b=None,
 498         obj_a=None,
 499         obj_b=None,
 500         logging_level_a=None,
 501         logging_level_b=None,
 502     ):
 503         specs = [
 504             bt2.AutoSourceComponentSpec(
 505                 self._dir_a, params=params_a, obj=obj_a, logging_level=logging_level_a
 506             ),
 507             bt2.AutoSourceComponentSpec(
 508                 self._dir_b, params=params_b, obj=obj_b, logging_level=logging_level_b
 509             ),
 510         ]
 511         it = bt2.TraceCollectionMessageIterator(specs)
 512         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 513
 514         self.assertEqual(len(msgs), 2)
 515
 516         return msgs
 517
 518     def test_params_two_comps_from_two_specs(self):
 519         msgs = self._test_two_comps_from_two_specs(
 520             params_a={"test-allo": "madame", "what": "test-params"},
 521             params_b={"test-bonjour": "monsieur", "what": "test-params"},
 522         )
 523
 524         self.assertEqual(msgs[0].stream.name, "TestSourceA: ('test-allo', 'madame')")
 525         self.assertEqual(
 526             msgs[1].stream.name, "TestSourceB: ('test-bonjour', 'monsieur')"
 527         )
 528
 529     def test_obj_two_comps_from_two_specs(self):
 530         msgs = self._test_two_comps_from_two_specs(
 531             params_a={"what": "python-obj"},
 532             params_b={"what": "python-obj"},
 533             obj_a="deore",
 534             obj_b="alivio",
 535         )
 536
 537         self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
 538         self.assertEqual(msgs[1].stream.name, "TestSourceB: alivio")
 539
 540     def test_log_level_two_comps_from_two_specs(self):
 541         msgs = self._test_two_comps_from_two_specs(
 542             params_a={"what": "log-level"},
 543             params_b={"what": "log-level"},
 544             logging_level_a=bt2.LoggingLevel.DEBUG,
 545             logging_level_b=bt2.LoggingLevel.TRACE,
 546         )
 547
 548         self.assertEqual(
 549             msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.DEBUG)
 550         )
 551         self.assertEqual(
 552             msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.TRACE)
 553         )
 554
 555     def _test_one_comp_from_one_spec_one_comp_from_both_1(
 556         self,
 557         params_a=None,
 558         params_ab=None,
 559         obj_a=None,
 560         obj_ab=None,
 561         logging_level_a=None,
 562         logging_level_ab=None,
 563     ):
 564         specs = [
 565             bt2.AutoSourceComponentSpec(
 566                 self._dir_a, params=params_a, obj=obj_a, logging_level=logging_level_a
 567             ),
 568             bt2.AutoSourceComponentSpec(
 569                 self._dir_ab,
 570                 params=params_ab,
 571                 obj=obj_ab,
 572                 logging_level=logging_level_ab,
 573             ),
 574         ]
 575         it = bt2.TraceCollectionMessageIterator(specs)
 576         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 577
 578         self.assertEqual(len(msgs), 2)
 579
 580         return msgs
 581
 582     def test_params_one_comp_from_one_spec_one_comp_from_both_1(self):
 583         msgs = self._test_one_comp_from_one_spec_one_comp_from_both_1(
 584             params_a={"test-allo": "madame", "what": "test-params"},
 585             params_ab={"test-bonjour": "monsieur", "what": "test-params"},
 586         )
 587
 588         self.assertEqual(
 589             msgs[0].stream.name,
 590             "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
 591         )
 592         self.assertEqual(
 593             msgs[1].stream.name, "TestSourceB: ('test-bonjour', 'monsieur')"
 594         )
 595
 596     def test_obj_one_comp_from_one_spec_one_comp_from_both_1(self):
 597         msgs = self._test_one_comp_from_one_spec_one_comp_from_both_1(
 598             params_a={"what": "python-obj"},
 599             params_ab={"what": "python-obj"},
 600             obj_a="deore",
 601             obj_ab="alivio",
 602         )
 603
 604         self.assertEqual(msgs[0].stream.name, "TestSourceA: alivio")
 605         self.assertEqual(msgs[1].stream.name, "TestSourceB: alivio")
 606
 607     def test_log_level_one_comp_from_one_spec_one_comp_from_both_1(self):
 608         msgs = self._test_one_comp_from_one_spec_one_comp_from_both_1(
 609             params_a={"what": "log-level"},
 610             params_ab={"what": "log-level"},
 611             logging_level_a=bt2.LoggingLevel.DEBUG,
 612             logging_level_ab=bt2.LoggingLevel.TRACE,
 613         )
 614
 615         self.assertEqual(
 616             msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.TRACE)
 617         )
 618         self.assertEqual(
 619             msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.TRACE)
 620         )
 621
 622     def _test_one_comp_from_one_spec_one_comp_from_both_2(
 623         self,
 624         params_ab=None,
 625         params_a=None,
 626         obj_ab=None,
 627         obj_a=None,
 628         logging_level_ab=None,
 629         logging_level_a=None,
 630     ):
 631         specs = [
 632             bt2.AutoSourceComponentSpec(
 633                 self._dir_ab,
 634                 params=params_ab,
 635                 obj=obj_ab,
 636                 logging_level=logging_level_ab,
 637             ),
 638             bt2.AutoSourceComponentSpec(
 639                 self._dir_a, params=params_a, obj=obj_a, logging_level=logging_level_a
 640             ),
 641         ]
 642         it = bt2.TraceCollectionMessageIterator(specs)
 643         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 644
 645         self.assertEqual(len(msgs), 2)
 646
 647         return msgs
 648
 649     def test_params_one_comp_from_one_spec_one_comp_from_both_2(self):
 650         msgs = self._test_one_comp_from_one_spec_one_comp_from_both_2(
 651             params_ab={
 652                 "test-bonjour": "madame",
 653                 "test-salut": "les amis",
 654                 "what": "test-params",
 655             },
 656             params_a={"test-bonjour": "monsieur", "what": "test-params"},
 657         )
 658
 659         self.assertEqual(
 660             msgs[0].stream.name,
 661             "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
 662         )
 663         self.assertEqual(
 664             msgs[1].stream.name,
 665             "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
 666         )
 667
 668     def test_obj_one_comp_from_one_spec_one_comp_from_both_2(self):
 669         msgs = self._test_one_comp_from_one_spec_one_comp_from_both_2(
 670             params_ab={"what": "python-obj"},
 671             params_a={"what": "python-obj"},
 672             obj_ab="deore",
 673             obj_a="alivio",
 674         )
 675
 676         self.assertEqual(msgs[0].stream.name, "TestSourceA: alivio")
 677         self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
 678
 679     def test_log_level_one_comp_from_one_spec_one_comp_from_both_2(self):
 680         msgs = self._test_one_comp_from_one_spec_one_comp_from_both_2(
 681             params_ab={"what": "log-level"},
 682             params_a={"what": "log-level"},
 683             logging_level_ab=bt2.LoggingLevel.DEBUG,
 684             logging_level_a=bt2.LoggingLevel.TRACE,
 685         )
 686
 687         self.assertEqual(
 688             msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.TRACE)
 689         )
 690         self.assertEqual(
 691             msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.DEBUG)
 692         )
 693
 694     def test_obj_override_with_none(self):
 695         specs = [
 696             bt2.AutoSourceComponentSpec(
 697                 self._dir_ab, params={"what": "python-obj"}, obj="deore"
 698             ),
 699             bt2.AutoSourceComponentSpec(
 700                 self._dir_a, params={"what": "python-obj"}, obj=None
 701             ),
 702         ]
 703         it = bt2.TraceCollectionMessageIterator(specs)
 704         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 705
 706         self.assertEqual(len(msgs), 2)
 707         self.assertEqual(msgs[0].stream.name, "TestSourceA: None")
 708         self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
 709
 710     def test_obj_no_override_with_no_obj(self):
 711         specs = [
 712             bt2.AutoSourceComponentSpec(
 713                 self._dir_ab, params={"what": "python-obj"}, obj="deore"
 714             ),
 715             bt2.AutoSourceComponentSpec(self._dir_a, params={"what": "python-obj"}),
 716         ]
 717         it = bt2.TraceCollectionMessageIterator(specs)
 718         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 719
 720         self.assertEqual(len(msgs), 2)
 721         self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
 722         self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
 723
 724
 725 class TestAutoDiscoverFailures(unittest.TestCase):
 726     def test_metadata_syntax_error(self):
 727         with self.assertRaisesRegex(
 728             bt2._Error,
 729             'At line 3 in metadata stream: syntax error, unexpected CTF_RSBRAC: token="]"',
 730         ):
 731             specs = [bt2.AutoSourceComponentSpec(_METADATA_SYNTAX_ERROR_TRACE_PATH)]
 732             bt2.TraceCollectionMessageIterator(specs)
 733
 734
 735 if __name__ == "__main__":
 736     unittest.main()