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