1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
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"
18 _NOINTERSECT_TRACE_PATH
= os
.path
.join(
19 _BT_CTF_TRACES_PATH
, "intersection", "nointersect"
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"
25 _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
= os
.path
.join(
26 _BT_TESTS_DATADIR
, "auto-source-discovery", "params-log-level"
31 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
37 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
42 class _SomeSink(bt2
._UserSinkComponent
):
43 def _user_consume(self
):
47 class ComponentSpecTestCase(unittest
.TestCase
):
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
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
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
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")
65 def test_create_source_from_plugin(self
):
66 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
)
67 self
.assertEqual(spec
.component_class
.name
, "dmesg")
69 def test_create_source_from_user(self
):
70 spec
= bt2
.ComponentSpec(_SomeSource
)
71 self
.assertEqual(spec
.component_class
.name
, "_SomeSource")
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")
77 def test_create_filter_from_object(self
):
78 spec
= bt2
.ComponentSpec(self
._muxer
_cc
)
79 self
.assertEqual(spec
.component_class
.name
, "muxer")
81 def test_create_sink_from_name(self
):
82 with self
.assertRaisesRegex(
84 "source or filter component class `pretty` not found in plugin `text`",
86 bt2
.ComponentSpec
.from_named_plugin_and_component_class("text", "pretty")
88 def test_create_sink_from_object(self
):
89 with self
.assertRaisesRegex(
91 "'_SinkComponentClassConst' is not a source or filter component class",
93 bt2
.ComponentSpec(self
._pretty
_cc
)
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)
99 def test_create_from_name_with_params(self
):
100 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
101 "text", "dmesg", {"salut": 23}
103 self
.assertEqual(spec
.params
["salut"], 23)
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"])
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"
113 self
.assertEqual(spec
.params
["inputs"], ["a path"])
115 def test_create_wrong_comp_class_type(self
):
116 with self
.assertRaisesRegex(
117 TypeError, "'int' is not a source or filter component class"
119 bt2
.ComponentSpec(18)
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")
125 def test_create_from_name_non_existent_plugin(self
):
126 with self
.assertRaisesRegex(
127 ValueError, "no such plugin: this_plugin_does_not_exist"
129 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
130 "this_plugin_does_not_exist", "compcls"
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)
137 def test_create_wrong_params_type(self
):
138 with self
.assertRaisesRegex(
139 TypeError, "cannot create value object from 'datetime' object"
141 bt2
.ComponentSpec(self
._dmesg
_cc
, params
=datetime
.datetime
.now())
143 def test_create_from_name_wrong_params_type(self
):
144 with self
.assertRaisesRegex(
145 TypeError, "cannot create value object from 'datetime' object"
147 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
148 "text", "dmesg", datetime
.datetime
.now()
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")
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"
162 # Return a map, msg type -> number of messages of this type.
165 def _count_msgs_by_type(msgs
):
176 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
177 def test_create_wrong_stream_intersection_mode_type(self
):
179 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
180 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
184 with self
.assertRaises(TypeError):
185 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
187 def test_create_wrong_begin_type(self
):
189 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
190 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
194 with self
.assertRaises(TypeError):
195 bt2
.TraceCollectionMessageIterator(specs
, begin
="hi")
197 def test_create_wrong_end_type(self
):
199 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
200 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
204 with self
.assertRaises(TypeError):
205 bt2
.TraceCollectionMessageIterator(specs
, begin
="lel")
207 def test_create_begin_s(self
):
209 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
210 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
213 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
215 def test_create_end_s(self
):
217 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
218 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
221 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
223 def test_create_begin_datetime(self
):
225 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
226 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
229 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
231 def test_create_end_datetime(self
):
233 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
234 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
237 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
239 def test_iter_no_intersection(self
):
241 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
242 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
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)
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
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)
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
266 flt_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
267 "utils", "trimmer", {"end": "13515309.000000075"}
269 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
270 hist
= _count_msgs_by_type(msg_iter
)
271 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
273 def test_iter_intersection(self
):
275 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
276 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
279 msg_iter
= bt2
.TraceCollectionMessageIterator(
280 specs
, stream_intersection_mode
=True
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)
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.
291 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
295 "inputs": [_3EVENTS_INTERSECT_TRACE_PATH
],
296 "clock-class-offset-s": 1000,
301 msg_iter
= bt2
.TraceCollectionMessageIterator(
302 specs
, stream_intersection_mode
=True
305 event_msgs
= [x
for x
in msg_iter
if type(x
) is bt2
._EventMessageConst
]
306 self
.assertEqual(len(event_msgs
), 3)
308 event_msgs
[0].default_clock_snapshot
.ns_from_origin
, 13516309000000071
311 event_msgs
[1].default_clock_snapshot
.ns_from_origin
, 13516309000000072
314 event_msgs
[2].default_clock_snapshot
.ns_from_origin
, 13516309000000082
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
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)
328 def test_iter_no_intersection_begin(self
):
330 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
331 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
334 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
335 hist
= _count_msgs_by_type(msg_iter
)
336 self
.assertEqual(hist
[bt2
._EventMessageConst
], 6)
338 def test_iter_no_intersection_end(self
):
340 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
341 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
344 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
345 hist
= _count_msgs_by_type(msg_iter
)
346 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
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)
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)
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)
370 def test_iter_mixed_inputs(self
):
371 msg_iter
= bt2
.TraceCollectionMessageIterator(
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
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)
385 def test_auto_source_component_non_existent(self
):
386 with self
.assertRaisesRegex(
388 "Some auto source component specs did not produce any component",
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"]
397 class _TestAutoDiscoverSourceComponentSpecs(unittest
.TestCase
):
399 self
._saved
_babeltrace
_plugin
_path
= os
.environ
["BABELTRACE_PLUGIN_PATH"]
400 os
.environ
["BABELTRACE_PLUGIN_PATH"] += os
.pathsep
+ self
._plugin
_path
403 os
.environ
["BABELTRACE_PLUGIN_PATH"] = self
._saved
_babeltrace
_plugin
_path
406 class TestAutoDiscoverSourceComponentSpecsGrouping(
407 _TestAutoDiscoverSourceComponentSpecs
409 _plugin_path
= _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
411 def test_grouping(self
):
413 bt2
.AutoSourceComponentSpec("ABCDE"),
414 bt2
.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH
),
416 it
= bt2
.TraceCollectionMessageIterator(specs
)
417 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
419 self
.assertEqual(len(msgs
), 8)
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")
431 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
432 _TestAutoDiscoverSourceComponentSpecs
434 _plugin_path
= _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
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")
440 def _test_two_comps_from_one_spec(self
, params
, obj
=None, logging_level
=None):
442 bt2
.AutoSourceComponentSpec(
443 self
._dir
_ab
, params
=params
, obj
=obj
, logging_level
=logging_level
446 it
= bt2
.TraceCollectionMessageIterator(specs
)
447 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
449 self
.assertEqual(len(msgs
), 2)
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"}
458 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
459 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: ('test-allo', 'madame')")
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"
466 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
467 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
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
475 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
478 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
481 def _test_two_comps_from_two_specs(
487 logging_level_a
=None,
488 logging_level_b
=None,
491 bt2
.AutoSourceComponentSpec(
492 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
494 bt2
.AutoSourceComponentSpec(
495 self
._dir
_b
, params
=params_b
, obj
=obj_b
, logging_level
=logging_level_b
498 it
= bt2
.TraceCollectionMessageIterator(specs
)
499 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
501 self
.assertEqual(len(msgs
), 2)
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"},
511 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
513 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
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"},
524 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
525 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
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
,
536 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
539 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
542 def _test_one_comp_from_one_spec_one_comp_from_both_1(
548 logging_level_a
=None,
549 logging_level_ab
=None,
552 bt2
.AutoSourceComponentSpec(
553 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
555 bt2
.AutoSourceComponentSpec(
559 logging_level
=logging_level_ab
,
562 it
= bt2
.TraceCollectionMessageIterator(specs
)
563 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
565 self
.assertEqual(len(msgs
), 2)
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"},
577 "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
580 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
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"},
591 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
592 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
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
,
603 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
606 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
609 def _test_one_comp_from_one_spec_one_comp_from_both_2(
615 logging_level_ab
=None,
616 logging_level_a
=None,
619 bt2
.AutoSourceComponentSpec(
623 logging_level
=logging_level_ab
,
625 bt2
.AutoSourceComponentSpec(
626 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
629 it
= bt2
.TraceCollectionMessageIterator(specs
)
630 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
632 self
.assertEqual(len(msgs
), 2)
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(
639 "test-bonjour": "madame",
640 "test-salut": "les amis",
641 "what": "test-params",
643 params_a
={"test-bonjour": "monsieur", "what": "test-params"},
648 "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
652 "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
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"},
663 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
664 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
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
,
675 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
678 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
681 def test_obj_override_with_none(self
):
683 bt2
.AutoSourceComponentSpec(
684 self
._dir
_ab
, params
={"what": "python-obj"}, obj
="deore"
686 bt2
.AutoSourceComponentSpec(
687 self
._dir
_a
, params
={"what": "python-obj"}, obj
=None
690 it
= bt2
.TraceCollectionMessageIterator(specs
)
691 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
693 self
.assertEqual(len(msgs
), 2)
694 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: None")
695 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
697 def test_obj_no_override_with_no_obj(self
):
699 bt2
.AutoSourceComponentSpec(
700 self
._dir
_ab
, params
={"what": "python-obj"}, obj
="deore"
702 bt2
.AutoSourceComponentSpec(self
._dir
_a
, params
={"what": "python-obj"}),
704 it
= bt2
.TraceCollectionMessageIterator(specs
)
705 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
707 self
.assertEqual(len(msgs
), 2)
708 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
709 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
712 if __name__
== "__main__":