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"
29 _METADATA_SYNTAX_ERROR_TRACE_PATH
= os
.path
.join(
30 _BT_CTF_TRACES_PATH
, "fail", "metadata-syntax-error"
35 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
41 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
46 class _SomeSink(bt2
._UserSinkComponent
):
47 def _user_consume(self
):
51 class ComponentSpecTestCase(unittest
.TestCase
):
53 # A source CC from a plugin.
54 self
._dmesg
_cc
= bt2
.find_plugin("text").source_component_classes
["dmesg"]
55 assert self
._dmesg
_cc
is not None
57 # A filter CC from a plugin.
58 self
._muxer
_cc
= bt2
.find_plugin("utils").filter_component_classes
["muxer"]
59 assert self
._muxer
_cc
is not None
61 # A sink CC from a plugin.
62 self
._pretty
_cc
= bt2
.find_plugin("text").sink_component_classes
["pretty"]
63 assert self
._pretty
_cc
is not None
65 def test_create_source_from_name(self
):
66 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class("text", "dmesg")
67 self
.assertEqual(spec
.component_class
.name
, "dmesg")
69 def test_create_source_from_plugin(self
):
70 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
)
71 self
.assertEqual(spec
.component_class
.name
, "dmesg")
73 def test_create_source_from_user(self
):
74 spec
= bt2
.ComponentSpec(_SomeSource
)
75 self
.assertEqual(spec
.component_class
.name
, "_SomeSource")
77 def test_create_filter_from_name(self
):
78 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class("utils", "muxer")
79 self
.assertEqual(spec
.component_class
.name
, "muxer")
81 def test_create_filter_from_object(self
):
82 spec
= bt2
.ComponentSpec(self
._muxer
_cc
)
83 self
.assertEqual(spec
.component_class
.name
, "muxer")
85 def test_create_sink_from_name(self
):
86 with self
.assertRaisesRegex(
88 "source or filter component class `pretty` not found in plugin `text`",
90 bt2
.ComponentSpec
.from_named_plugin_and_component_class("text", "pretty")
92 def test_create_sink_from_object(self
):
93 with self
.assertRaisesRegex(
95 "'_SinkComponentClassConst' is not a source or filter component class",
97 bt2
.ComponentSpec(self
._pretty
_cc
)
99 def test_create_from_object_with_params(self
):
100 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, {"salut": 23})
101 self
.assertEqual(spec
.params
["salut"], 23)
103 def test_create_from_name_with_params(self
):
104 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
105 "text", "dmesg", {"salut": 23}
107 self
.assertEqual(spec
.params
["salut"], 23)
109 def test_create_from_object_with_path_params(self
):
110 spec
= spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, "a path")
111 self
.assertEqual(spec
.params
["inputs"], ["a path"])
113 def test_create_from_name_with_path_params(self
):
114 spec
= spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
115 "text", "dmesg", "a path"
117 self
.assertEqual(spec
.params
["inputs"], ["a path"])
119 def test_create_wrong_comp_class_type(self
):
120 with self
.assertRaisesRegex(
121 TypeError, "'int' is not a source or filter component class"
123 bt2
.ComponentSpec(18)
125 def test_create_from_name_wrong_plugin_name_type(self
):
126 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
127 bt2
.ComponentSpec
.from_named_plugin_and_component_class(23, "compcls")
129 def test_create_from_name_non_existent_plugin(self
):
130 with self
.assertRaisesRegex(
131 ValueError, "no such plugin: this_plugin_does_not_exist"
133 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
134 "this_plugin_does_not_exist", "compcls"
137 def test_create_from_name_wrong_component_class_name_type(self
):
138 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
139 bt2
.ComponentSpec
.from_named_plugin_and_component_class("utils", 190)
141 def test_create_wrong_params_type(self
):
142 with self
.assertRaisesRegex(
143 TypeError, "cannot create value object from 'datetime' object"
145 bt2
.ComponentSpec(self
._dmesg
_cc
, params
=datetime
.datetime
.now())
147 def test_create_from_name_wrong_params_type(self
):
148 with self
.assertRaisesRegex(
149 TypeError, "cannot create value object from 'datetime' object"
151 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
152 "text", "dmesg", datetime
.datetime
.now()
155 def test_create_wrong_log_level_type(self
):
156 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
157 bt2
.ComponentSpec(self
._dmesg
_cc
, logging_level
="banane")
159 def test_create_from_name_wrong_log_level_type(self
):
160 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
161 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
162 "text", "dmesg", logging_level
="banane"
166 # Return a map, msg type -> number of messages of this type.
169 def _count_msgs_by_type(msgs
):
180 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
181 def test_create_wrong_stream_intersection_mode_type(self
):
183 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
184 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
188 with self
.assertRaises(TypeError):
189 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
191 def test_create_wrong_begin_type(self
):
193 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
194 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
198 with self
.assertRaises(TypeError):
199 bt2
.TraceCollectionMessageIterator(specs
, begin
="hi")
201 def test_create_wrong_end_type(self
):
203 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
204 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
208 with self
.assertRaises(TypeError):
209 bt2
.TraceCollectionMessageIterator(specs
, begin
="lel")
211 def test_create_begin_s(self
):
213 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
214 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
217 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
219 def test_create_end_s(self
):
221 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
222 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
225 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
227 def test_create_begin_datetime(self
):
229 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
230 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
233 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
235 def test_create_end_datetime(self
):
237 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
238 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
241 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
243 def test_iter_no_intersection(self
):
245 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
246 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
249 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
250 msgs
= list(msg_iter
)
251 self
.assertEqual(len(msgs
), 28)
252 hist
= _count_msgs_by_type(msgs
)
253 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
255 # Same as the above, but we pass a single spec instead of a spec list.
256 def test_iter_specs_not_list(self
):
257 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
258 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
260 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
261 msgs
= list(msg_iter
)
262 self
.assertEqual(len(msgs
), 28)
263 hist
= _count_msgs_by_type(msgs
)
264 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
266 def test_iter_custom_filter(self
):
267 src_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
268 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
270 flt_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
271 "utils", "trimmer", {"end": "13515309.000000075"}
273 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
274 hist
= _count_msgs_by_type(msg_iter
)
275 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
277 def test_iter_intersection(self
):
279 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
280 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
283 msg_iter
= bt2
.TraceCollectionMessageIterator(
284 specs
, stream_intersection_mode
=True
286 msgs
= list(msg_iter
)
287 self
.assertEqual(len(msgs
), 15)
288 hist
= _count_msgs_by_type(msgs
)
289 self
.assertEqual(hist
[bt2
._EventMessageConst
], 3)
291 def test_iter_intersection_params(self
):
292 # Check that all params used to create the source component are passed
293 # to the `babeltrace.trace-infos` query.
295 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
299 "inputs": [_3EVENTS_INTERSECT_TRACE_PATH
],
300 "clock-class-offset-s": 1000,
305 msg_iter
= bt2
.TraceCollectionMessageIterator(
306 specs
, stream_intersection_mode
=True
309 event_msgs
= [x
for x
in msg_iter
if type(x
) is bt2
._EventMessageConst
]
310 self
.assertEqual(len(event_msgs
), 3)
312 event_msgs
[0].default_clock_snapshot
.ns_from_origin
, 13516309000000071
315 event_msgs
[1].default_clock_snapshot
.ns_from_origin
, 13516309000000072
318 event_msgs
[2].default_clock_snapshot
.ns_from_origin
, 13516309000000082
321 def test_iter_no_intersection_two_traces(self
):
322 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
323 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
326 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
327 msgs
= list(msg_iter
)
328 self
.assertEqual(len(msgs
), 56)
329 hist
= _count_msgs_by_type(msgs
)
330 self
.assertEqual(hist
[bt2
._EventMessageConst
], 16)
332 def test_iter_no_intersection_begin(self
):
334 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
335 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
338 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
339 hist
= _count_msgs_by_type(msg_iter
)
340 self
.assertEqual(hist
[bt2
._EventMessageConst
], 6)
342 def test_iter_no_intersection_end(self
):
344 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
345 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
348 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
349 hist
= _count_msgs_by_type(msg_iter
)
350 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
352 def test_iter_auto_source_component_spec(self
):
353 specs
= [bt2
.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH
)]
354 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
355 msgs
= list(msg_iter
)
356 self
.assertEqual(len(msgs
), 28)
357 hist
= _count_msgs_by_type(msgs
)
358 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
360 def test_iter_auto_source_component_spec_list_of_strings(self
):
361 msg_iter
= bt2
.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH
])
362 msgs
= list(msg_iter
)
363 self
.assertEqual(len(msgs
), 28)
364 hist
= _count_msgs_by_type(msgs
)
365 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
367 def test_iter_auto_source_component_spec_string(self
):
368 msg_iter
= bt2
.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH
)
369 msgs
= list(msg_iter
)
370 self
.assertEqual(len(msgs
), 28)
371 hist
= _count_msgs_by_type(msgs
)
372 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
374 def test_iter_mixed_inputs(self
):
375 msg_iter
= bt2
.TraceCollectionMessageIterator(
377 _3EVENTS_INTERSECT_TRACE_PATH
,
378 bt2
.AutoSourceComponentSpec(_SEQUENCE_TRACE_PATH
),
379 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
380 "ctf", "fs", _NOINTERSECT_TRACE_PATH
384 msgs
= list(msg_iter
)
385 self
.assertEqual(len(msgs
), 76)
386 hist
= _count_msgs_by_type(msgs
)
387 self
.assertEqual(hist
[bt2
._EventMessageConst
], 24)
389 def test_auto_source_component_non_existent(self
):
390 with self
.assertRaisesRegex(
392 "Some auto source component specs did not produce any component",
394 # Test with one path known to contain a trace and one path known
395 # to not contain any trace.
396 bt2
.TraceCollectionMessageIterator(
397 [_SEQUENCE_TRACE_PATH
, "/this/path/better/not/exist"]
401 class _TestAutoDiscoverSourceComponentSpecs(unittest
.TestCase
):
403 self
._saved
_babeltrace
_plugin
_path
= os
.environ
["BABELTRACE_PLUGIN_PATH"]
404 os
.environ
["BABELTRACE_PLUGIN_PATH"] += os
.pathsep
+ self
._plugin
_path
407 os
.environ
["BABELTRACE_PLUGIN_PATH"] = self
._saved
_babeltrace
_plugin
_path
410 class TestAutoDiscoverSourceComponentSpecsGrouping(
411 _TestAutoDiscoverSourceComponentSpecs
413 _plugin_path
= _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
415 def test_grouping(self
):
417 bt2
.AutoSourceComponentSpec("ABCDE"),
418 bt2
.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH
),
420 it
= bt2
.TraceCollectionMessageIterator(specs
)
421 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
423 self
.assertEqual(len(msgs
), 8)
425 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceABCDE: ABCDE")
426 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceExt: aaa1, aaa2, aaa3")
427 self
.assertEqual(msgs
[2].stream
.name
, "TestSourceExt: bbb1, bbb2")
428 self
.assertEqual(msgs
[3].stream
.name
, "TestSourceExt: ccc1")
429 self
.assertEqual(msgs
[4].stream
.name
, "TestSourceExt: ccc2")
430 self
.assertEqual(msgs
[5].stream
.name
, "TestSourceExt: ccc3")
431 self
.assertEqual(msgs
[6].stream
.name
, "TestSourceExt: ccc4")
432 self
.assertEqual(msgs
[7].stream
.name
, "TestSourceSomeDir: some-dir")
435 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
436 _TestAutoDiscoverSourceComponentSpecs
438 _plugin_path
= _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
440 _dir_a
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, "dir-a")
441 _dir_b
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, "dir-b")
442 _dir_ab
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, "dir-ab")
444 def _test_two_comps_from_one_spec(self
, params
, obj
=None, logging_level
=None):
446 bt2
.AutoSourceComponentSpec(
447 self
._dir
_ab
, params
=params
, obj
=obj
, logging_level
=logging_level
450 it
= bt2
.TraceCollectionMessageIterator(specs
)
451 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
453 self
.assertEqual(len(msgs
), 2)
457 def test_params_two_comps_from_one_spec(self
):
458 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
459 params
={"test-allo": "madame", "what": "test-params"}
462 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
463 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: ('test-allo', 'madame')")
465 def test_obj_two_comps_from_one_spec(self
):
466 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
467 params
={"what": "python-obj"}, obj
="deore"
470 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
471 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
473 def test_log_level_two_comps_from_one_spec(self
):
474 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
475 params
={"what": "log-level"}, logging_level
=bt2
.LoggingLevel
.DEBUG
479 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
482 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
485 def _test_two_comps_from_two_specs(
491 logging_level_a
=None,
492 logging_level_b
=None,
495 bt2
.AutoSourceComponentSpec(
496 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
498 bt2
.AutoSourceComponentSpec(
499 self
._dir
_b
, params
=params_b
, obj
=obj_b
, logging_level
=logging_level_b
502 it
= bt2
.TraceCollectionMessageIterator(specs
)
503 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
505 self
.assertEqual(len(msgs
), 2)
509 def test_params_two_comps_from_two_specs(self
):
510 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
511 params_a
={"test-allo": "madame", "what": "test-params"},
512 params_b
={"test-bonjour": "monsieur", "what": "test-params"},
515 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
517 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
520 def test_obj_two_comps_from_two_specs(self
):
521 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
522 params_a
={"what": "python-obj"},
523 params_b
={"what": "python-obj"},
528 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
529 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
531 def test_log_level_two_comps_from_two_specs(self
):
532 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
533 params_a
={"what": "log-level"},
534 params_b
={"what": "log-level"},
535 logging_level_a
=bt2
.LoggingLevel
.DEBUG
,
536 logging_level_b
=bt2
.LoggingLevel
.TRACE
,
540 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
543 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
546 def _test_one_comp_from_one_spec_one_comp_from_both_1(
552 logging_level_a
=None,
553 logging_level_ab
=None,
556 bt2
.AutoSourceComponentSpec(
557 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
559 bt2
.AutoSourceComponentSpec(
563 logging_level
=logging_level_ab
,
566 it
= bt2
.TraceCollectionMessageIterator(specs
)
567 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
569 self
.assertEqual(len(msgs
), 2)
573 def test_params_one_comp_from_one_spec_one_comp_from_both_1(self
):
574 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
575 params_a
={"test-allo": "madame", "what": "test-params"},
576 params_ab
={"test-bonjour": "monsieur", "what": "test-params"},
581 "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
584 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
587 def test_obj_one_comp_from_one_spec_one_comp_from_both_1(self
):
588 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
589 params_a
={"what": "python-obj"},
590 params_ab
={"what": "python-obj"},
595 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
596 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
598 def test_log_level_one_comp_from_one_spec_one_comp_from_both_1(self
):
599 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
600 params_a
={"what": "log-level"},
601 params_ab
={"what": "log-level"},
602 logging_level_a
=bt2
.LoggingLevel
.DEBUG
,
603 logging_level_ab
=bt2
.LoggingLevel
.TRACE
,
607 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
610 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
613 def _test_one_comp_from_one_spec_one_comp_from_both_2(
619 logging_level_ab
=None,
620 logging_level_a
=None,
623 bt2
.AutoSourceComponentSpec(
627 logging_level
=logging_level_ab
,
629 bt2
.AutoSourceComponentSpec(
630 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
633 it
= bt2
.TraceCollectionMessageIterator(specs
)
634 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
636 self
.assertEqual(len(msgs
), 2)
640 def test_params_one_comp_from_one_spec_one_comp_from_both_2(self
):
641 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
643 "test-bonjour": "madame",
644 "test-salut": "les amis",
645 "what": "test-params",
647 params_a
={"test-bonjour": "monsieur", "what": "test-params"},
652 "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
656 "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
659 def test_obj_one_comp_from_one_spec_one_comp_from_both_2(self
):
660 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
661 params_ab
={"what": "python-obj"},
662 params_a
={"what": "python-obj"},
667 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
668 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
670 def test_log_level_one_comp_from_one_spec_one_comp_from_both_2(self
):
671 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
672 params_ab
={"what": "log-level"},
673 params_a
={"what": "log-level"},
674 logging_level_ab
=bt2
.LoggingLevel
.DEBUG
,
675 logging_level_a
=bt2
.LoggingLevel
.TRACE
,
679 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
682 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
685 def test_obj_override_with_none(self
):
687 bt2
.AutoSourceComponentSpec(
688 self
._dir
_ab
, params
={"what": "python-obj"}, obj
="deore"
690 bt2
.AutoSourceComponentSpec(
691 self
._dir
_a
, params
={"what": "python-obj"}, obj
=None
694 it
= bt2
.TraceCollectionMessageIterator(specs
)
695 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
697 self
.assertEqual(len(msgs
), 2)
698 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: None")
699 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
701 def test_obj_no_override_with_no_obj(self
):
703 bt2
.AutoSourceComponentSpec(
704 self
._dir
_ab
, params
={"what": "python-obj"}, obj
="deore"
706 bt2
.AutoSourceComponentSpec(self
._dir
_a
, params
={"what": "python-obj"}),
708 it
= bt2
.TraceCollectionMessageIterator(specs
)
709 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
711 self
.assertEqual(len(msgs
), 2)
712 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
713 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
716 class TestAutoDiscoverFailures(unittest
.TestCase
):
717 def test_metadata_syntax_error(self
):
718 with self
.assertRaisesRegex(
720 'At line 3 in metadata stream: syntax error, unexpected CTF_RSBRAC: token="]"',
722 specs
= [bt2
.AutoSourceComponentSpec(_METADATA_SYNTAX_ERROR_TRACE_PATH
)]
723 bt2
.TraceCollectionMessageIterator(specs
)
726 if __name__
== "__main__":