1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
10 class UserComponentClassTestCase(unittest
.TestCase
):
11 def _test_no_init(self
, cls
):
12 with self
.assertRaises(RuntimeError):
15 def test_no_init_source(self
):
17 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
21 self
._test
_no
_init
(MySource
)
23 def test_no_init_filter(self
):
25 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
29 self
._test
_no
_init
(MyFilter
)
31 def test_no_init_sink(self
):
32 class MySink(bt2
._UserSinkComponent
):
33 def _user_consume(self
):
36 self
._test
_no
_init
(MySink
)
38 def test_incomplete_source_no_msg_iter_cls(self
):
39 with self
.assertRaises(bt2
._IncompleteUserClass
):
41 class MySource(bt2
._UserSourceComponent
):
44 def test_incomplete_source_wrong_msg_iter_cls_type(self
):
45 with self
.assertRaises(bt2
._IncompleteUserClass
):
47 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=int):
50 def test_incomplete_filter_no_msg_iter_cls(self
):
51 with self
.assertRaises(bt2
._IncompleteUserClass
):
53 class MyFilter(bt2
._UserFilterComponent
):
56 def test_incomplete_sink_no_consume_method(self
):
57 with self
.assertRaises(bt2
._IncompleteUserClass
):
59 class MySink(bt2
._UserSinkComponent
):
62 def test_minimal_source(self
):
64 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
68 def test_minimal_filter(self
):
70 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
74 def test_minimal_sink(self
):
75 class MySink(bt2
._UserSinkComponent
):
76 def _user_consume(self
):
79 def test_default_name(self
):
80 class MySink(bt2
._UserSinkComponent
):
81 def _user_consume(self
):
84 self
.assertEqual(MySink
.name
, 'MySink')
86 def test_custom_name(self
):
87 class MySink(bt2
._UserSinkComponent
, name
='salut'):
88 def _user_consume(self
):
91 self
.assertEqual(MySink
.name
, 'salut')
93 def test_invalid_custom_name(self
):
94 with self
.assertRaises(TypeError):
96 class MySink(bt2
._UserSinkComponent
, name
=23):
97 def _user_consume(self
):
100 def test_description(self
):
101 class MySink(bt2
._UserSinkComponent
):
105 Bacon ipsum dolor amet ribeye t-bone corned beef, beef jerky
106 porchetta burgdoggen prosciutto chicken frankfurter boudin
107 hamburger doner bacon turducken. Sirloin shank sausage,
108 boudin meatloaf alcatra meatball t-bone tongue pastrami
109 cupim flank tenderloin.
112 def _user_consume(self
):
115 self
.assertEqual(MySink
.description
, 'The description.')
117 def test_empty_description_no_lines(self
):
118 class MySink(bt2
._UserSinkComponent
):
123 def _user_consume(self
):
126 self
.assertIsNone(MySink
.description
)
128 def test_empty_description_no_contents(self
):
129 class MySink(bt2
._UserSinkComponent
):
135 def _user_consume(self
):
138 self
.assertIsNone(MySink
.description
)
140 def test_empty_description_single_line(self
):
141 class MySink(bt2
._UserSinkComponent
):
144 def _user_consume(self
):
147 self
.assertEqual(MySink
.description
, "my description")
150 class MySink(bt2
._UserSinkComponent
):
159 def _user_consume(self
):
162 self
.assertEqual(MySink
.help, 'The help\ntext is\nhere.')
165 class MySink(bt2
._UserSinkComponent
):
166 def _user_consume(self
):
169 self
.assertIsInstance(MySink
.addr
, int)
170 self
.assertNotEqual(MySink
.addr
, 0)
172 def test_query_not_implemented(self
):
173 class MySink(bt2
._UserSinkComponent
):
174 def _user_consume(self
):
177 with self
.assertRaises(bt2
.UnknownObject
):
178 bt2
.QueryExecutor(MySink
, 'obj', 23).query()
180 def test_query_raises(self
):
181 class MySink(bt2
._UserSinkComponent
):
182 def _user_consume(self
):
186 def _user_query(cls
, priv_query_exec
, obj
, params
):
189 with self
.assertRaises(bt2
._Error
):
190 bt2
.QueryExecutor(MySink
, 'obj', 23).query()
192 def test_query_wrong_return_type(self
):
193 class MySink(bt2
._UserSinkComponent
):
194 def _user_consume(self
):
198 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
201 with self
.assertRaises(bt2
._Error
):
202 bt2
.QueryExecutor(MySink
, 'obj', 23).query()
204 def test_query_params_none(self
):
205 class MySink(bt2
._UserSinkComponent
):
206 def _user_consume(self
):
210 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
211 nonlocal query_params
212 query_params
= params
217 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
218 self
.assertEqual(query_params
, params
)
219 self
.assertIsNone(res
)
222 def test_query_logging_level(self
):
223 class MySink(bt2
._UserSinkComponent
):
224 def _user_consume(self
):
228 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
229 nonlocal query_log_level
230 query_log_level
= priv_query_exec
.logging_level
232 query_log_level
= None
233 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', None)
234 query_exec
.logging_level
= bt2
.LoggingLevel
.WARNING
236 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.WARNING
)
239 def test_query_returns_none(self
):
240 class MySink(bt2
._UserSinkComponent
):
241 def _user_consume(self
):
245 def _user_query(priv_query_exec
, obj
, params
, method_obj
):
248 res
= bt2
.QueryExecutor(MySink
, 'obj', None).query()
249 self
.assertIsNone(res
)
251 def test_query_simple(self
):
252 class MySink(bt2
._UserSinkComponent
):
253 def _user_consume(self
):
257 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
258 nonlocal query_params
259 query_params
= params
263 params
= ['coucou', 23, None]
264 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
265 self
.assertEqual(query_params
, params
)
266 self
.assertEqual(res
, 17.5)
269 def test_query_complex(self
):
270 class MySink(bt2
._UserSinkComponent
):
271 def _user_consume(self
):
275 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
276 nonlocal query_params
277 query_params
= params
278 return {'null': None, 'bt2': 'BT2'}
282 'array': ['coucou', 23, None],
283 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
287 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
288 self
.assertEqual(query_params
, params
)
289 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
293 class MySink(bt2
._UserSinkComponent
):
294 def _user_consume(self
):
297 self
.assertEqual(MySink
, MySink
)
300 class ComponentClassTestCase(unittest
.TestCase
):
302 class MySink(bt2
._UserSinkComponent
):
309 def _user_consume(self
):
313 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
314 return [obj
, params
, 23]
316 self
._py
_comp
_cls
= MySink
318 comp
= graph
.add_component(MySink
, 'salut')
319 self
._comp
_cls
= comp
.cls
320 self
.assertIs(type(self
._comp
_cls
), bt2
._SinkComponentClassConst
)
323 del self
._py
_comp
_cls
326 def test_description(self
):
327 self
.assertEqual(self
._comp
_cls
.description
, 'The description.')
330 self
.assertEqual(self
._comp
_cls
.help, 'The help.')
333 self
.assertEqual(self
._comp
_cls
.name
, 'MySink')
336 self
.assertIsInstance(self
._comp
_cls
.addr
, int)
337 self
.assertNotEqual(self
._comp
_cls
.addr
, 0)
339 def test_eq_invalid(self
):
340 self
.assertFalse(self
._comp
_cls
== 23)
343 self
.assertEqual(self
._comp
_cls
, self
._comp
_cls
)
344 self
.assertEqual(self
._py
_comp
_cls
, self
._comp
_cls
)
346 def test_query(self
):
347 res
= bt2
.QueryExecutor(
348 self
._comp
_cls
, 'an object', {'yes': 'no', 'book': -17}
350 expected
= ['an object', {'yes': 'no', 'book': -17}, 23]
351 self
.assertEqual(res
, expected
)
354 if __name__
== '__main__':
This page took 0.037122 seconds and 4 git commands to generate.