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_CTF_TRACES_PATH
= os
.environ
['BT_CTF_TRACES_PATH']
27 _3EVENTS_INTERSECT_TRACE_PATH
= os
.path
.join(_BT_CTF_TRACES_PATH
,
32 class ComponentSpecTestCase(unittest
.TestCase
):
33 def test_create_good_no_params(self
):
34 bt2
.ComponentSpec('plugin', 'compcls')
36 def test_create_good_with_params(self
):
37 bt2
.ComponentSpec('plugin', 'compcls', {'salut': 23})
39 def test_create_good_with_path_params(self
):
40 spec
= bt2
.ComponentSpec('plugin', 'compcls', 'a path')
41 self
.assertEqual(spec
.params
['paths'], ['a path'])
43 def test_create_wrong_plugin_name_type(self
):
44 with self
.assertRaises(TypeError):
45 bt2
.ComponentSpec(23, 'compcls')
47 def test_create_wrong_component_class_name_type(self
):
48 with self
.assertRaises(TypeError):
49 bt2
.ComponentSpec('plugin', 190)
51 def test_create_wrong_params_type(self
):
52 with self
.assertRaises(TypeError):
53 bt2
.ComponentSpec('dwdw', 'compcls', datetime
.datetime
.now())
56 # Return a map, msg type -> number of messages of this type.
58 def _count_msgs_by_type(msgs
):
69 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
70 def test_create_wrong_stream_intersection_mode_type(self
):
71 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
73 with self
.assertRaises(TypeError):
74 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
76 def test_create_wrong_begin_type(self
):
77 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
79 with self
.assertRaises(TypeError):
80 bt2
.TraceCollectionMessageIterator(specs
, begin
='hi')
82 def test_create_wrong_end_type(self
):
83 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
85 with self
.assertRaises(TypeError):
86 bt2
.TraceCollectionMessageIterator(specs
, begin
='lel')
88 def test_create_no_such_plugin(self
):
89 specs
= [bt2
.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH
)]
91 with self
.assertRaises(bt2
.Error
):
92 bt2
.TraceCollectionMessageIterator(specs
)
94 def test_create_begin_s(self
):
95 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
96 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
98 def test_create_end_s(self
):
99 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
100 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
102 def test_create_begin_datetime(self
):
103 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
104 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
106 def test_create_end_datetime(self
):
107 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
108 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
110 def test_iter_no_intersection(self
):
111 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
112 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
113 msgs
= list(msg_iter
)
114 self
.assertEqual(len(msgs
), 32)
115 hist
= _count_msgs_by_type(msgs
)
116 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 8)
118 # Same as the above, but we pass a single spec instead of a spec list.
119 def test_iter_specs_not_list(self
):
120 spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
121 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
122 msgs
= list(msg_iter
)
123 self
.assertEqual(len(msgs
), 32)
124 hist
= _count_msgs_by_type(msgs
)
125 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 8)
127 def test_iter_custom_filter(self
):
128 src_spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
129 flt_spec
= bt2
.ComponentSpec('utils', 'trimmer', {
130 'end': '13515309.000000075',
132 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
133 hist
= _count_msgs_by_type(msg_iter
)
134 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 5)
136 def test_iter_intersection(self
):
137 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
138 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=True)
139 msgs
= list(msg_iter
)
140 self
.assertEqual(len(msgs
), 19)
141 hist
= _count_msgs_by_type(msgs
)
142 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 3)
144 def test_iter_intersection_no_path_param(self
):
145 specs
= [bt2
.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
147 with self
.assertRaises(bt2
.Error
):
148 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=True)
150 def test_iter_no_intersection_two_traces(self
):
151 spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
153 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
154 msgs
= list(msg_iter
)
155 self
.assertEqual(len(msgs
), 64)
156 hist
= _count_msgs_by_type(msgs
)
157 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 16)
159 def test_iter_no_intersection_begin(self
):
160 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
161 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
162 hist
= _count_msgs_by_type(msg_iter
)
163 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 6)
165 def test_iter_no_intersection_end(self
):
166 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
167 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
168 hist
= _count_msgs_by_type(msg_iter
)
169 self
.assertEqual(hist
[bt2
.message
._EventMessage
], 5)