Fix: bt2: autodisc: remove thread error while inserting status in map
[babeltrace.git] / tests / bindings / python / bt2 / test_trace_collection_message_iterator.py
1 # SPDX-License-Identifier: GPL-2.0-only
2 #
3 # Copyright (C) 2019 EfficiOS Inc.
4 #
5
6 import unittest
7 import datetime
8 import bt2
9 import os
10 import os.path
11
12
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"
17 )
18 _NOINTERSECT_TRACE_PATH = os.path.join(
19 _BT_CTF_TRACES_PATH, "intersection", "nointersect"
20 )
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"
24 )
25 _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH = os.path.join(
26 _BT_TESTS_DATADIR, "auto-source-discovery", "params-log-level"
27 )
28
29 _METADATA_SYNTAX_ERROR_TRACE_PATH = os.path.join(
30 _BT_CTF_TRACES_PATH, "fail", "metadata-syntax-error"
31 )
32
33
34 class _SomeSource(
35 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
36 ):
37 pass
38
39
40 class _SomeFilter(
41 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
42 ):
43 pass
44
45
46 class _SomeSink(bt2._UserSinkComponent):
47 def _user_consume(self):
48 pass
49
50
51 class ComponentSpecTestCase(unittest.TestCase):
52 def setUp(self):
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
56
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
60
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
64
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")
68
69 def test_create_source_from_plugin(self):
70 spec = bt2.ComponentSpec(self._dmesg_cc)
71 self.assertEqual(spec.component_class.name, "dmesg")
72
73 def test_create_source_from_user(self):
74 spec = bt2.ComponentSpec(_SomeSource)
75 self.assertEqual(spec.component_class.name, "_SomeSource")
76
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")
80
81 def test_create_filter_from_object(self):
82 spec = bt2.ComponentSpec(self._muxer_cc)
83 self.assertEqual(spec.component_class.name, "muxer")
84
85 def test_create_sink_from_name(self):
86 with self.assertRaisesRegex(
87 KeyError,
88 "source or filter component class `pretty` not found in plugin `text`",
89 ):
90 bt2.ComponentSpec.from_named_plugin_and_component_class("text", "pretty")
91
92 def test_create_sink_from_object(self):
93 with self.assertRaisesRegex(
94 TypeError,
95 "'_SinkComponentClassConst' is not a source or filter component class",
96 ):
97 bt2.ComponentSpec(self._pretty_cc)
98
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)
102
103 def test_create_from_name_with_params(self):
104 spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
105 "text", "dmesg", {"salut": 23}
106 )
107 self.assertEqual(spec.params["salut"], 23)
108
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"])
112
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"
116 )
117 self.assertEqual(spec.params["inputs"], ["a path"])
118
119 def test_create_wrong_comp_class_type(self):
120 with self.assertRaisesRegex(
121 TypeError, "'int' is not a source or filter component class"
122 ):
123 bt2.ComponentSpec(18)
124
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")
128
129 def test_create_from_name_non_existent_plugin(self):
130 with self.assertRaisesRegex(
131 ValueError, "no such plugin: this_plugin_does_not_exist"
132 ):
133 bt2.ComponentSpec.from_named_plugin_and_component_class(
134 "this_plugin_does_not_exist", "compcls"
135 )
136
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)
140
141 def test_create_wrong_params_type(self):
142 with self.assertRaisesRegex(
143 TypeError, "cannot create value object from 'datetime' object"
144 ):
145 bt2.ComponentSpec(self._dmesg_cc, params=datetime.datetime.now())
146
147 def test_create_from_name_wrong_params_type(self):
148 with self.assertRaisesRegex(
149 TypeError, "cannot create value object from 'datetime' object"
150 ):
151 bt2.ComponentSpec.from_named_plugin_and_component_class(
152 "text", "dmesg", datetime.datetime.now()
153 )
154
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")
158
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"
163 )
164
165
166 # Return a map, msg type -> number of messages of this type.
167
168
169 def _count_msgs_by_type(msgs):
170 res = {}
171
172 for msg in msgs:
173 t = type(msg)
174 n = res.get(t, 0)
175 res[t] = n + 1
176
177 return res
178
179
180 class TraceCollectionMessageIteratorTestCase(unittest.TestCase):
181 def test_create_wrong_stream_intersection_mode_type(self):
182 specs = [
183 bt2.ComponentSpec.from_named_plugin_and_component_class(
184 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
185 )
186 ]
187
188 with self.assertRaises(TypeError):
189 bt2.TraceCollectionMessageIterator(specs, stream_intersection_mode=23)
190
191 def test_create_wrong_begin_type(self):
192 specs = [
193 bt2.ComponentSpec.from_named_plugin_and_component_class(
194 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
195 )
196 ]
197
198 with self.assertRaises(TypeError):
199 bt2.TraceCollectionMessageIterator(specs, begin="hi")
200
201 def test_create_wrong_end_type(self):
202 specs = [
203 bt2.ComponentSpec.from_named_plugin_and_component_class(
204 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
205 )
206 ]
207
208 with self.assertRaises(TypeError):
209 bt2.TraceCollectionMessageIterator(specs, begin="lel")
210
211 def test_create_begin_s(self):
212 specs = [
213 bt2.ComponentSpec.from_named_plugin_and_component_class(
214 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
215 )
216 ]
217 bt2.TraceCollectionMessageIterator(specs, begin=19457.918232)
218
219 def test_create_end_s(self):
220 specs = [
221 bt2.ComponentSpec.from_named_plugin_and_component_class(
222 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
223 )
224 ]
225 bt2.TraceCollectionMessageIterator(specs, end=123.12312)
226
227 def test_create_begin_datetime(self):
228 specs = [
229 bt2.ComponentSpec.from_named_plugin_and_component_class(
230 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
231 )
232 ]
233 bt2.TraceCollectionMessageIterator(specs, begin=datetime.datetime.now())
234
235 def test_create_end_datetime(self):
236 specs = [
237 bt2.ComponentSpec.from_named_plugin_and_component_class(
238 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
239 )
240 ]
241 bt2.TraceCollectionMessageIterator(specs, end=datetime.datetime.now())
242
243 def test_iter_no_intersection(self):
244 specs = [
245 bt2.ComponentSpec.from_named_plugin_and_component_class(
246 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
247 )
248 ]
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)
254
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
259 )
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)
265
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
269 )
270 flt_spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
271 "utils", "trimmer", {"end": "13515309.000000075"}
272 )
273 msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
274 hist = _count_msgs_by_type(msg_iter)
275 self.assertEqual(hist[bt2._EventMessageConst], 5)
276
277 def test_iter_intersection(self):
278 specs = [
279 bt2.ComponentSpec.from_named_plugin_and_component_class(
280 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
281 )
282 ]
283 msg_iter = bt2.TraceCollectionMessageIterator(
284 specs, stream_intersection_mode=True
285 )
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)
290
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.
294 specs = [
295 bt2.ComponentSpec.from_named_plugin_and_component_class(
296 "ctf",
297 "fs",
298 {
299 "inputs": [_3EVENTS_INTERSECT_TRACE_PATH],
300 "clock-class-offset-s": 1000,
301 },
302 )
303 ]
304
305 msg_iter = bt2.TraceCollectionMessageIterator(
306 specs, stream_intersection_mode=True
307 )
308
309 event_msgs = [x for x in msg_iter if type(x) is bt2._EventMessageConst]
310 self.assertEqual(len(event_msgs), 3)
311 self.assertEqual(
312 event_msgs[0].default_clock_snapshot.ns_from_origin, 13516309000000071
313 )
314 self.assertEqual(
315 event_msgs[1].default_clock_snapshot.ns_from_origin, 13516309000000072
316 )
317 self.assertEqual(
318 event_msgs[2].default_clock_snapshot.ns_from_origin, 13516309000000082
319 )
320
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
324 )
325 specs = [spec, spec]
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)
331
332 def test_iter_no_intersection_begin(self):
333 specs = [
334 bt2.ComponentSpec.from_named_plugin_and_component_class(
335 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
336 )
337 ]
338 msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
339 hist = _count_msgs_by_type(msg_iter)
340 self.assertEqual(hist[bt2._EventMessageConst], 6)
341
342 def test_iter_no_intersection_end(self):
343 specs = [
344 bt2.ComponentSpec.from_named_plugin_and_component_class(
345 "ctf", "fs", _3EVENTS_INTERSECT_TRACE_PATH
346 )
347 ]
348 msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
349 hist = _count_msgs_by_type(msg_iter)
350 self.assertEqual(hist[bt2._EventMessageConst], 5)
351
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)
359
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)
366
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)
373
374 def test_iter_mixed_inputs(self):
375 msg_iter = bt2.TraceCollectionMessageIterator(
376 [
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
381 ),
382 ]
383 )
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)
388
389 def test_auto_source_component_non_existent(self):
390 with self.assertRaisesRegex(
391 RuntimeError,
392 "Some auto source component specs did not produce any component",
393 ):
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"]
398 )
399
400
401 class _TestAutoDiscoverSourceComponentSpecs(unittest.TestCase):
402 def setUp(self):
403 self._saved_babeltrace_plugin_path = os.environ["BABELTRACE_PLUGIN_PATH"]
404 os.environ["BABELTRACE_PLUGIN_PATH"] += os.pathsep + self._plugin_path
405
406 def tearDown(self):
407 os.environ["BABELTRACE_PLUGIN_PATH"] = self._saved_babeltrace_plugin_path
408
409
410 class TestAutoDiscoverSourceComponentSpecsGrouping(
411 _TestAutoDiscoverSourceComponentSpecs
412 ):
413 _plugin_path = _AUTO_SOURCE_DISCOVERY_GROUPING_PATH
414
415 def test_grouping(self):
416 specs = [
417 bt2.AutoSourceComponentSpec("ABCDE"),
418 bt2.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH),
419 ]
420 it = bt2.TraceCollectionMessageIterator(specs)
421 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
422
423 self.assertEqual(len(msgs), 8)
424
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")
433
434
435 class TestAutoDiscoverSourceComponentSpecsParamsObjLogLevel(
436 _TestAutoDiscoverSourceComponentSpecs
437 ):
438 _plugin_path = _AUTO_SOURCE_DISCOVERY_PARAMS_LOG_LEVEL_PATH
439
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")
443
444 def _test_two_comps_from_one_spec(self, params, obj=None, logging_level=None):
445 specs = [
446 bt2.AutoSourceComponentSpec(
447 self._dir_ab, params=params, obj=obj, logging_level=logging_level
448 )
449 ]
450 it = bt2.TraceCollectionMessageIterator(specs)
451 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
452
453 self.assertEqual(len(msgs), 2)
454
455 return msgs
456
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"}
460 )
461
462 self.assertEqual(msgs[0].stream.name, "TestSourceA: ('test-allo', 'madame')")
463 self.assertEqual(msgs[1].stream.name, "TestSourceB: ('test-allo', 'madame')")
464
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"
468 )
469
470 self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
471 self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
472
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
476 )
477
478 self.assertEqual(
479 msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.DEBUG)
480 )
481 self.assertEqual(
482 msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.DEBUG)
483 )
484
485 def _test_two_comps_from_two_specs(
486 self,
487 params_a=None,
488 params_b=None,
489 obj_a=None,
490 obj_b=None,
491 logging_level_a=None,
492 logging_level_b=None,
493 ):
494 specs = [
495 bt2.AutoSourceComponentSpec(
496 self._dir_a, params=params_a, obj=obj_a, logging_level=logging_level_a
497 ),
498 bt2.AutoSourceComponentSpec(
499 self._dir_b, params=params_b, obj=obj_b, logging_level=logging_level_b
500 ),
501 ]
502 it = bt2.TraceCollectionMessageIterator(specs)
503 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
504
505 self.assertEqual(len(msgs), 2)
506
507 return msgs
508
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"},
513 )
514
515 self.assertEqual(msgs[0].stream.name, "TestSourceA: ('test-allo', 'madame')")
516 self.assertEqual(
517 msgs[1].stream.name, "TestSourceB: ('test-bonjour', 'monsieur')"
518 )
519
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"},
524 obj_a="deore",
525 obj_b="alivio",
526 )
527
528 self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
529 self.assertEqual(msgs[1].stream.name, "TestSourceB: alivio")
530
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,
537 )
538
539 self.assertEqual(
540 msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.DEBUG)
541 )
542 self.assertEqual(
543 msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.TRACE)
544 )
545
546 def _test_one_comp_from_one_spec_one_comp_from_both_1(
547 self,
548 params_a=None,
549 params_ab=None,
550 obj_a=None,
551 obj_ab=None,
552 logging_level_a=None,
553 logging_level_ab=None,
554 ):
555 specs = [
556 bt2.AutoSourceComponentSpec(
557 self._dir_a, params=params_a, obj=obj_a, logging_level=logging_level_a
558 ),
559 bt2.AutoSourceComponentSpec(
560 self._dir_ab,
561 params=params_ab,
562 obj=obj_ab,
563 logging_level=logging_level_ab,
564 ),
565 ]
566 it = bt2.TraceCollectionMessageIterator(specs)
567 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
568
569 self.assertEqual(len(msgs), 2)
570
571 return msgs
572
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"},
577 )
578
579 self.assertEqual(
580 msgs[0].stream.name,
581 "TestSourceA: ('test-allo', 'madame'), ('test-bonjour', 'monsieur')",
582 )
583 self.assertEqual(
584 msgs[1].stream.name, "TestSourceB: ('test-bonjour', 'monsieur')"
585 )
586
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"},
591 obj_a="deore",
592 obj_ab="alivio",
593 )
594
595 self.assertEqual(msgs[0].stream.name, "TestSourceA: alivio")
596 self.assertEqual(msgs[1].stream.name, "TestSourceB: alivio")
597
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,
604 )
605
606 self.assertEqual(
607 msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.TRACE)
608 )
609 self.assertEqual(
610 msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.TRACE)
611 )
612
613 def _test_one_comp_from_one_spec_one_comp_from_both_2(
614 self,
615 params_ab=None,
616 params_a=None,
617 obj_ab=None,
618 obj_a=None,
619 logging_level_ab=None,
620 logging_level_a=None,
621 ):
622 specs = [
623 bt2.AutoSourceComponentSpec(
624 self._dir_ab,
625 params=params_ab,
626 obj=obj_ab,
627 logging_level=logging_level_ab,
628 ),
629 bt2.AutoSourceComponentSpec(
630 self._dir_a, params=params_a, obj=obj_a, logging_level=logging_level_a
631 ),
632 ]
633 it = bt2.TraceCollectionMessageIterator(specs)
634 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
635
636 self.assertEqual(len(msgs), 2)
637
638 return msgs
639
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(
642 params_ab={
643 "test-bonjour": "madame",
644 "test-salut": "les amis",
645 "what": "test-params",
646 },
647 params_a={"test-bonjour": "monsieur", "what": "test-params"},
648 )
649
650 self.assertEqual(
651 msgs[0].stream.name,
652 "TestSourceA: ('test-bonjour', 'monsieur'), ('test-salut', 'les amis')",
653 )
654 self.assertEqual(
655 msgs[1].stream.name,
656 "TestSourceB: ('test-bonjour', 'madame'), ('test-salut', 'les amis')",
657 )
658
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"},
663 obj_ab="deore",
664 obj_a="alivio",
665 )
666
667 self.assertEqual(msgs[0].stream.name, "TestSourceA: alivio")
668 self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
669
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,
676 )
677
678 self.assertEqual(
679 msgs[0].stream.name, "TestSourceA: {}".format(bt2.LoggingLevel.TRACE)
680 )
681 self.assertEqual(
682 msgs[1].stream.name, "TestSourceB: {}".format(bt2.LoggingLevel.DEBUG)
683 )
684
685 def test_obj_override_with_none(self):
686 specs = [
687 bt2.AutoSourceComponentSpec(
688 self._dir_ab, params={"what": "python-obj"}, obj="deore"
689 ),
690 bt2.AutoSourceComponentSpec(
691 self._dir_a, params={"what": "python-obj"}, obj=None
692 ),
693 ]
694 it = bt2.TraceCollectionMessageIterator(specs)
695 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
696
697 self.assertEqual(len(msgs), 2)
698 self.assertEqual(msgs[0].stream.name, "TestSourceA: None")
699 self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
700
701 def test_obj_no_override_with_no_obj(self):
702 specs = [
703 bt2.AutoSourceComponentSpec(
704 self._dir_ab, params={"what": "python-obj"}, obj="deore"
705 ),
706 bt2.AutoSourceComponentSpec(self._dir_a, params={"what": "python-obj"}),
707 ]
708 it = bt2.TraceCollectionMessageIterator(specs)
709 msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
710
711 self.assertEqual(len(msgs), 2)
712 self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
713 self.assertEqual(msgs[1].stream.name, "TestSourceB: deore")
714
715
716 class TestAutoDiscoverFailures(unittest.TestCase):
717 def test_metadata_syntax_error(self):
718 with self.assertRaisesRegex(
719 bt2._Error,
720 'At line 3 in metadata stream: syntax error, unexpected CTF_RSBRAC: token="]"',
721 ):
722 specs = [bt2.AutoSourceComponentSpec(_METADATA_SYNTAX_ERROR_TRACE_PATH)]
723 bt2.TraceCollectionMessageIterator(specs)
724
725
726 if __name__ == "__main__":
727 unittest.main()
This page took 0.071502 seconds and 4 git commands to generate.