a405c005d5c858bac09ce354da0029bc2e507f6c
1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
11 class UserComponentClassTestCase(unittest
.TestCase
):
12 def _test_no_init(self
, cls
):
13 with self
.assertRaises(RuntimeError):
16 def test_no_init_source(self
):
18 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
22 self
._test
_no
_init
(MySource
)
24 def test_no_init_filter(self
):
26 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
30 self
._test
_no
_init
(MyFilter
)
32 def test_no_init_sink(self
):
33 class MySink(bt2
._UserSinkComponent
):
34 def _user_consume(self
):
37 self
._test
_no
_init
(MySink
)
39 def test_incomplete_source_no_msg_iter_cls(self
):
40 with self
.assertRaises(bt2
._IncompleteUserClass
):
42 class MySource(bt2
._UserSourceComponent
):
45 def test_incomplete_source_wrong_msg_iter_cls_type(self
):
46 with self
.assertRaises(bt2
._IncompleteUserClass
):
48 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=int):
51 def test_incomplete_filter_no_msg_iter_cls(self
):
52 with self
.assertRaises(bt2
._IncompleteUserClass
):
54 class MyFilter(bt2
._UserFilterComponent
):
57 def test_incomplete_sink_no_consume_method(self
):
58 with self
.assertRaises(bt2
._IncompleteUserClass
):
60 class MySink(bt2
._UserSinkComponent
):
63 def test_minimal_source(self
):
65 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
69 def test_minimal_filter(self
):
71 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
75 def test_minimal_sink(self
):
76 class MySink(bt2
._UserSinkComponent
):
77 def _user_consume(self
):
80 def test_default_name(self
):
81 class MySink(bt2
._UserSinkComponent
):
82 def _user_consume(self
):
85 self
.assertEqual(MySink
.name
, "MySink")
87 def test_custom_name(self
):
88 class MySink(bt2
._UserSinkComponent
, name
="salut"):
89 def _user_consume(self
):
92 self
.assertEqual(MySink
.name
, "salut")
94 def test_invalid_custom_name(self
):
95 with self
.assertRaises(TypeError):
97 class MySink(bt2
._UserSinkComponent
, name
=23):
98 def _user_consume(self
):
101 def test_description(self
):
102 class MySink(bt2
._UserSinkComponent
):
106 Bacon ipsum dolor amet ribeye t-bone corned beef, beef jerky
107 porchetta burgdoggen prosciutto chicken frankfurter boudin
108 hamburger doner bacon turducken. Sirloin shank sausage,
109 boudin meatloaf alcatra meatball t-bone tongue pastrami
110 cupim flank tenderloin.
113 def _user_consume(self
):
116 self
.assertEqual(MySink
.description
, "The description.")
118 def test_empty_description_no_lines(self
):
119 class MySink(bt2
._UserSinkComponent
):
124 def _user_consume(self
):
127 self
.assertIsNone(MySink
.description
)
129 def test_empty_description_no_contents(self
):
130 class MySink(bt2
._UserSinkComponent
):
136 def _user_consume(self
):
139 self
.assertIsNone(MySink
.description
)
141 def test_empty_description_single_line(self
):
142 class MySink(bt2
._UserSinkComponent
):
145 def _user_consume(self
):
148 self
.assertEqual(MySink
.description
, "my description")
151 class MySink(bt2
._UserSinkComponent
):
160 def _user_consume(self
):
163 self
.assertEqual(MySink
.help, "The help\ntext is\nhere.")
166 class MySink(bt2
._UserSinkComponent
):
167 def _user_consume(self
):
170 self
.assertIsInstance(MySink
.addr
, int)
171 self
.assertNotEqual(MySink
.addr
, 0)
173 def test_query_not_implemented(self
):
174 class MySink(bt2
._UserSinkComponent
):
175 def _user_consume(self
):
178 with self
.assertRaises(bt2
.UnknownObject
):
179 bt2
.QueryExecutor(MySink
, "obj", 23).query()
181 def test_query_raises(self
):
182 class MySink(bt2
._UserSinkComponent
):
183 def _user_consume(self
):
187 def _user_query(cls
, priv_query_exec
, obj
, params
):
190 with self
.assertRaises(bt2
._Error
):
191 bt2
.QueryExecutor(MySink
, "obj", 23).query()
193 def test_query_wrong_return_type(self
):
194 class MySink(bt2
._UserSinkComponent
):
195 def _user_consume(self
):
199 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
202 with self
.assertRaises(bt2
._Error
):
203 bt2
.QueryExecutor(MySink
, "obj", 23).query()
205 def test_query_params_none(self
):
206 class MySink(bt2
._UserSinkComponent
):
207 def _user_consume(self
):
211 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
212 nonlocal query_params
213 query_params
= params
218 res
= bt2
.QueryExecutor(MySink
, "obj", params
).query()
219 self
.assertEqual(query_params
, params
)
220 self
.assertIsNone(res
)
223 def test_query_logging_level(self
):
224 class MySink(bt2
._UserSinkComponent
):
225 def _user_consume(self
):
229 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
230 nonlocal query_log_level
231 query_log_level
= priv_query_exec
.logging_level
233 query_log_level
= None
234 query_exec
= bt2
.QueryExecutor(MySink
, "obj", None)
235 query_exec
.logging_level
= bt2
.LoggingLevel
.WARNING
237 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.WARNING
)
240 def test_query_returns_none(self
):
241 class MySink(bt2
._UserSinkComponent
):
242 def _user_consume(self
):
246 def _user_query(priv_query_exec
, obj
, params
, method_obj
):
249 res
= bt2
.QueryExecutor(MySink
, "obj", None).query()
250 self
.assertIsNone(res
)
252 def test_query_simple(self
):
253 class MySink(bt2
._UserSinkComponent
):
254 def _user_consume(self
):
258 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
259 nonlocal query_params
260 query_params
= params
264 params
= ["coucou", 23, None]
265 res
= bt2
.QueryExecutor(MySink
, "obj", params
).query()
266 self
.assertEqual(query_params
, params
)
267 self
.assertEqual(res
, 17.5)
270 def test_query_complex(self
):
271 class MySink(bt2
._UserSinkComponent
):
272 def _user_consume(self
):
276 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
277 nonlocal query_params
278 query_params
= params
279 return {"null": None, "bt2": "BT2"}
283 "array": ["coucou", 23, None],
284 "other_map": {"yes": "yeah", "19": 19, "minus 1.5": -1.5},
288 res
= bt2
.QueryExecutor(MySink
, "obj", params
).query()
289 self
.assertEqual(query_params
, params
)
290 self
.assertEqual(res
, {"null": None, "bt2": "BT2"})
294 class MySink(bt2
._UserSinkComponent
):
295 def _user_consume(self
):
298 self
.assertEqual(MySink
, MySink
)
301 class ComponentClassTestCase(unittest
.TestCase
):
303 class MySink(bt2
._UserSinkComponent
):
310 def _user_consume(self
):
314 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
315 return [obj
, params
, 23]
317 self
._py
_comp
_cls
= MySink
319 comp
= graph
.add_component(MySink
, "salut")
320 self
._comp
_cls
= comp
.cls
321 self
.assertIs(type(self
._comp
_cls
), bt2
._SinkComponentClassConst
)
324 del self
._py
_comp
_cls
327 def test_description(self
):
328 self
.assertEqual(self
._comp
_cls
.description
, "The description.")
331 self
.assertEqual(self
._comp
_cls
.help, "The help.")
334 self
.assertEqual(self
._comp
_cls
.name
, "MySink")
337 self
.assertIsInstance(self
._comp
_cls
.addr
, int)
338 self
.assertNotEqual(self
._comp
_cls
.addr
, 0)
340 def test_eq_invalid(self
):
341 self
.assertFalse(self
._comp
_cls
== 23)
344 self
.assertEqual(self
._comp
_cls
, self
._comp
_cls
)
345 self
.assertEqual(self
._py
_comp
_cls
, self
._comp
_cls
)
347 def test_query(self
):
348 res
= bt2
.QueryExecutor(
349 self
._comp
_cls
, "an object", {"yes": "no", "book": -17}
351 expected
= ["an object", {"yes": "no", "book": -17}, 23]
352 self
.assertEqual(res
, expected
)
355 if __name__
== "__main__":
This page took 0.036754 seconds and 5 git commands to generate.