8 _BT_CTF_TRACES_PATH
= os
.environ
['BT_CTF_TRACES_PATH']
9 _3EVENTS_INTERSECT_TRACE_PATH
= os
.path
.join(_BT_CTF_TRACES_PATH
,
14 class ComponentSpecTestCase(unittest
.TestCase
):
15 def test_create_good_no_params(self
):
16 bt2
.ComponentSpec('plugin', 'compcls')
18 def test_create_good_with_params(self
):
19 bt2
.ComponentSpec('plugin', 'compcls', {'salut': 23})
21 def test_create_good_with_path_params(self
):
22 spec
= bt2
.ComponentSpec('plugin', 'compcls', 'a path')
23 self
.assertEqual(spec
.params
['paths'], ['a path'])
25 def test_create_wrong_plugin_name_type(self
):
26 with self
.assertRaises(TypeError):
27 bt2
.ComponentSpec(23, 'compcls')
29 def test_create_wrong_component_class_name_type(self
):
30 with self
.assertRaises(TypeError):
31 bt2
.ComponentSpec('plugin', 190)
33 def test_create_wrong_params_type(self
):
34 with self
.assertRaises(TypeError):
35 bt2
.ComponentSpec('dwdw', 'compcls', datetime
.datetime
.now())
38 # Return a map, msg type -> number of messages of this type.
40 def _count_msgs_by_type(msgs
):
51 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
52 def test_create_wrong_stream_intersection_mode_type(self
):
53 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
55 with self
.assertRaises(TypeError):
56 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
58 def test_create_wrong_begin_type(self
):
59 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
61 with self
.assertRaises(TypeError):
62 bt2
.TraceCollectionMessageIterator(specs
, begin
='hi')
64 def test_create_wrong_end_type(self
):
65 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
67 with self
.assertRaises(TypeError):
68 bt2
.TraceCollectionMessageIterator(specs
, begin
='lel')
70 def test_create_no_such_plugin(self
):
71 specs
= [bt2
.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH
)]
73 with self
.assertRaises(bt2
.Error
):
74 bt2
.TraceCollectionMessageIterator(specs
)
76 def test_create_begin_s(self
):
77 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
78 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
80 def test_create_end_s(self
):
81 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
82 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
84 def test_create_begin_datetime(self
):
85 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
86 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
88 def test_create_end_datetime(self
):
89 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
90 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
92 def test_iter_no_intersection(self
):
93 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
94 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
96 self
.assertEqual(len(msgs
), 32)
97 hist
= _count_msgs_by_type(msgs
)
98 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 8)
100 # Same as the above, but we pass a single spec instead of a spec list.
101 def test_iter_specs_not_list(self
):
102 spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
103 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
104 msgs
= list(msg_iter
)
105 self
.assertEqual(len(msgs
), 32)
106 hist
= _count_msgs_by_type(msgs
)
107 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 8)
109 def test_iter_custom_filter(self
):
110 src_spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
111 flt_spec
= bt2
.ComponentSpec('utils', 'trimmer', {
112 'end': '13515309.000000075',
114 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
115 hist
= _count_msgs_by_type(msg_iter
)
116 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 5)
118 def test_iter_intersection(self
):
119 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
120 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=True)
121 msgs
= list(msg_iter
)
122 self
.assertEqual(len(msgs
), 19)
123 hist
= _count_msgs_by_type(msgs
)
124 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 3)
126 def test_iter_intersection_no_path_param(self
):
127 specs
= [bt2
.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
129 with self
.assertRaises(bt2
.Error
):
130 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=True)
132 def test_iter_no_intersection_two_traces(self
):
133 spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
135 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
136 msgs
= list(msg_iter
)
137 self
.assertEqual(len(msgs
), 64)
138 hist
= _count_msgs_by_type(msgs
)
139 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 16)
141 def test_iter_no_intersection_begin(self
):
142 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
143 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
144 hist
= _count_msgs_by_type(msg_iter
)
145 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 6)
147 def test_iter_no_intersection_end(self
):
148 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
149 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
150 hist
= _count_msgs_by_type(msg_iter
)
151 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 5)