Add query executor
[babeltrace.git] / tests / bindings / python / bt2 / test_component_class.py
1 from bt2 import values
2 import unittest
3 import copy
4 import bt2
5
6
7 class UserComponentClassTestCase(unittest.TestCase):
8 def _test_no_init(self, cls):
9 with self.assertRaises(bt2.Error):
10 cls()
11
12 def test_no_init_source(self):
13 class MyIter(bt2._UserNotificationIterator):
14 def __next__(self):
15 raise bt2.Stop
16
17 class MySource(bt2._UserSourceComponent,
18 notification_iterator_class=MyIter):
19 pass
20
21 self._test_no_init(MySource)
22
23 def test_no_init_filter(self):
24 class MyIter(bt2._UserNotificationIterator):
25 def __next__(self):
26 raise bt2.Stop
27
28 class MyFilter(bt2._UserFilterComponent,
29 notification_iterator_class=MyIter):
30 pass
31
32 self._test_no_init(MyFilter)
33
34 def test_no_init_sink(self):
35 class MySink(bt2._UserSinkComponent):
36 def _consume(self):
37 pass
38
39 self._test_no_init(MySink)
40
41 def test_incomplete_source_no_notif_iter_cls(self):
42 class MyIter(bt2._UserNotificationIterator):
43 pass
44
45 with self.assertRaises(bt2.IncompleteUserClass):
46 class MySource(bt2._UserSourceComponent):
47 pass
48
49 def test_incomplete_source_wrong_notif_iter_cls_type(self):
50 class MyIter(bt2._UserNotificationIterator):
51 pass
52
53 with self.assertRaises(bt2.IncompleteUserClass):
54 class MySource(bt2._UserSourceComponent,
55 notification_iterator_class=int):
56 pass
57
58 def test_incomplete_filter_no_notif_iter_cls(self):
59 class MyIter(bt2._UserNotificationIterator):
60 pass
61
62 with self.assertRaises(bt2.IncompleteUserClass):
63 class MyFilter(bt2._UserFilterComponent):
64 pass
65
66 def test_incomplete_sink_no_consume_method(self):
67 class MyIter(bt2._UserNotificationIterator):
68 pass
69
70 with self.assertRaises(bt2.IncompleteUserClass):
71 class MySink(bt2._UserSinkComponent):
72 pass
73
74 def test_minimal_source(self):
75 class MyIter(bt2._UserNotificationIterator):
76 pass
77
78 class MySource(bt2._UserSourceComponent,
79 notification_iterator_class=MyIter):
80 pass
81
82 def test_minimal_filter(self):
83 class MyIter(bt2._UserNotificationIterator):
84 pass
85
86 class MyFilter(bt2._UserFilterComponent,
87 notification_iterator_class=MyIter):
88 pass
89
90 def test_minimal_sink(self):
91 class MySink(bt2._UserSinkComponent):
92 def _consume(self):
93 pass
94
95 def test_default_name(self):
96 class MySink(bt2._UserSinkComponent):
97 def _consume(self):
98 pass
99
100 self.assertEqual(MySink.name, 'MySink')
101
102 def test_custom_name(self):
103 class MySink(bt2._UserSinkComponent, name='salut'):
104 def _consume(self):
105 pass
106
107 self.assertEqual(MySink.name, 'salut')
108
109 def test_invalid_custom_name(self):
110 with self.assertRaises(TypeError):
111 class MySink(bt2._UserSinkComponent, name=23):
112 def _consume(self):
113 pass
114
115 def test_description(self):
116 class MySink(bt2._UserSinkComponent):
117 """
118 The description.
119
120 Bacon ipsum dolor amet ribeye t-bone corned beef, beef jerky
121 porchetta burgdoggen prosciutto chicken frankfurter boudin
122 hamburger doner bacon turducken. Sirloin shank sausage,
123 boudin meatloaf alcatra meatball t-bone tongue pastrami
124 cupim flank tenderloin.
125 """
126
127 def _consume(self):
128 pass
129
130 self.assertEqual(MySink.description, 'The description.')
131
132 def test_empty_description(self):
133 class MySink(bt2._UserSinkComponent):
134 """
135 """
136
137 def _consume(self):
138 pass
139
140 self.assertIsNone(MySink.description)
141
142 def test_help(self):
143 class MySink(bt2._UserSinkComponent):
144 """
145 The description.
146
147 The help
148 text is
149 here.
150 """
151
152 def _consume(self):
153 pass
154
155 self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
156
157 def test_addr(self):
158 class MySink(bt2._UserSinkComponent):
159 def _consume(self):
160 pass
161
162 self.assertIsInstance(MySink.addr, int)
163 self.assertNotEqual(MySink.addr, 0)
164
165 def test_query_not_implemented(self):
166 class MySink(bt2._UserSinkComponent):
167 def _consume(self):
168 pass
169
170 with self.assertRaises(bt2.Error):
171 bt2.QueryExecutor().query(MySink, 'obj', 23)
172
173 def test_query_raises(self):
174 class MySink(bt2._UserSinkComponent):
175 def _consume(self):
176 pass
177
178 @classmethod
179 def _query(cls, query_exec, obj, params):
180 raise ValueError
181
182 with self.assertRaises(bt2.Error):
183 bt2.QueryExecutor().query(MySink, 'obj', 23)
184
185 def test_query_wrong_return_type(self):
186 class MySink(bt2._UserSinkComponent):
187 def _consume(self):
188 pass
189
190 @classmethod
191 def _query(cls, query_exec, obj, params):
192 return ...
193
194 with self.assertRaises(bt2.Error):
195 bt2.QueryExecutor().query(MySink, 'obj', 23)
196
197 def test_query_params_none(self):
198 class MySink(bt2._UserSinkComponent):
199 def _consume(self):
200 pass
201
202 @classmethod
203 def _query(cls, query_exec, obj, params):
204 nonlocal query_params
205 query_params = params
206 return None
207
208 query_params = None
209 params = None
210 res = bt2.QueryExecutor().query(MySink, 'obj', params)
211 self.assertEqual(query_params, params)
212 self.assertIsNone(res)
213 del query_params
214
215 def test_query_simple(self):
216 class MySink(bt2._UserSinkComponent):
217 def _consume(self):
218 pass
219
220 @classmethod
221 def _query(cls, query_exec, obj, params):
222 nonlocal query_params
223 query_params = params
224 return 17.5
225
226 query_params = None
227 params = ['coucou', 23, None]
228 res = bt2.QueryExecutor().query(MySink, 'obj', params)
229 self.assertEqual(query_params, params)
230 self.assertEqual(res, 17.5)
231 del query_params
232
233 def test_query_complex(self):
234 class MySink(bt2._UserSinkComponent):
235 def _consume(self):
236 pass
237
238 @classmethod
239 def _query(cls, query_exec, obj, params):
240 nonlocal query_params
241 query_params = params
242 return {
243 'null': None,
244 'bt2': 'BT2',
245 }
246
247 query_params = None
248 params = {
249 'array': ['coucou', 23, None],
250 'other_map': {
251 'yes': 'yeah',
252 '19': 19,
253 'minus 1.5': -1.5,
254 },
255 'null': None,
256 }
257
258 res = bt2.QueryExecutor().query(MySink, 'obj', params)
259 self.assertEqual(query_params, params)
260 self.assertEqual(res, {
261 'null': None,
262 'bt2': 'BT2',
263 })
264 del query_params
265
266 def test_eq(self):
267 class MySink(bt2._UserSinkComponent):
268 def _consume(self):
269 pass
270
271 self.assertEqual(MySink, MySink)
272
273
274 class GenericComponentClassTestCase(unittest.TestCase):
275 def setUp(self):
276 class MySink(bt2._UserSinkComponent):
277 '''
278 The description.
279
280 The help.
281 '''
282 def _consume(self):
283 pass
284
285 @classmethod
286 def _query(cls, query_exec, obj, params):
287 return [obj, params, 23]
288
289 self._py_comp_cls = MySink
290 graph = bt2.Graph()
291 comp = graph.add_component(MySink, 'salut')
292 self._comp_cls = comp.component_class
293 self.assertTrue(issubclass(type(self._comp_cls),
294 bt2.component._GenericComponentClass))
295
296 def tearDown(self):
297 del self._py_comp_cls
298 del self._comp_cls
299
300 def test_description(self):
301 self.assertEqual(self._comp_cls.description, 'The description.')
302
303 def test_help(self):
304 self.assertEqual(self._comp_cls.help, 'The help.')
305
306 def test_name(self):
307 self.assertEqual(self._comp_cls.name, 'MySink')
308
309 def test_addr(self):
310 self.assertIsInstance(self._comp_cls.addr, int)
311 self.assertNotEqual(self._comp_cls.addr, 0)
312
313 def test_eq_invalid(self):
314 self.assertFalse(self._comp_cls == 23)
315
316 def test_eq(self):
317 self.assertEqual(self._comp_cls, self._comp_cls)
318 self.assertEqual(self._py_comp_cls, self._comp_cls)
319
320 def test_query(self):
321 res = bt2.QueryExecutor().query(self._comp_cls, 'an object',
322 {'yes': 'no', 'book': -17})
323 expected = ['an object', {'yes': 'no', 'book': -17}, 23]
324 self.assertEqual(res, expected)
This page took 0.038482 seconds and 4 git commands to generate.