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(
28 _BT_CTF_TRACES_PATH
, 'intersection', '3eventsintersect'
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
['inputs'], ['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.
59 def _count_msgs_by_type(msgs
):
70 class TraceCollectionMessageIteratorTestCase(unittest
.TestCase
):
71 def test_create_wrong_stream_intersection_mode_type(self
):
72 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
74 with self
.assertRaises(TypeError):
75 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=23)
77 def test_create_wrong_begin_type(self
):
78 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
80 with self
.assertRaises(TypeError):
81 bt2
.TraceCollectionMessageIterator(specs
, begin
='hi')
83 def test_create_wrong_end_type(self
):
84 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
86 with self
.assertRaises(TypeError):
87 bt2
.TraceCollectionMessageIterator(specs
, begin
='lel')
89 def test_create_no_such_plugin(self
):
90 specs
= [bt2
.ComponentSpec('77', '101', _3EVENTS_INTERSECT_TRACE_PATH
)]
92 with self
.assertRaises(ValueError):
93 bt2
.TraceCollectionMessageIterator(specs
)
95 def test_create_begin_s(self
):
96 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
97 bt2
.TraceCollectionMessageIterator(specs
, begin
=19457.918232)
99 def test_create_end_s(self
):
100 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
101 bt2
.TraceCollectionMessageIterator(specs
, end
=123.12312)
103 def test_create_begin_datetime(self
):
104 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
105 bt2
.TraceCollectionMessageIterator(specs
, begin
=datetime
.datetime
.now())
107 def test_create_end_datetime(self
):
108 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
109 bt2
.TraceCollectionMessageIterator(specs
, end
=datetime
.datetime
.now())
111 def test_iter_no_intersection(self
):
112 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
113 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
114 msgs
= list(msg_iter
)
115 self
.assertEqual(len(msgs
), 28)
116 hist
= _count_msgs_by_type(msgs
)
117 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
119 # Same as the above, but we pass a single spec instead of a spec list.
120 def test_iter_specs_not_list(self
):
121 spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
122 msg_iter
= bt2
.TraceCollectionMessageIterator(spec
)
123 msgs
= list(msg_iter
)
124 self
.assertEqual(len(msgs
), 28)
125 hist
= _count_msgs_by_type(msgs
)
126 self
.assertEqual(hist
[bt2
._EventMessage
], 8)
128 def test_iter_custom_filter(self
):
129 src_spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
130 flt_spec
= bt2
.ComponentSpec('utils', 'trimmer', {'end': '13515309.000000075'})
131 msg_iter
= bt2
.TraceCollectionMessageIterator(src_spec
, flt_spec
)
132 hist
= _count_msgs_by_type(msg_iter
)
133 self
.assertEqual(hist
[bt2
._EventMessage
], 5)
135 def test_iter_intersection(self
):
136 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
137 msg_iter
= bt2
.TraceCollectionMessageIterator(
138 specs
, stream_intersection_mode
=True
140 msgs
= list(msg_iter
)
141 self
.assertEqual(len(msgs
), 15)
142 hist
= _count_msgs_by_type(msgs
)
143 self
.assertEqual(hist
[bt2
._EventMessage
], 3)
145 def test_iter_intersection_no_inputs_param(self
):
146 specs
= [bt2
.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
148 with self
.assertRaises(ValueError):
149 bt2
.TraceCollectionMessageIterator(specs
, stream_intersection_mode
=True)
151 def test_iter_no_intersection_two_traces(self
):
152 spec
= bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)
154 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
)
155 msgs
= list(msg_iter
)
156 self
.assertEqual(len(msgs
), 56)
157 hist
= _count_msgs_by_type(msgs
)
158 self
.assertEqual(hist
[bt2
._EventMessage
], 16)
160 def test_iter_no_intersection_begin(self
):
161 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
162 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, begin
=13515309.000000023)
163 hist
= _count_msgs_by_type(msg_iter
)
164 self
.assertEqual(hist
[bt2
._EventMessage
], 6)
166 def test_iter_no_intersection_end(self
):
167 specs
= [bt2
.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH
)]
168 msg_iter
= bt2
.TraceCollectionMessageIterator(specs
, end
=13515309.000000075)
169 hist
= _count_msgs_by_type(msg_iter
)
170 self
.assertEqual(hist
[bt2
._EventMessage
], 5)