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