2 # Copyright (C) 2019 EfficiOS Inc.
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
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.
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.
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'
31 _NOINTERSECT_TRACE_PATH
= os
.path
.join(
32 _BT_CTF_TRACES_PATH
, 'intersection', 'nointersect'
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'
38 _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
= os
.path
.join(
39 _BT_TESTS_DATADIR
, 'auto-source-discovery', 'params-log-level'
44 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
50 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
55 class _SomeSink(bt2
._UserSinkComponent
):
56 def _user_consume(self
):
60 class ComponentSpecTestCase(unittest
.TestCase
):
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
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
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
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')
78 def test_create_source_from_plugin(self
):
79 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
)
80 self
.assertEqual(spec
.component_class
.name
, 'dmesg')
82 def test_create_source_from_user(self
):
83 spec
= bt2
.ComponentSpec(_SomeSource
)
84 self
.assertEqual(spec
.component_class
.name
, '_SomeSource')
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')
90 def test_create_filter_from_object(self
):
91 spec
= bt2
.ComponentSpec(self
._muxer
_cc
)
92 self
.assertEqual(spec
.component_class
.name
, 'muxer')
94 def test_create_sink_from_name(self
):
95 with self
.assertRaisesRegex(
97 'source or filter component class `pretty` not found in plugin `text`',
99 bt2
.ComponentSpec
.from_named_plugin_and_component_class('text', 'pretty')
101 def test_create_sink_from_object(self
):
102 with self
.assertRaisesRegex(
103 TypeError, "'_SinkComponentClass' is not a source or filter component class"
105 bt2
.ComponentSpec(self
._pretty
_cc
)
107 def test_create_from_object_with_params(self
):
108 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, {'salut': 23})
109 self
.assertEqual(spec
.params
['salut'], 23)
111 def test_create_from_name_with_params(self
):
112 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
113 'text', 'dmesg', {'salut': 23}
115 self
.assertEqual(spec
.params
['salut'], 23)
117 def test_create_from_object_with_path_params(self
):
118 spec
= spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, 'a path')
119 self
.assertEqual(spec
.params
['inputs'], ['a path'])
121 def test_create_from_name_with_path_params(self
):
122 spec
= spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
123 'text', 'dmesg', 'a path'
125 self
.assertEqual(spec
.params
['inputs'], ['a path'])
127 def test_create_wrong_comp_class_type(self
):
128 with self
.assertRaisesRegex(
129 TypeError, "'int' is not a source or filter component class"
131 bt2
.ComponentSpec(18)
133 def test_create_from_name_wrong_plugin_name_type(self
):
134 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
135 bt2
.ComponentSpec
.from_named_plugin_and_component_class(23, 'compcls')
137 def test_create_from_name_non_existent_plugin(self
):
138 with self
.assertRaisesRegex(
139 ValueError, "no such plugin: this_plugin_does_not_exist"
141 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
142 'this_plugin_does_not_exist', 'compcls'
145 def test_create_from_name_wrong_component_class_name_type(self
):
146 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
147 bt2
.ComponentSpec
.from_named_plugin_and_component_class('utils', 190)
149 def test_create_wrong_params_type(self
):
150 with self
.assertRaisesRegex(
151 TypeError, "cannot create value object from 'datetime' object"
153 bt2
.ComponentSpec(self
._dmesg
_cc
, params
=datetime
.datetime
.now())
155 def test_create_from_name_wrong_params_type(self
):
156 with self
.assertRaisesRegex(
157 TypeError, "cannot create value object from 'datetime' object"
159 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
160 'text', 'dmesg', datetime
.datetime
.now()
163 def test_create_wrong_log_level_type(self
):
164 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
165 bt2
.ComponentSpec(self
._dmesg
_cc
, logging_level
='banane')
167 def test_create_from_name_wrong_log_level_type(self
):
168 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
169 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
170 'text', 'dmesg', logging_level
='banane'
174 # Return a map, msg type -> number of messages of this type.
177 def _count_msgs_by_type(msgs
):
188 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
189 def test_create_wrong_stream_intersection_mode_type(self
):
191 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
192 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
196 with self
.assertRaises(TypeError):
197 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
199 def test_create_wrong_begin_type(self
):
201 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
202 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
206 with self
.assertRaises(TypeError):
207 bt2
.TraceCollectionMessageIterator(specs
, begin
='hi')
209 def test_create_wrong_end_type(self
):
211 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
212 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
216 with self
.assertRaises(TypeError):
217 bt2
.TraceCollectionMessageIterator(specs
, begin
='lel')
219 def test_create_begin_s(self
):
221 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
222 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
225 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
227 def test_create_end_s(self
):
229 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
230 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
233 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
235 def test_create_begin_datetime(self
):
237 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
238 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
241 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
243 def test_create_end_datetime(self
):
245 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
246 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
249 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
251 def test_iter_no_intersection(self
):
253 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
254 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
257 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
258 msgs
= list(msg_iter
)
259 self
.assertEqual(len(msgs
), 28)
260 hist
= _count_msgs_by_type(msgs
)
261 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
263 # Same as the above, but we pass a single spec instead of a spec list.
264 def test_iter_specs_not_list(self
):
265 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
266 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
268 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
269 msgs
= list(msg_iter
)
270 self
.assertEqual(len(msgs
), 28)
271 hist
= _count_msgs_by_type(msgs
)
272 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
274 def test_iter_custom_filter(self
):
275 src_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
276 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
278 flt_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
279 'utils', 'trimmer', {'end': '13515309.000000075'}
281 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
282 hist
= _count_msgs_by_type(msg_iter
)
283 self
.assertEqual(hist
[bt2
._EventMessage
], 5)
285 def test_iter_intersection(self
):
287 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
288 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
291 msg_iter
= bt2
.TraceCollectionMessageIterator(
292 specs
, stream_intersection_mode
=True
294 msgs
= list(msg_iter
)
295 self
.assertEqual(len(msgs
), 15)
296 hist
= _count_msgs_by_type(msgs
)
297 self
.assertEqual(hist
[bt2
._EventMessage
], 3)
299 def test_iter_intersection_no_inputs_param(self
):
301 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
302 'text', 'dmesg', {'read-from-stdin': True}
306 with self
.assertRaises(ValueError):
307 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=True)
309 def test_iter_no_intersection_two_traces(self
):
310 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
311 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
314 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
315 msgs
= list(msg_iter
)
316 self
.assertEqual(len(msgs
), 56)
317 hist
= _count_msgs_by_type(msgs
)
318 self
.assertEqual(hist
[bt2
._EventMessage
], 16)
320 def test_iter_no_intersection_begin(self
):
322 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
323 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
326 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
327 hist
= _count_msgs_by_type(msg_iter
)
328 self
.assertEqual(hist
[bt2
._EventMessage
], 6)
330 def test_iter_no_intersection_end(self
):
332 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
333 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
336 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
337 hist
= _count_msgs_by_type(msg_iter
)
338 self
.assertEqual(hist
[bt2
._EventMessage
], 5)
340 def test_iter_auto_source_component_spec(self
):
341 specs
= [bt2
.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH
)]
342 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
343 msgs
= list(msg_iter
)
344 self
.assertEqual(len(msgs
), 28)
345 hist
= _count_msgs_by_type(msgs
)
346 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
348 def test_iter_auto_source_component_spec_list_of_strings(self
):
349 msg_iter
= bt2
.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH
])
350 msgs
= list(msg_iter
)
351 self
.assertEqual(len(msgs
), 28)
352 hist
= _count_msgs_by_type(msgs
)
353 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
355 def test_iter_auto_source_component_spec_string(self
):
356 msg_iter
= bt2
.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH
)
357 msgs
= list(msg_iter
)
358 self
.assertEqual(len(msgs
), 28)
359 hist
= _count_msgs_by_type(msgs
)
360 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
362 def test_iter_mixed_inputs(self
):
363 msg_iter
= bt2
.TraceCollectionMessageIterator(
365 _3EVENTS_INTERSECT_TRACE_PATH
,
366 bt2
.AutoSourceComponentSpec(_SEQUENCE_TRACE_PATH
),
367 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
368 'ctf', 'fs', _NOINTERSECT_TRACE_PATH
372 msgs
= list(msg_iter
)
373 self
.assertEqual(len(msgs
), 76)
374 hist
= _count_msgs_by_type(msgs
)
375 self
.assertEqual(hist
[bt2
._EventMessage
], 24)
377 def test_auto_source_component_non_existent(self
):
378 with self
.assertRaisesRegex(
380 'Some auto source component specs did not produce any component',
382 # Test with one path known to contain a trace and one path known
383 # to not contain any trace.
384 bt2
.TraceCollectionMessageIterator(
385 [_SEQUENCE_TRACE_PATH
, '/this/path/better/not/exist']
389 class _TestAutoDiscoverSourceComponentSpecs(unittest
.TestCase
):
391 self
._saved
_babeltrace
_plugin
_path
= os
.environ
['BABELTRACE_PLUGIN_PATH']
392 os
.environ
['BABELTRACE_PLUGIN_PATH'] += os
.pathsep
+ self
._plugin
_path
395 os
.environ
['BABELTRACE_PLUGIN_PATH'] = self
._saved
_babeltrace
_plugin
_path
398 class TestAutoDiscoverSourceComponentSpecsGrouping(
399 _TestAutoDiscoverSourceComponentSpecs
401 _plugin_path
= _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
403 def test_grouping(self
):
405 bt2
.AutoSourceComponentSpec('ABCDE'),
406 bt2
.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH
),
408 it
= bt2
.TraceCollectionMessageIterator(specs
)
409 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
411 self
.assertEqual(len(msgs
), 8)
413 self
.assertEqual(msgs
[0].stream
.name
, 'TestSourceABCDE: ABCDE')
414 self
.assertEqual(msgs
[1].stream
.name
, 'TestSourceExt: aaa1, aaa2, aaa3')
415 self
.assertEqual(msgs
[2].stream
.name
, 'TestSourceExt: bbb1, bbb2')
416 self
.assertEqual(msgs
[3].stream
.name
, 'TestSourceExt: ccc1')
417 self
.assertEqual(msgs
[4].stream
.name
, 'TestSourceExt: ccc2')
418 self
.assertEqual(msgs
[5].stream
.name
, 'TestSourceExt: ccc3')
419 self
.assertEqual(msgs
[6].stream
.name
, 'TestSourceExt: ccc4')
420 self
.assertEqual(msgs
[7].stream
.name
, 'TestSourceSomeDir: some-dir')
423 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
424 _TestAutoDiscoverSourceComponentSpecs
426 _plugin_path
= _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
428 _dir_a
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, 'dir-a')
429 _dir_b
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, 'dir-b')
430 _dir_ab
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, 'dir-ab')
432 def _test_two_comps_from_one_spec(self
, params
, obj
=None, logging_level
=None):
434 bt2
.AutoSourceComponentSpec(
435 self
._dir
_ab
, params
=params
, obj
=obj
, logging_level
=logging_level
438 it
= bt2
.TraceCollectionMessageIterator(specs
)
439 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
441 self
.assertEqual(len(msgs
), 2)
445 def test_params_two_comps_from_one_spec(self
):
446 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
447 params
={'test-allo': 'madame', 'what': 'test-params'}
450 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
451 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: ('test-allo', 'madame')")
453 def test_obj_two_comps_from_one_spec(self
):
454 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
455 params
={'what': 'python-obj'}, obj
='deore'
458 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
459 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
461 def test_log_level_two_comps_from_one_spec(self
):
462 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
463 params
={'what': 'log-level'}, logging_level
=bt2
.LoggingLevel
.DEBUG
467 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
470 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
473 def _test_two_comps_from_two_specs(
479 logging_level_a
=None,
480 logging_level_b
=None,
483 bt2
.AutoSourceComponentSpec(
484 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
486 bt2
.AutoSourceComponentSpec(
487 self
._dir
_b
, params
=params_b
, obj
=obj_b
, logging_level
=logging_level_b
490 it
= bt2
.TraceCollectionMessageIterator(specs
)
491 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
493 self
.assertEqual(len(msgs
), 2)
497 def test_params_two_comps_from_two_specs(self
):
498 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
499 params_a
={'test-allo': 'madame', 'what': 'test-params'},
500 params_b
={'test-bonjour': 'monsieur', 'what': 'test-params'},
503 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
505 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
508 def test_obj_two_comps_from_two_specs(self
):
509 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
510 params_a
={'what': 'python-obj'},
511 params_b
={'what': 'python-obj'},
516 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
517 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
519 def test_log_level_two_comps_from_two_specs(self
):
520 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
521 params_a
={'what': 'log-level'},
522 params_b
={'what': 'log-level'},
523 logging_level_a
=bt2
.LoggingLevel
.DEBUG
,
524 logging_level_b
=bt2
.LoggingLevel
.TRACE
,
528 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
531 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
534 def _test_one_comp_from_one_spec_one_comp_from_both_1(
540 logging_level_a
=None,
541 logging_level_ab
=None,
544 bt2
.AutoSourceComponentSpec(
545 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
547 bt2
.AutoSourceComponentSpec(
551 logging_level
=logging_level_ab
,
554 it
= bt2
.TraceCollectionMessageIterator(specs
)
555 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
557 self
.assertEqual(len(msgs
), 2)
561 def test_params_one_comp_from_one_spec_one_comp_from_both_1(self
):
562 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
563 params_a
={'test-allo': 'madame', 'what': 'test-params'},
564 params_ab
={'test-bonjour': 'monsieur', 'what': 'test-params'},
569 "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
572 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
575 def test_obj_one_comp_from_one_spec_one_comp_from_both_1(self
):
576 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
577 params_a
={'what': 'python-obj'},
578 params_ab
={'what': 'python-obj'},
583 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
584 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
586 def test_log_level_one_comp_from_one_spec_one_comp_from_both_1(self
):
587 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
588 params_a
={'what': 'log-level'},
589 params_ab
={'what': 'log-level'},
590 logging_level_a
=bt2
.LoggingLevel
.DEBUG
,
591 logging_level_ab
=bt2
.LoggingLevel
.TRACE
,
595 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
598 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
601 def _test_one_comp_from_one_spec_one_comp_from_both_2(
607 logging_level_ab
=None,
608 logging_level_a
=None,
611 bt2
.AutoSourceComponentSpec(
615 logging_level
=logging_level_ab
,
617 bt2
.AutoSourceComponentSpec(
618 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
621 it
= bt2
.TraceCollectionMessageIterator(specs
)
622 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
624 self
.assertEqual(len(msgs
), 2)
628 def test_params_one_comp_from_one_spec_one_comp_from_both_2(self
):
629 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
631 'test-bonjour': 'madame',
632 'test-salut': 'les amis',
633 'what': 'test-params',
635 params_a
={'test-bonjour': 'monsieur', 'what': 'test-params'},
640 "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
644 "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
647 def test_obj_one_comp_from_one_spec_one_comp_from_both_2(self
):
648 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
649 params_ab
={'what': 'python-obj'},
650 params_a
={'what': 'python-obj'},
655 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
656 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
658 def test_log_level_one_comp_from_one_spec_one_comp_from_both_2(self
):
659 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
660 params_ab
={'what': 'log-level'},
661 params_a
={'what': 'log-level'},
662 logging_level_ab
=bt2
.LoggingLevel
.DEBUG
,
663 logging_level_a
=bt2
.LoggingLevel
.TRACE
,
667 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
670 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
673 def test_obj_override_with_none(self
):
675 bt2
.AutoSourceComponentSpec(
676 self
._dir
_ab
, params
={'what': 'python-obj'}, obj
='deore'
678 bt2
.AutoSourceComponentSpec(
679 self
._dir
_a
, params
={'what': 'python-obj'}, obj
=None
682 it
= bt2
.TraceCollectionMessageIterator(specs
)
683 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
685 self
.assertEqual(len(msgs
), 2)
686 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: None")
687 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
689 def test_obj_no_override_with_no_obj(self
):
691 bt2
.AutoSourceComponentSpec(
692 self
._dir
_ab
, params
={'what': 'python-obj'}, obj
='deore'
694 bt2
.AutoSourceComponentSpec(self
._dir
_a
, params
={'what': 'python-obj'}),
696 it
= bt2
.TraceCollectionMessageIterator(specs
)
697 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessage
]
699 self
.assertEqual(len(msgs
), 2)
700 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
701 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
704 if __name__
== '__main__':