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(
104 "'_SinkComponentClassConst' is not a source or filter component class",
106 bt2
.ComponentSpec(self
._pretty
_cc
)
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)
112 def test_create_from_name_with_params(self
):
113 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
114 'text', 'dmesg', {'salut': 23}
116 self
.assertEqual(spec
.params
['salut'], 23)
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'])
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'
126 self
.assertEqual(spec
.params
['inputs'], ['a path'])
128 def test_create_wrong_comp_class_type(self
):
129 with self
.assertRaisesRegex(
130 TypeError, "'int' is not a source or filter component class"
132 bt2
.ComponentSpec(18)
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')
138 def test_create_from_name_non_existent_plugin(self
):
139 with self
.assertRaisesRegex(
140 ValueError, "no such plugin: this_plugin_does_not_exist"
142 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
143 'this_plugin_does_not_exist', 'compcls'
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)
150 def test_create_wrong_params_type(self
):
151 with self
.assertRaisesRegex(
152 TypeError, "cannot create value object from 'datetime' object"
154 bt2
.ComponentSpec(self
._dmesg
_cc
, params
=datetime
.datetime
.now())
156 def test_create_from_name_wrong_params_type(self
):
157 with self
.assertRaisesRegex(
158 TypeError, "cannot create value object from 'datetime' object"
160 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
161 'text', 'dmesg', datetime
.datetime
.now()
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')
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'
175 # Return a map, msg type -> number of messages of this type.
178 def _count_msgs_by_type(msgs
):
189 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
190 def test_create_wrong_stream_intersection_mode_type(self
):
192 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
193 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
197 with self
.assertRaises(TypeError):
198 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
200 def test_create_wrong_begin_type(self
):
202 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
203 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
207 with self
.assertRaises(TypeError):
208 bt2
.TraceCollectionMessageIterator(specs
, begin
='hi')
210 def test_create_wrong_end_type(self
):
212 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
213 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
217 with self
.assertRaises(TypeError):
218 bt2
.TraceCollectionMessageIterator(specs
, begin
='lel')
220 def test_create_begin_s(self
):
222 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
223 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
226 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
228 def test_create_end_s(self
):
230 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
231 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
234 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
236 def test_create_begin_datetime(self
):
238 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
239 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
242 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
244 def test_create_end_datetime(self
):
246 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
247 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
250 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
252 def test_iter_no_intersection(self
):
254 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
255 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
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)
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
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)
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
279 flt_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
280 'utils', 'trimmer', {'end': '13515309.000000075'}
282 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
283 hist
= _count_msgs_by_type(msg_iter
)
284 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
286 def test_iter_intersection(self
):
288 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
289 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
292 msg_iter
= bt2
.TraceCollectionMessageIterator(
293 specs
, stream_intersection_mode
=True
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)
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.
304 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
308 'inputs': [_3EVENTS_INTERSECT_TRACE_PATH
],
309 'clock-class-offset-s': 1000,
314 msg_iter
= bt2
.TraceCollectionMessageIterator(
315 specs
, stream_intersection_mode
=True
318 event_msgs
= [x
for x
in msg_iter
if type(x
) is bt2
._EventMessageConst
]
319 self
.assertEqual(len(event_msgs
), 3)
321 event_msgs
[0].default_clock_snapshot
.ns_from_origin
, 13516309000000071
324 event_msgs
[1].default_clock_snapshot
.ns_from_origin
, 13516309000000072
327 event_msgs
[2].default_clock_snapshot
.ns_from_origin
, 13516309000000082
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
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)
341 def test_iter_no_intersection_begin(self
):
343 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
344 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
347 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
348 hist
= _count_msgs_by_type(msg_iter
)
349 self
.assertEqual(hist
[bt2
._EventMessageConst
], 6)
351 def test_iter_no_intersection_end(self
):
353 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
354 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
357 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
358 hist
= _count_msgs_by_type(msg_iter
)
359 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
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)
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)
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)
383 def test_iter_mixed_inputs(self
):
384 msg_iter
= bt2
.TraceCollectionMessageIterator(
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
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)
398 def test_auto_source_component_non_existent(self
):
399 with self
.assertRaisesRegex(
401 'Some auto source component specs did not produce any component',
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']
410 class _TestAutoDiscoverSourceComponentSpecs(unittest
.TestCase
):
412 self
._saved
_babeltrace
_plugin
_path
= os
.environ
['BABELTRACE_PLUGIN_PATH']
413 os
.environ
['BABELTRACE_PLUGIN_PATH'] += os
.pathsep
+ self
._plugin
_path
416 os
.environ
['BABELTRACE_PLUGIN_PATH'] = self
._saved
_babeltrace
_plugin
_path
419 class TestAutoDiscoverSourceComponentSpecsGrouping(
420 _TestAutoDiscoverSourceComponentSpecs
422 _plugin_path
= _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
424 def test_grouping(self
):
426 bt2
.AutoSourceComponentSpec('ABCDE'),
427 bt2
.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH
),
429 it
= bt2
.TraceCollectionMessageIterator(specs
)
430 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
432 self
.assertEqual(len(msgs
), 8)
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')
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 if __name__
== '__main__':