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.
25 class QueryExecutorTestCase(unittest
.TestCase
):
27 class MySink(bt2
._UserSinkComponent
):
28 def _user_consume(self
):
32 def _user_query(cls
, priv_query_exec
, obj
, params
):
35 return {'null': None, 'bt2': 'BT2'}
39 'array': ['coucou', 23, None],
40 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
44 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
45 self
.assertEqual(query_params
, params
)
46 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
49 def test_query_params_none(self
):
50 class MySink(bt2
._UserSinkComponent
):
51 def _user_consume(self
):
55 def _user_query(cls
, priv_query_exec
, obj
, params
):
60 res
= bt2
.QueryExecutor(MySink
, 'obj', None).query()
61 self
.assertIs(query_params
, None)
64 def test_query_no_params(self
):
65 class MySink(bt2
._UserSinkComponent
):
66 def _user_consume(self
):
70 def _user_query(cls
, priv_query_exec
, obj
, params
):
75 res
= bt2
.QueryExecutor(MySink
, 'obj').query()
76 self
.assertIs(query_params
, None)
79 def test_query_logging_level(self
):
80 class MySink(bt2
._UserSinkComponent
):
81 def _user_consume(self
):
85 def _user_query(cls
, priv_query_exec
, obj
, params
):
86 nonlocal query_log_level
87 query_log_level
= priv_query_exec
.logging_level
89 query_log_level
= None
90 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', None)
91 query_exec
.logging_level
= bt2
.LoggingLevel
.INFO
93 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
96 def test_query_gen_error(self
):
97 class MySink(bt2
._UserSinkComponent
):
98 def _user_consume(self
):
102 def _user_query(cls
, priv_query_exec
, obj
, params
):
105 with self
.assertRaises(bt2
._Error
) as ctx
:
106 res
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
109 self
.assertEqual(len(exc
), 2)
111 self
.assertIsInstance(cause
, bt2
._ComponentClassErrorCause
)
112 self
.assertIn('raise ValueError', cause
.message
)
113 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
114 self
.assertEqual(cause
.component_class_name
, 'MySink')
116 def test_query_unknown_object(self
):
117 class MySink(bt2
._UserSinkComponent
):
118 def _user_consume(self
):
122 def _user_query(cls
, priv_query_exec
, obj
, params
):
123 raise bt2
.UnknownObject
125 with self
.assertRaises(bt2
.UnknownObject
):
126 res
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
128 def test_query_logging_level_invalid_type(self
):
129 class MySink(bt2
._UserSinkComponent
):
130 def _user_consume(self
):
134 def _user_query(cls
, priv_query_exec
, obj
, params
):
137 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
139 with self
.assertRaises(TypeError):
140 query_exec
.logging_level
= 'yeah'
142 def test_query_logging_level_invalid_value(self
):
143 class MySink(bt2
._UserSinkComponent
):
144 def _user_consume(self
):
148 def _user_query(cls
, priv_query_exec
, obj
, params
):
151 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
153 with self
.assertRaises(ValueError):
154 query_exec
.logging_level
= 12345
156 def test_query_try_again(self
):
157 class MySink(bt2
._UserSinkComponent
):
158 def _user_consume(self
):
162 def _user_query(cls
, priv_query_exec
, obj
, params
):
165 with self
.assertRaises(bt2
.TryAgain
):
166 res
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
168 def test_query_add_interrupter(self
):
169 class MySink(bt2
._UserSinkComponent
):
170 def _user_consume(self
):
174 def _user_query(cls
, priv_query_exec
, obj
, params
):
175 nonlocal interrupter2
176 test_self
.assertFalse(query_exec
.is_interrupted
)
178 test_self
.assertTrue(query_exec
.is_interrupted
)
180 test_self
.assertFalse(query_exec
.is_interrupted
)
182 interrupter1
= bt2
.Interrupter()
183 interrupter2
= bt2
.Interrupter()
185 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
186 query_exec
.add_interrupter(interrupter1
)
187 query_exec
.add_interrupter(interrupter2
)
190 def test_query_interrupt(self
):
191 class MySink(bt2
._UserSinkComponent
):
192 def _user_consume(self
):
196 def _user_query(cls
, priv_query_exec
, obj
, params
):
197 test_self
.assertFalse(query_exec
.is_interrupted
)
198 query_exec
.interrupt()
199 test_self
.assertTrue(query_exec
.is_interrupted
)
202 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
205 def test_query_priv_executor_invalid_after(self
):
206 class MySink(bt2
._UserSinkComponent
):
207 def _user_consume(self
):
211 def _user_query(cls
, priv_query_exec
, obj
, params
):
212 nonlocal test_priv_query_exec
213 test_priv_query_exec
= priv_query_exec
215 test_priv_query_exec
= None
216 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
218 assert test_priv_query_exec
is not None
220 with self
.assertRaises(RuntimeError):
221 test_priv_query_exec
.logging_level
223 del test_priv_query_exec