tests/bindings/python/bt2/test_trace_collection_message_iterator.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; only version 2
   7 # of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  17 #
  18
  19 import unittest
  20 import datetime
  21 import bt2
  22 import os
  23 import os.path
  24
  25
  26 _BT_TESTS_DATADIR = os.environ['BT_TESTS_DATADIR']
  27 _BT_CTF_TRACES_PATH = os.environ['BT_CTF_TRACES_PATH']
  28 _3EVENTS_INTERSECT_TRACE_PATH = os.path.join(
  29     _BT_CTF_TRACES_PATH, 'intersection', '3eventsintersect'
  30 )
  31 _NOINTERSECT_TRACE_PATH = os.path.join(
  32     _BT_CTF_TRACES_PATH, 'intersection', 'nointersect'
  33 )
  34 _SEQUENCE_TRACE_PATH = os.path.join(_BT_CTF_TRACES_PATH, 'succeed', 'sequence')
  35 _AUTO_SOURCE_DISCOVERY_GROUPING_PATH = os.path.join(
  36     _BT_TESTS_DATADIR, 'auto-source-discovery', 'grouping'
  37 )
  38 _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH = os.path.join(
  39     _BT_TESTS_DATADIR, 'auto-source-discovery', 'params-log-level'
  40 )
  41
  42
  43 class _SomeSource(
  44     bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  45 ):
  46     pass
  47
  48
  49 class _SomeFilter(
  50     bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
  51 ):
  52     pass
  53
  54
  55 class _SomeSink(bt2._UserSinkComponent):
  56     def _user_consume(self):
  57         pass
  58
  59
  60 class ComponentSpecTestCase(unittest.TestCase):
  61     def setUp(self):
  62         # A source CC from a plugin.
  63         self._dmesg_cc = bt2.find_plugin('text').source_component_classes['dmesg']
  64         assert self._dmesg_cc is not None
  65
  66         # A filter CC from a plugin.
  67         self._muxer_cc = bt2.find_plugin('utils').filter_component_classes['muxer']
  68         assert self._muxer_cc is not None
  69
  70         # A sink CC from a plugin.
  71         self._pretty_cc = bt2.find_plugin('text').sink_component_classes['pretty']
  72         assert self._pretty_cc is not None
  73
  74     def test_create_source_from_name(self):
  75         spec = bt2.ComponentSpec.from_named_plugin_and_component_class('text', 'dmesg')
  76         self.assertEqual(spec.component_class.name, 'dmesg')
  77
  78     def test_create_source_from_plugin(self):
  79         spec = bt2.ComponentSpec(self._dmesg_cc)
  80         self.assertEqual(spec.component_class.name, 'dmesg')
  81
  82     def test_create_source_from_user(self):
  83         spec = bt2.ComponentSpec(_SomeSource)
  84         self.assertEqual(spec.component_class.name, '_SomeSource')
  85
  86     def test_create_filter_from_name(self):
  87         spec = bt2.ComponentSpec.from_named_plugin_and_component_class('utils', 'muxer')
  88         self.assertEqual(spec.component_class.name, 'muxer')
  89
  90     def test_create_filter_from_object(self):
  91         spec = bt2.ComponentSpec(self._muxer_cc)
  92         self.assertEqual(spec.component_class.name, 'muxer')
  93
  94     def test_create_sink_from_name(self):
  95         with self.assertRaisesRegex(
  96             KeyError,
  97             'source or filter component class `pretty` not found in plugin `text`',
  98         ):
  99             bt2.ComponentSpec.from_named_plugin_and_component_class('text', 'pretty')
 100
 101     def test_create_sink_from_object(self):
 102         with self.assertRaisesRegex(
 103             TypeError,
 104             "'_SinkComponentClassConst' is not a source or filter component class",
 105         ):
 106             bt2.ComponentSpec(self._pretty_cc)
 107
 108     def test_create_from_object_with_params(self):
 109         spec = bt2.ComponentSpec(self._dmesg_cc, {'salut': 23})
 110         self.assertEqual(spec.params['salut'], 23)
 111
 112     def test_create_from_name_with_params(self):
 113         spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 114             'text', 'dmesg', {'salut': 23}
 115         )
 116         self.assertEqual(spec.params['salut'], 23)
 117
 118     def test_create_from_object_with_path_params(self):
 119         spec = spec = bt2.ComponentSpec(self._dmesg_cc, 'a path')
 120         self.assertEqual(spec.params['inputs'], ['a path'])
 121
 122     def test_create_from_name_with_path_params(self):
 123         spec = spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 124             'text', 'dmesg', 'a path'
 125         )
 126         self.assertEqual(spec.params['inputs'], ['a path'])
 127
 128     def test_create_wrong_comp_class_type(self):
 129         with self.assertRaisesRegex(
 130             TypeError, "'int' is not a source or filter component class"
 131         ):
 132             bt2.ComponentSpec(18)
 133
 134     def test_create_from_name_wrong_plugin_name_type(self):
 135         with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
 136             bt2.ComponentSpec.from_named_plugin_and_component_class(23, 'compcls')
 137
 138     def test_create_from_name_non_existent_plugin(self):
 139         with self.assertRaisesRegex(
 140             ValueError, "no such plugin: this_plugin_does_not_exist"
 141         ):
 142             bt2.ComponentSpec.from_named_plugin_and_component_class(
 143                 'this_plugin_does_not_exist', 'compcls'
 144             )
 145
 146     def test_create_from_name_wrong_component_class_name_type(self):
 147         with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
 148             bt2.ComponentSpec.from_named_plugin_and_component_class('utils', 190)
 149
 150     def test_create_wrong_params_type(self):
 151         with self.assertRaisesRegex(
 152             TypeError, "cannot create value object from 'datetime' object"
 153         ):
 154             bt2.ComponentSpec(self._dmesg_cc, params=datetime.datetime.now())
 155
 156     def test_create_from_name_wrong_params_type(self):
 157         with self.assertRaisesRegex(
 158             TypeError, "cannot create value object from 'datetime' object"
 159         ):
 160             bt2.ComponentSpec.from_named_plugin_and_component_class(
 161                 'text', 'dmesg', datetime.datetime.now()
 162             )
 163
 164     def test_create_wrong_log_level_type(self):
 165         with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
 166             bt2.ComponentSpec(self._dmesg_cc, logging_level='banane')
 167
 168     def test_create_from_name_wrong_log_level_type(self):
 169         with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
 170             bt2.ComponentSpec.from_named_plugin_and_component_class(
 171                 'text', 'dmesg', logging_level='banane'
 172             )
 173
 174
 175 # Return a map, msg type -> number of messages of this type.
 176
 177
 178 def _count_msgs_by_type(msgs):
 179     res = {}
 180
 181     for msg in msgs:
 182         t = type(msg)
 183         n = res.get(t, 0)
 184         res[t] = n + 1
 185
 186     return res
 187
 188
 189 class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
 190     def test_create_wrong_stream_intersection_mode_type(self):
 191         specs = [
 192             bt2.ComponentSpec.from_named_plugin_and_component_class(
 193                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 194             )
 195         ]
 196
 197         with self.assertRaises(TypeError):
 198             bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
 199
 200     def test_create_wrong_begin_type(self):
 201         specs = [
 202             bt2.ComponentSpec.from_named_plugin_and_component_class(
 203                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 204             )
 205         ]
 206
 207         with self.assertRaises(TypeError):
 208             bt2.TraceCollectionMessageIterator(specs, begin='hi')
 209
 210     def test_create_wrong_end_type(self):
 211         specs = [
 212             bt2.ComponentSpec.from_named_plugin_and_component_class(
 213                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 214             )
 215         ]
 216
 217         with self.assertRaises(TypeError):
 218             bt2.TraceCollectionMessageIterator(specs, begin='lel')
 219
 220     def test_create_begin_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, begin=19457.918232)
 227
 228     def test_create_end_s(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, end=123.12312)
 235
 236     def test_create_begin_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, begin=datetime.datetime.now())
 243
 244     def test_create_end_datetime(self):
 245         specs = [
 246             bt2.ComponentSpec.from_named_plugin_and_component_class(
 247                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 248             )
 249         ]
 250         bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
 251
 252     def test_iter_no_intersection(self):
 253         specs = [
 254             bt2.ComponentSpec.from_named_plugin_and_component_class(
 255                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 256             )
 257         ]
 258         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 259         msgs = list(msg_iter)
 260         self.assertEqual(len(msgs), 28)
 261         hist = _count_msgs_by_type(msgs)
 262         self.assertEqual(hist[bt2._EventMessageConst], 8)
 263
 264     # Same as the above, but we pass a single spec instead of a spec list.
 265     def test_iter_specs_not_list(self):
 266         spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 267             'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 268         )
 269         msg_iter = bt2.TraceCollectionMessageIterator(spec)
 270         msgs = list(msg_iter)
 271         self.assertEqual(len(msgs), 28)
 272         hist = _count_msgs_by_type(msgs)
 273         self.assertEqual(hist[bt2._EventMessageConst], 8)
 274
 275     def test_iter_custom_filter(self):
 276         src_spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 277             'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 278         )
 279         flt_spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 280             'utils', 'trimmer', {'end': '13515309.000000075'}
 281         )
 282         msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
 283         hist = _count_msgs_by_type(msg_iter)
 284         self.assertEqual(hist[bt2._EventMessageConst], 5)
 285
 286     def test_iter_intersection(self):
 287         specs = [
 288             bt2.ComponentSpec.from_named_plugin_and_component_class(
 289                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 290             )
 291         ]
 292         msg_iter = bt2.TraceCollectionMessageIterator(
 293             specs, stream_intersection_mode=True
 294         )
 295         msgs = list(msg_iter)
 296         self.assertEqual(len(msgs), 15)
 297         hist = _count_msgs_by_type(msgs)
 298         self.assertEqual(hist[bt2._EventMessageConst], 3)
 299
 300     def test_iter_intersection_params(self):
 301         # Check that all params used to create the source component are passed
 302         # to the `babeltrace.trace-infos` query.
 303         specs = [
 304             bt2.ComponentSpec.from_named_plugin_and_component_class(
 305                 'ctf',
 306                 'fs',
 307                 {
 308                     'inputs': [_3EVENTS_INTERSECT_TRACE_PATH],
 309                     'clock-class-offset-s': 1000,
 310                 },
 311             )
 312         ]
 313
 314         msg_iter = bt2.TraceCollectionMessageIterator(
 315             specs, stream_intersection_mode=True
 316         )
 317
 318         event_msgs = [x for x in msg_iter if type(x) is bt2._EventMessageConst]
 319         self.assertEqual(len(event_msgs), 3)
 320         self.assertEqual(
 321             event_msgs[0].default_clock_snapshot.ns_from_origin, 13516309000000071
 322         )
 323         self.assertEqual(
 324             event_msgs[1].default_clock_snapshot.ns_from_origin, 13516309000000072
 325         )
 326         self.assertEqual(
 327             event_msgs[2].default_clock_snapshot.ns_from_origin, 13516309000000082
 328         )
 329
 330     def test_iter_no_intersection_two_traces(self):
 331         spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
 332             'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 333         )
 334         specs = [spec, spec]
 335         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 336         msgs = list(msg_iter)
 337         self.assertEqual(len(msgs), 56)
 338         hist = _count_msgs_by_type(msgs)
 339         self.assertEqual(hist[bt2._EventMessageConst], 16)
 340
 341     def test_iter_no_intersection_begin(self):
 342         specs = [
 343             bt2.ComponentSpec.from_named_plugin_and_component_class(
 344                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 345             )
 346         ]
 347         msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
 348         hist = _count_msgs_by_type(msg_iter)
 349         self.assertEqual(hist[bt2._EventMessageConst], 6)
 350
 351     def test_iter_no_intersection_end(self):
 352         specs = [
 353             bt2.ComponentSpec.from_named_plugin_and_component_class(
 354                 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
 355             )
 356         ]
 357         msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
 358         hist = _count_msgs_by_type(msg_iter)
 359         self.assertEqual(hist[bt2._EventMessageConst], 5)
 360
 361     def test_iter_auto_source_component_spec(self):
 362         specs = [bt2.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH)]
 363         msg_iter = bt2.TraceCollectionMessageIterator(specs)
 364         msgs = list(msg_iter)
 365         self.assertEqual(len(msgs), 28)
 366         hist = _count_msgs_by_type(msgs)
 367         self.assertEqual(hist[bt2._EventMessageConst], 8)
 368
 369     def test_iter_auto_source_component_spec_list_of_strings(self):
 370         msg_iter = bt2.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH])
 371         msgs = list(msg_iter)
 372         self.assertEqual(len(msgs), 28)
 373         hist = _count_msgs_by_type(msgs)
 374         self.assertEqual(hist[bt2._EventMessageConst], 8)
 375
 376     def test_iter_auto_source_component_spec_string(self):
 377         msg_iter = bt2.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH)
 378         msgs = list(msg_iter)
 379         self.assertEqual(len(msgs), 28)
 380         hist = _count_msgs_by_type(msgs)
 381         self.assertEqual(hist[bt2._EventMessageConst], 8)
 382
 383     def test_iter_mixed_inputs(self):
 384         msg_iter = bt2.TraceCollectionMessageIterator(
 385             [
 386                 _3EVENTS_INTERSECT_TRACE_PATH,
 387                 bt2.AutoSourceComponentSpec(_SEQUENCE_TRACE_PATH),
 388                 bt2.ComponentSpec.from_named_plugin_and_component_class(
 389                     'ctf', 'fs', _NOINTERSECT_TRACE_PATH
 390                 ),
 391             ]
 392         )
 393         msgs = list(msg_iter)
 394         self.assertEqual(len(msgs), 76)
 395         hist = _count_msgs_by_type(msgs)
 396         self.assertEqual(hist[bt2._EventMessageConst], 24)
 397
 398     def test_auto_source_component_non_existent(self):
 399         with self.assertRaisesRegex(
 400             RuntimeError,
 401             'Some auto source component specs did not produce any component',
 402         ):
 403             # Test with one path known to contain a trace and one path known
 404             # to not contain any trace.
 405             bt2.TraceCollectionMessageIterator(
 406                 [_SEQUENCE_TRACE_PATH, '/this/path/better/not/exist']
 407             )
 408
 409
 410 class _TestAutoDiscoverSourceComponentSpecs(unittest.TestCase):
 411     def setUp(self):
 412         self._saved_babeltrace_plugin_path = os.environ['BABELTRACE_PLUGIN_PATH']
 413         os.environ['BABELTRACE_PLUGIN_PATH'] += os.pathsep + self._plugin_path
 414
 415     def tearDown(self):
 416         os.environ['BABELTRACE_PLUGIN_PATH'] = self._saved_babeltrace_plugin_path
 417
 418
 419 class TestAutoDiscoverSourceComponentSpecsGrouping(
 420     _TestAutoDiscoverSourceComponentSpecs
 421 ):
 422     _plugin_path = _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
 423
 424     def test_grouping(self):
 425         specs = [
 426             bt2.AutoSourceComponentSpec('ABCDE'),
 427             bt2.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH),
 428         ]
 429         it = bt2.TraceCollectionMessageIterator(specs)
 430         msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
 431
 432         self.assertEqual(len(msgs), 8)
 433
 434         self.assertEqual(msgs[0].stream.name, 'TestSourceABCDE: ABCDE')
 435         self.assertEqual(msgs[1].stream.name, 'TestSourceExt: aaa1, aaa2, aaa3')
 436         self.assertEqual(msgs[2].stream.name, 'TestSourceExt: bbb1, bbb2')
 437         self.assertEqual(msgs[3].stream.name, 'TestSourceExt: ccc1')
 438         self.assertEqual(msgs[4].stream.name, 'TestSourceExt: ccc2')
 439         self.assertEqual(msgs[5].stream.name, 'TestSourceExt: ccc3')
 440         self.assertEqual(msgs[6].stream.name, 'TestSourceExt: ccc4')
 441         self.assertEqual(msgs[7].stream.name, 'TestSourceSomeDir: some-dir')
 442
 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 if __name__ == '__main__':
 726     unittest.main()