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'
42 _METADATA_SYNTAX_ERROR_TRACE_PATH
= os
.path
.join(
43 _BT_CTF_TRACES_PATH
, "fail", "metadata-syntax-error"
48 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
54 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
59 class _SomeSink(bt2
._UserSinkComponent
):
60 def _user_consume(self
):
64 class ComponentSpecTestCase(unittest
.TestCase
):
66 # A source CC from a plugin.
67 self
._dmesg
_cc
= bt2
.find_plugin('text').source_component_classes
['dmesg']
68 assert self
._dmesg
_cc
is not None
70 # A filter CC from a plugin.
71 self
._muxer
_cc
= bt2
.find_plugin('utils').filter_component_classes
['muxer']
72 assert self
._muxer
_cc
is not None
74 # A sink CC from a plugin.
75 self
._pretty
_cc
= bt2
.find_plugin('text').sink_component_classes
['pretty']
76 assert self
._pretty
_cc
is not None
78 def test_create_source_from_name(self
):
79 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class('text', 'dmesg')
80 self
.assertEqual(spec
.component_class
.name
, 'dmesg')
82 def test_create_source_from_plugin(self
):
83 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
)
84 self
.assertEqual(spec
.component_class
.name
, 'dmesg')
86 def test_create_source_from_user(self
):
87 spec
= bt2
.ComponentSpec(_SomeSource
)
88 self
.assertEqual(spec
.component_class
.name
, '_SomeSource')
90 def test_create_filter_from_name(self
):
91 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class('utils', 'muxer')
92 self
.assertEqual(spec
.component_class
.name
, 'muxer')
94 def test_create_filter_from_object(self
):
95 spec
= bt2
.ComponentSpec(self
._muxer
_cc
)
96 self
.assertEqual(spec
.component_class
.name
, 'muxer')
98 def test_create_sink_from_name(self
):
99 with self
.assertRaisesRegex(
101 'source or filter component class `pretty` not found in plugin `text`',
103 bt2
.ComponentSpec
.from_named_plugin_and_component_class('text', 'pretty')
105 def test_create_sink_from_object(self
):
106 with self
.assertRaisesRegex(
108 "'_SinkComponentClassConst' is not a source or filter component class",
110 bt2
.ComponentSpec(self
._pretty
_cc
)
112 def test_create_from_object_with_params(self
):
113 spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, {'salut': 23})
114 self
.assertEqual(spec
.params
['salut'], 23)
116 def test_create_from_name_with_params(self
):
117 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
118 'text', 'dmesg', {'salut': 23}
120 self
.assertEqual(spec
.params
['salut'], 23)
122 def test_create_from_object_with_path_params(self
):
123 spec
= spec
= bt2
.ComponentSpec(self
._dmesg
_cc
, 'a path')
124 self
.assertEqual(spec
.params
['inputs'], ['a path'])
126 def test_create_from_name_with_path_params(self
):
127 spec
= spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
128 'text', 'dmesg', 'a path'
130 self
.assertEqual(spec
.params
['inputs'], ['a path'])
132 def test_create_wrong_comp_class_type(self
):
133 with self
.assertRaisesRegex(
134 TypeError, "'int' is not a source or filter component class"
136 bt2
.ComponentSpec(18)
138 def test_create_from_name_wrong_plugin_name_type(self
):
139 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
140 bt2
.ComponentSpec
.from_named_plugin_and_component_class(23, 'compcls')
142 def test_create_from_name_non_existent_plugin(self
):
143 with self
.assertRaisesRegex(
144 ValueError, "no such plugin: this_plugin_does_not_exist"
146 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
147 'this_plugin_does_not_exist', 'compcls'
150 def test_create_from_name_wrong_component_class_name_type(self
):
151 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
152 bt2
.ComponentSpec
.from_named_plugin_and_component_class('utils', 190)
154 def test_create_wrong_params_type(self
):
155 with self
.assertRaisesRegex(
156 TypeError, "cannot create value object from 'datetime' object"
158 bt2
.ComponentSpec(self
._dmesg
_cc
, params
=datetime
.datetime
.now())
160 def test_create_from_name_wrong_params_type(self
):
161 with self
.assertRaisesRegex(
162 TypeError, "cannot create value object from 'datetime' object"
164 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
165 'text', 'dmesg', datetime
.datetime
.now()
168 def test_create_wrong_log_level_type(self
):
169 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
170 bt2
.ComponentSpec(self
._dmesg
_cc
, logging_level
='banane')
172 def test_create_from_name_wrong_log_level_type(self
):
173 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
174 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
175 'text', 'dmesg', logging_level
='banane'
179 # Return a map, msg type -> number of messages of this type.
182 def _count_msgs_by_type(msgs
):
193 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
194 def test_create_wrong_stream_intersection_mode_type(self
):
196 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
197 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
201 with self
.assertRaises(TypeError):
202 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
204 def test_create_wrong_begin_type(self
):
206 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
207 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
211 with self
.assertRaises(TypeError):
212 bt2
.TraceCollectionMessageIterator(specs
, begin
='hi')
214 def test_create_wrong_end_type(self
):
216 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
217 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
221 with self
.assertRaises(TypeError):
222 bt2
.TraceCollectionMessageIterator(specs
, begin
='lel')
224 def test_create_begin_s(self
):
226 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
227 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
230 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
232 def test_create_end_s(self
):
234 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
235 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
238 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
240 def test_create_begin_datetime(self
):
242 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
243 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
246 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
248 def test_create_end_datetime(self
):
250 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
251 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
254 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
256 def test_iter_no_intersection(self
):
258 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
259 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
262 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
263 msgs
= list(msg_iter
)
264 self
.assertEqual(len(msgs
), 28)
265 hist
= _count_msgs_by_type(msgs
)
266 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
268 # Same as the above, but we pass a single spec instead of a spec list.
269 def test_iter_specs_not_list(self
):
270 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
271 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
273 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
274 msgs
= list(msg_iter
)
275 self
.assertEqual(len(msgs
), 28)
276 hist
= _count_msgs_by_type(msgs
)
277 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
279 def test_iter_custom_filter(self
):
280 src_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
281 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
283 flt_spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
284 'utils', 'trimmer', {'end': '13515309.000000075'}
286 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
287 hist
= _count_msgs_by_type(msg_iter
)
288 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
290 def test_iter_intersection(self
):
292 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
293 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
296 msg_iter
= bt2
.TraceCollectionMessageIterator(
297 specs
, stream_intersection_mode
=True
299 msgs
= list(msg_iter
)
300 self
.assertEqual(len(msgs
), 15)
301 hist
= _count_msgs_by_type(msgs
)
302 self
.assertEqual(hist
[bt2
._EventMessageConst
], 3)
304 def test_iter_intersection_params(self
):
305 # Check that all params used to create the source component are passed
306 # to the `babeltrace.trace-infos` query.
308 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
312 'inputs': [_3EVENTS_INTERSECT_TRACE_PATH
],
313 'clock-class-offset-s': 1000,
318 msg_iter
= bt2
.TraceCollectionMessageIterator(
319 specs
, stream_intersection_mode
=True
322 event_msgs
= [x
for x
in msg_iter
if type(x
) is bt2
._EventMessageConst
]
323 self
.assertEqual(len(event_msgs
), 3)
325 event_msgs
[0].default_clock_snapshot
.ns_from_origin
, 13516309000000071
328 event_msgs
[1].default_clock_snapshot
.ns_from_origin
, 13516309000000072
331 event_msgs
[2].default_clock_snapshot
.ns_from_origin
, 13516309000000082
334 def test_iter_no_intersection_two_traces(self
):
335 spec
= bt2
.ComponentSpec
.from_named_plugin_and_component_class(
336 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
339 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
340 msgs
= list(msg_iter
)
341 self
.assertEqual(len(msgs
), 56)
342 hist
= _count_msgs_by_type(msgs
)
343 self
.assertEqual(hist
[bt2
._EventMessageConst
], 16)
345 def test_iter_no_intersection_begin(self
):
347 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
348 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
351 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
352 hist
= _count_msgs_by_type(msg_iter
)
353 self
.assertEqual(hist
[bt2
._EventMessageConst
], 6)
355 def test_iter_no_intersection_end(self
):
357 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
358 'ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
361 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
362 hist
= _count_msgs_by_type(msg_iter
)
363 self
.assertEqual(hist
[bt2
._EventMessageConst
], 5)
365 def test_iter_auto_source_component_spec(self
):
366 specs
= [bt2
.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH
)]
367 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
368 msgs
= list(msg_iter
)
369 self
.assertEqual(len(msgs
), 28)
370 hist
= _count_msgs_by_type(msgs
)
371 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
373 def test_iter_auto_source_component_spec_list_of_strings(self
):
374 msg_iter
= bt2
.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH
])
375 msgs
= list(msg_iter
)
376 self
.assertEqual(len(msgs
), 28)
377 hist
= _count_msgs_by_type(msgs
)
378 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
380 def test_iter_auto_source_component_spec_string(self
):
381 msg_iter
= bt2
.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH
)
382 msgs
= list(msg_iter
)
383 self
.assertEqual(len(msgs
), 28)
384 hist
= _count_msgs_by_type(msgs
)
385 self
.assertEqual(hist
[bt2
._EventMessageConst
], 8)
387 def test_iter_mixed_inputs(self
):
388 msg_iter
= bt2
.TraceCollectionMessageIterator(
390 _3EVENTS_INTERSECT_TRACE_PATH
,
391 bt2
.AutoSourceComponentSpec(_SEQUENCE_TRACE_PATH
),
392 bt2
.ComponentSpec
.from_named_plugin_and_component_class(
393 'ctf', 'fs', _NOINTERSECT_TRACE_PATH
397 msgs
= list(msg_iter
)
398 self
.assertEqual(len(msgs
), 76)
399 hist
= _count_msgs_by_type(msgs
)
400 self
.assertEqual(hist
[bt2
._EventMessageConst
], 24)
402 def test_auto_source_component_non_existent(self
):
403 with self
.assertRaisesRegex(
405 'Some auto source component specs did not produce any component',
407 # Test with one path known to contain a trace and one path known
408 # to not contain any trace.
409 bt2
.TraceCollectionMessageIterator(
410 [_SEQUENCE_TRACE_PATH
, '/this/path/better/not/exist']
414 class _TestAutoDiscoverSourceComponentSpecs(unittest
.TestCase
):
416 self
._saved
_babeltrace
_plugin
_path
= os
.environ
['BABELTRACE_PLUGIN_PATH']
417 os
.environ
['BABELTRACE_PLUGIN_PATH'] += os
.pathsep
+ self
._plugin
_path
420 os
.environ
['BABELTRACE_PLUGIN_PATH'] = self
._saved
_babeltrace
_plugin
_path
423 class TestAutoDiscoverSourceComponentSpecsGrouping(
424 _TestAutoDiscoverSourceComponentSpecs
426 _plugin_path
= _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
428 def test_grouping(self
):
430 bt2
.AutoSourceComponentSpec('ABCDE'),
431 bt2
.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH
),
433 it
= bt2
.TraceCollectionMessageIterator(specs
)
434 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
436 self
.assertEqual(len(msgs
), 8)
438 self
.assertEqual(msgs
[0].stream
.name
, 'TestSourceABCDE: ABCDE')
439 self
.assertEqual(msgs
[1].stream
.name
, 'TestSourceExt: aaa1, aaa2, aaa3')
440 self
.assertEqual(msgs
[2].stream
.name
, 'TestSourceExt: bbb1, bbb2')
441 self
.assertEqual(msgs
[3].stream
.name
, 'TestSourceExt: ccc1')
442 self
.assertEqual(msgs
[4].stream
.name
, 'TestSourceExt: ccc2')
443 self
.assertEqual(msgs
[5].stream
.name
, 'TestSourceExt: ccc3')
444 self
.assertEqual(msgs
[6].stream
.name
, 'TestSourceExt: ccc4')
445 self
.assertEqual(msgs
[7].stream
.name
, 'TestSourceSomeDir: some-dir')
448 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
449 _TestAutoDiscoverSourceComponentSpecs
451 _plugin_path
= _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
453 _dir_a
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, 'dir-a')
454 _dir_b
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, 'dir-b')
455 _dir_ab
= os
.path
.join(_AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
, 'dir-ab')
457 def _test_two_comps_from_one_spec(self
, params
, obj
=None, logging_level
=None):
459 bt2
.AutoSourceComponentSpec(
460 self
._dir
_ab
, params
=params
, obj
=obj
, logging_level
=logging_level
463 it
= bt2
.TraceCollectionMessageIterator(specs
)
464 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
466 self
.assertEqual(len(msgs
), 2)
470 def test_params_two_comps_from_one_spec(self
):
471 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
472 params
={'test-allo': 'madame', 'what': 'test-params'}
475 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
476 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: ('test-allo', 'madame')")
478 def test_obj_two_comps_from_one_spec(self
):
479 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
480 params
={'what': 'python-obj'}, obj
='deore'
483 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
484 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
486 def test_log_level_two_comps_from_one_spec(self
):
487 msgs
= self
._test
_two
_comps
_from
_one
_spec
(
488 params
={'what': 'log-level'}, logging_level
=bt2
.LoggingLevel
.DEBUG
492 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
495 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
498 def _test_two_comps_from_two_specs(
504 logging_level_a
=None,
505 logging_level_b
=None,
508 bt2
.AutoSourceComponentSpec(
509 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
511 bt2
.AutoSourceComponentSpec(
512 self
._dir
_b
, params
=params_b
, obj
=obj_b
, logging_level
=logging_level_b
515 it
= bt2
.TraceCollectionMessageIterator(specs
)
516 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
518 self
.assertEqual(len(msgs
), 2)
522 def test_params_two_comps_from_two_specs(self
):
523 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
524 params_a
={'test-allo': 'madame', 'what': 'test-params'},
525 params_b
={'test-bonjour': 'monsieur', 'what': 'test-params'},
528 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: ('test-allo', 'madame')")
530 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
533 def test_obj_two_comps_from_two_specs(self
):
534 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
535 params_a
={'what': 'python-obj'},
536 params_b
={'what': 'python-obj'},
541 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
542 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
544 def test_log_level_two_comps_from_two_specs(self
):
545 msgs
= self
._test
_two
_comps
_from
_two
_specs
(
546 params_a
={'what': 'log-level'},
547 params_b
={'what': 'log-level'},
548 logging_level_a
=bt2
.LoggingLevel
.DEBUG
,
549 logging_level_b
=bt2
.LoggingLevel
.TRACE
,
553 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.DEBUG
)
556 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
559 def _test_one_comp_from_one_spec_one_comp_from_both_1(
565 logging_level_a
=None,
566 logging_level_ab
=None,
569 bt2
.AutoSourceComponentSpec(
570 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
572 bt2
.AutoSourceComponentSpec(
576 logging_level
=logging_level_ab
,
579 it
= bt2
.TraceCollectionMessageIterator(specs
)
580 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
582 self
.assertEqual(len(msgs
), 2)
586 def test_params_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
={'test-allo': 'madame', 'what': 'test-params'},
589 params_ab
={'test-bonjour': 'monsieur', 'what': 'test-params'},
594 "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
597 msgs
[1].stream
.name
, "TestSourceB: ('test-bonjour', 'monsieur')"
600 def test_obj_one_comp_from_one_spec_one_comp_from_both_1(self
):
601 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
602 params_a
={'what': 'python-obj'},
603 params_ab
={'what': 'python-obj'},
608 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
609 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: alivio")
611 def test_log_level_one_comp_from_one_spec_one_comp_from_both_1(self
):
612 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_1(
613 params_a
={'what': 'log-level'},
614 params_ab
={'what': 'log-level'},
615 logging_level_a
=bt2
.LoggingLevel
.DEBUG
,
616 logging_level_ab
=bt2
.LoggingLevel
.TRACE
,
620 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
623 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.TRACE
)
626 def _test_one_comp_from_one_spec_one_comp_from_both_2(
632 logging_level_ab
=None,
633 logging_level_a
=None,
636 bt2
.AutoSourceComponentSpec(
640 logging_level
=logging_level_ab
,
642 bt2
.AutoSourceComponentSpec(
643 self
._dir
_a
, params
=params_a
, obj
=obj_a
, logging_level
=logging_level_a
646 it
= bt2
.TraceCollectionMessageIterator(specs
)
647 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
649 self
.assertEqual(len(msgs
), 2)
653 def test_params_one_comp_from_one_spec_one_comp_from_both_2(self
):
654 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
656 'test-bonjour': 'madame',
657 'test-salut': 'les amis',
658 'what': 'test-params',
660 params_a
={'test-bonjour': 'monsieur', 'what': 'test-params'},
665 "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
669 "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
672 def test_obj_one_comp_from_one_spec_one_comp_from_both_2(self
):
673 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
674 params_ab
={'what': 'python-obj'},
675 params_a
={'what': 'python-obj'},
680 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: alivio")
681 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
683 def test_log_level_one_comp_from_one_spec_one_comp_from_both_2(self
):
684 msgs
= self
._test
_one
_comp
_from
_one
_spec
_one
_comp
_from
_both
_2(
685 params_ab
={'what': 'log-level'},
686 params_a
={'what': 'log-level'},
687 logging_level_ab
=bt2
.LoggingLevel
.DEBUG
,
688 logging_level_a
=bt2
.LoggingLevel
.TRACE
,
692 msgs
[0].stream
.name
, "TestSourceA: {}".format(bt2
.LoggingLevel
.TRACE
)
695 msgs
[1].stream
.name
, "TestSourceB: {}".format(bt2
.LoggingLevel
.DEBUG
)
698 def test_obj_override_with_none(self
):
700 bt2
.AutoSourceComponentSpec(
701 self
._dir
_ab
, params
={'what': 'python-obj'}, obj
='deore'
703 bt2
.AutoSourceComponentSpec(
704 self
._dir
_a
, params
={'what': 'python-obj'}, obj
=None
707 it
= bt2
.TraceCollectionMessageIterator(specs
)
708 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
710 self
.assertEqual(len(msgs
), 2)
711 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: None")
712 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
714 def test_obj_no_override_with_no_obj(self
):
716 bt2
.AutoSourceComponentSpec(
717 self
._dir
_ab
, params
={'what': 'python-obj'}, obj
='deore'
719 bt2
.AutoSourceComponentSpec(self
._dir
_a
, params
={'what': 'python-obj'}),
721 it
= bt2
.TraceCollectionMessageIterator(specs
)
722 msgs
= [x
for x
in it
if type(x
) is bt2
._StreamBeginningMessageConst
]
724 self
.assertEqual(len(msgs
), 2)
725 self
.assertEqual(msgs
[0].stream
.name
, "TestSourceA: deore")
726 self
.assertEqual(msgs
[1].stream
.name
, "TestSourceB: deore")
729 class TestAutoDiscoverFailures(unittest
.TestCase
):
730 def test_metadata_syntax_error(self
):
731 with self
.assertRaisesRegex(
733 'Component class\'s "query" method failed',
735 specs
= [bt2
.AutoSourceComponentSpec(_METADATA_SYNTAX_ERROR_TRACE_PATH
)]
736 bt2
.TraceCollectionMessageIterator(specs
)
739 if __name__
== '__main__':