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
, "1", "intersection", "3eventsintersect"
18 _NOINTERSECT_TRACE_PATH
= os
.path
.join(
19 _BT_CTF_TRACES_PATH
, "1", "intersection", "nointersect"
21 _SEQUENCE_TRACE_PATH
= os
.path
.join(_BT_CTF_TRACES_PATH
, "1", "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"
29 _METADATA_SYNTAX_ERROR_TRACE_PATH
= os
.path
.join(
30 _BT_CTF_TRACES_PATH
, "1", "fail", "metadata-syntax-error"
32 _BT_ENABLE_PYTHON_PLUGINS
= os
.environ
["BT_TESTS_ENABLE_PYTHON_PLUGINS"] == "1"
36 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
42 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
47 class _SomeSink(bt2
._UserSinkComponent
):
48 def _user_consume(self
):
52 class ComponentSpecTestCase(unittest
.TestCase
):
54 # A source CC from a plugin.
55 self
._dmesg
_cc
= bt2
.find_plugin("text").source_component_classes
["dmesg"]
56 assert self
._dmesg
_cc
is not None
58 # A filter CC from a plugin.
59 self
._muxer
_cc
= bt2
.find_plugin("utils").filter_component_classes
["muxer"]
60 assert self
._muxer
_cc
is not None
62 # A sink CC from a plugin.
63 self
._pretty
_cc
= bt2
.find_plugin("text").sink_component_classes
["pretty"]
64 assert self
._pretty
_cc
is not None
66 def test_create_source_from_name(self
):
67 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class("text", "dmesg")
68 self
.assertEqual(spec
.component_class
.name
, "dmesg")
70 def test_create_source_from_plugin(self
):
71 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
)
72 self
.assertEqual(spec
.component_class
.name
, "dmesg")
74 def test_create_source_from_user(self
):
75 spec
= bt2
.ComponentSpec(_SomeSource
)
76 self
.assertEqual(spec
.component_class
.name
, "_SomeSource")
78 def test_create_filter_from_name(self
):
79 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class("utils", "muxer")
80 self
.assertEqual(spec
.component_class
.name
, "muxer")
82 def test_create_filter_from_object(self
):
83 spec
= bt2
.ComponentSpec(self
._muxer
_cc
)
84 self
.assertEqual(spec
.component_class
.name
, "muxer")
86 def test_create_sink_from_name(self
):
87 with self
.assertRaisesRegex(
89 "source or filter component class `pretty` not found in plugin `text`",
91 bt2
.ComponentSpec
.from_named_plugin_and_component_class("text", "pretty")
93 def test_create_sink_from_object(self
):
94 with self
.assertRaisesRegex(
96 "'_SinkComponentClassConst' is not a source or filter component class",
98 bt2
.ComponentSpec(self
._pretty
_cc
)
100 def test_create_from_object_with_params(self
):
101 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, {"salut": 23})
102 self
.assertEqual(spec
.params
["salut"], 23)
104 def test_create_from_name_with_params(self
):
105 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
106 "text", "dmesg", {"salut": 23}
108 self
.assertEqual(spec
.params
["salut"], 23)
110 def test_create_from_object_with_path_params(self
):
111 spec
= spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, "a path")
112 self
.assertEqual(spec
.params
["inputs"], ["a path"])
114 def test_create_from_name_with_path_params(self
):
115 spec
= spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
116 "text", "dmesg", "a path"
118 self
.assertEqual(spec
.params
["inputs"], ["a path"])
120 def test_create_wrong_comp_class_type(self
):
121 with self
.assertRaisesRegex(
122 TypeError, "'int' is not a source or filter component class"
124 bt2
.ComponentSpec(18)
126 def test_create_from_name_wrong_plugin_name_type(self
):
127 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
128 bt2
.ComponentSpec
.from_named_plugin_and_component_class(23, "compcls")
130 def test_create_from_name_non_existent_plugin(self
):
131 with self
.assertRaisesRegex(
132 ValueError, "no such plugin: this_plugin_does_not_exist"
134 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
135 "this_plugin_does_not_exist", "compcls"
138 def test_create_from_name_wrong_component_class_name_type(self
):
139 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
140 bt2
.ComponentSpec
.from_named_plugin_and_component_class("utils", 190)
142 def test_create_wrong_params_type(self
):
143 with self
.assertRaisesRegex(
144 TypeError, "cannot create value object from 'datetime' object"
146 bt2
.ComponentSpec(self
._dmesg
_cc
, params
=datetime
.datetime
.now())
148 def test_create_from_name_wrong_params_type(self
):
149 with self
.assertRaisesRegex(
150 TypeError, "cannot create value object from 'datetime' object"
152 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
153 "text", "dmesg", datetime
.datetime
.now()
156 def test_create_wrong_log_level_type(self
):
157 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
158 bt2
.ComponentSpec(self
._dmesg
_cc
, logging_level
="banane")
160 def test_create_from_name_wrong_log_level_type(self
):
161 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
162 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
163 "text", "dmesg", logging_level
="banane"
167 # Return a map, msg type -> number of messages of this type.
170 def _count_msgs_by_type(msgs
):
181 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
182 def test_create_wrong_stream_intersection_mode_type(self
):
184 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
185 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
189 with self
.assertRaises(TypeError):
190 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
192 def test_create_wrong_begin_type(self
):
194 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
195 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
199 with self
.assertRaises(TypeError):
200 bt2
.TraceCollectionMessageIterator(specs
, begin
="hi")
202 def test_create_wrong_end_type(self
):
204 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
205 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
209 with self
.assertRaises(TypeError):
210 bt2
.TraceCollectionMessageIterator(specs
, begin
="lel")
212 def test_create_begin_s(self
):
214 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
215 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
218 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
220 def test_create_end_s(self
):
222 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
223 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
226 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
228 def test_create_begin_datetime(self
):
230 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
231 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
234 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
236 def test_create_end_datetime(self
):
238 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
239 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
242 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
244 def test_iter_no_intersection(self
):
246 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
247 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
250 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
251 msgs
= list(msg_iter
)
252 self
.assertEqual(len(msgs
), 28)
253 hist
= _count_msgs_by_type(msgs
)
254 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
256 # Same as the above, but we pass a single spec instead of a spec list.
257 def test_iter_specs_not_list(self
):
258 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
259 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
261 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
262 msgs
= list(msg_iter
)
263 self
.assertEqual(len(msgs
), 28)
264 hist
= _count_msgs_by_type(msgs
)
265 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
267 def test_iter_custom_filter(self
):
268 src_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
269 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
271 flt_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
272 "utils", "trimmer", {"end": "13515309.000000075"}
274 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
275 hist
= _count_msgs_by_type(msg_iter
)
276 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
278 def test_iter_intersection(self
):
280 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
281 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
284 msg_iter
= bt2
.TraceCollectionMessageIterator(
285 specs
, stream_intersection_mode
=True
287 msgs
= list(msg_iter
)
288 self
.assertEqual(len(msgs
), 15)
289 hist
= _count_msgs_by_type(msgs
)
290 self
.assertEqual(hist
[bt2
._EventMessageConst
], 3)
292 def test_iter_intersection_params(self
):
293 # Check that all params used to create the source component are passed
294 # to the `babeltrace.trace-infos` query.
296 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
300 "inputs": [_3EVENTS_INTERSECT_TRACE_PATH
],
301 "clock-class-offset-s": 1000,
306 msg_iter
= bt2
.TraceCollectionMessageIterator(
307 specs
, stream_intersection_mode
=True
310 event_msgs
= [x
for x
in msg_iter
if type(x
) is bt2
._EventMessageConst
]
311 self
.assertEqual(len(event_msgs
), 3)
313 event_msgs
[0].default_clock_snapshot
.ns_from_origin
, 13516309000000071
316 event_msgs
[1].default_clock_snapshot
.ns_from_origin
, 13516309000000072
319 event_msgs
[2].default_clock_snapshot
.ns_from_origin
, 13516309000000082
322 def test_iter_no_intersection_two_traces(self
):
323 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
324 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
327 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
328 msgs
= list(msg_iter
)
329 self
.assertEqual(len(msgs
), 56)
330 hist
= _count_msgs_by_type(msgs
)
331 self
.assertEqual(hist
[bt2
._EventMessageConst
], 16)
333 def test_iter_no_intersection_begin(self
):
335 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
336 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
339 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
340 hist
= _count_msgs_by_type(msg_iter
)
341 self
.assertEqual(hist
[bt2
._EventMessageConst
], 6)
343 def test_iter_no_intersection_end(self
):
345 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
346 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
349 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
350 hist
= _count_msgs_by_type(msg_iter
)
351 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
353 def test_iter_auto_source_component_spec(self
):
354 specs
= [bt2
.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH
)]
355 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
356 msgs
= list(msg_iter
)
357 self
.assertEqual(len(msgs
), 28)
358 hist
= _count_msgs_by_type(msgs
)
359 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
361 def test_iter_auto_source_component_spec_list_of_strings(self
):
362 msg_iter
= bt2
.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH
])
363 msgs
= list(msg_iter
)
364 self
.assertEqual(len(msgs
), 28)
365 hist
= _count_msgs_by_type(msgs
)
366 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
368 def test_iter_auto_source_component_spec_string(self
):
369 msg_iter
= bt2
.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH
)
370 msgs
= list(msg_iter
)
371 self
.assertEqual(len(msgs
), 28)
372 hist
= _count_msgs_by_type(msgs
)
373 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
375 def test_iter_mixed_inputs(self
):
376 msg_iter
= bt2
.TraceCollectionMessageIterator(
378 _3EVENTS_INTERSECT_TRACE_PATH
,
379 bt2
.AutoSourceComponentSpec(_SEQUENCE_TRACE_PATH
),
380 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
381 "ctf", "fs", _NOINTERSECT_TRACE_PATH
385 msgs
= list(msg_iter
)
386 self
.assertEqual(len(msgs
), 76)
387 hist
= _count_msgs_by_type(msgs
)
388 self
.assertEqual(hist
[bt2
._EventMessageConst
], 24)
390 def test_auto_source_component_non_existent(self
):
391 with self
.assertRaisesRegex(
393 "Some auto source component specs did not produce any component",
395 # Test with one path known to contain a trace and one path known
396 # to not contain any trace.
397 bt2
.TraceCollectionMessageIterator(
398 [_SEQUENCE_TRACE_PATH
, "/this/path/better/not/exist"]
402 class _TestAutoDiscoverSourceComponentSpecs(unittest
.TestCase
):
404 self
._saved
_babeltrace
_plugin
_path
= os
.environ
["BABELTRACE_PLUGIN_PATH"]
405 os
.environ
["BABELTRACE_PLUGIN_PATH"] += os
.pathsep
+ self
._plugin
_path
408 os
.environ
["BABELTRACE_PLUGIN_PATH"] = self
._saved
_babeltrace
_plugin
_path
411 @unittest.skipUnless(
412 _BT_ENABLE_PYTHON_PLUGINS
,
413 "Support for Python plugins is disabled",
415 class TestAutoDiscoverSourceComponentSpecsGrouping(
416 _TestAutoDiscoverSourceComponentSpecs
418 _plugin_path
= _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
420 def test_grouping(self
):
422 bt2
.AutoSourceComponentSpec("ABCDE"),
423 bt2
.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH
),
425 it
= bt2
.TraceCollectionMessageIterator(specs
)
426 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
428 self
.assertEqual(len(msgs
), 8)
430 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceABCDE: ABCDE")
431 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceExt: aaa1, aaa2, aaa3")
432 self
.assertEqual(msgs
[2].stream
.name
, "TestSourceExt: bbb1, bbb2")
433 self
.assertEqual(msgs
[3].stream
.name
, "TestSourceExt: ccc1")
434 self
.assertEqual(msgs
[4].stream
.name
, "TestSourceExt: ccc2")
435 self
.assertEqual(msgs
[5].stream
.name
, "TestSourceExt: ccc3")
436 self
.assertEqual(msgs
[6].stream
.name
, "TestSourceExt: ccc4")
437 self
.assertEqual(msgs
[7].stream
.name
, "TestSourceSomeDir: some-dir")
440 @unittest.skipUnless(
441 _BT_ENABLE_PYTHON_PLUGINS
,
442 "Support for Python plugins is disabled",
444 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
445 _TestAutoDiscoverSourceComponentSpecs
447 _plugin_path
= _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
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")
453 def _test_two_comps_from_one_spec(self
, params
, obj
=None, logging_level
=None):
455 bt2
.AutoSourceComponentSpec(
456 self
._dir
_ab
, params
=params
, obj
=obj
, logging_level
=logging_level
459 it
= bt2
.TraceCollectionMessageIterator(specs
)
460 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
462 self
.assertEqual(len(msgs
), 2)
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"}
471 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
472 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: ('test-allo', 'madame')")
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"
479 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
480 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
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
488 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
491 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
494 def _test_two_comps_from_two_specs(
500 logging_level_a
=None,
501 logging_level_b
=None,
504 bt2
.AutoSourceComponentSpec(
505 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
507 bt2
.AutoSourceComponentSpec(
508 self
._dir
_b
, params
=params_b
, obj
=obj_b
, logging_level
=logging_level_b
511 it
= bt2
.TraceCollectionMessageIterator(specs
)
512 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
514 self
.assertEqual(len(msgs
), 2)
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"},
524 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
526 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
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"},
537 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
538 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
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
,
549 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
552 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
555 def _test_one_comp_from_one_spec_one_comp_from_both_1(
561 logging_level_a
=None,
562 logging_level_ab
=None,
565 bt2
.AutoSourceComponentSpec(
566 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
568 bt2
.AutoSourceComponentSpec(
572 logging_level
=logging_level_ab
,
575 it
= bt2
.TraceCollectionMessageIterator(specs
)
576 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
578 self
.assertEqual(len(msgs
), 2)
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"},
590 "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
593 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
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"},
604 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
605 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
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
,
616 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
619 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
622 def _test_one_comp_from_one_spec_one_comp_from_both_2(
628 logging_level_ab
=None,
629 logging_level_a
=None,
632 bt2
.AutoSourceComponentSpec(
636 logging_level
=logging_level_ab
,
638 bt2
.AutoSourceComponentSpec(
639 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
642 it
= bt2
.TraceCollectionMessageIterator(specs
)
643 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
645 self
.assertEqual(len(msgs
), 2)
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(
652 "test-bonjour": "madame",
653 "test-salut": "les amis",
654 "what": "test-params",
656 params_a
={"test-bonjour": "monsieur", "what": "test-params"},
661 "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
665 "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
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"},
676 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
677 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
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
,
688 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
691 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
694 def test_obj_override_with_none(self
):
696 bt2
.AutoSourceComponentSpec(
697 self
._dir
_ab
, params
={"what": "python-obj"}, obj
="deore"
699 bt2
.AutoSourceComponentSpec(
700 self
._dir
_a
, params
={"what": "python-obj"}, obj
=None
703 it
= bt2
.TraceCollectionMessageIterator(specs
)
704 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
706 self
.assertEqual(len(msgs
), 2)
707 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: None")
708 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
710 def test_obj_no_override_with_no_obj(self
):
712 bt2
.AutoSourceComponentSpec(
713 self
._dir
_ab
, params
={"what": "python-obj"}, obj
="deore"
715 bt2
.AutoSourceComponentSpec(self
._dir
_a
, params
={"what": "python-obj"}),
717 it
= bt2
.TraceCollectionMessageIterator(specs
)
718 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
720 self
.assertEqual(len(msgs
), 2)
721 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
722 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
725 class TestAutoDiscoverFailures(unittest
.TestCase
):
726 def test_metadata_syntax_error(self
):
727 with self
.assertRaisesRegex(
729 'At line 3 in metadata stream: syntax error, unexpected CTF_RSBRAC: token="]"',
731 specs
= [bt2
.AutoSourceComponentSpec(_METADATA_SYNTAX_ERROR_TRACE_PATH
)]
732 bt2
.TraceCollectionMessageIterator(specs
)
735 if __name__
== "__main__":