bc20a719b3dca0000ee9a1fa123ed51041ddc0ef
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
):
31 def _graph_is_configured(self
):
35 def _query(cls
, query_exec
, obj
, params
, log_level
):
38 return {'null': None, 'bt2': 'BT2'}
42 'array': ['coucou', 23, None],
43 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
47 res
= bt2
.QueryExecutor().query(MySink
, 'obj', params
)
48 self
.assertEqual(query_params
, params
)
49 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
52 def test_query_params_none(self
):
53 class MySink(bt2
._UserSinkComponent
):
57 def _graph_is_configured(self
):
61 def _query(cls
, query_exec
, obj
, params
, log_level
):
66 res
= bt2
.QueryExecutor().query(MySink
, 'obj', None)
67 self
.assertEqual(query_params
, None)
70 def test_query_logging_level(self
):
71 class MySink(bt2
._UserSinkComponent
):
75 def _graph_is_configured(self
):
79 def _query(cls
, query_exec
, obj
, params
, log_level
):
80 nonlocal query_log_level
81 query_log_level
= log_level
83 query_log_level
= None
84 res
= bt2
.QueryExecutor().query(MySink
, 'obj', None, bt2
.LoggingLevel
.INFO
)
85 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
88 def test_query_gen_error(self
):
89 class MySink(bt2
._UserSinkComponent
):
93 def _graph_is_configured(self
):
97 def _query(cls
, query_exec
, obj
, params
, log_level
):
100 with self
.assertRaises(bt2
._Error
) as ctx
:
101 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
104 self
.assertEqual(len(exc
), 1)
106 self
.assertIsInstance(cause
, bt2
.error
._ComponentClassErrorCause
)
107 self
.assertIn('raise ValueError', cause
.message
)
108 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
109 self
.assertEqual(cause
.component_class_name
, 'MySink')
111 def test_query_invalid_object(self
):
112 class MySink(bt2
._UserSinkComponent
):
116 def _graph_is_configured(self
):
120 def _query(cls
, query_exec
, obj
, params
, log_level
):
121 raise bt2
.InvalidObject
123 with self
.assertRaises(bt2
.InvalidObject
):
124 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
126 def test_query_logging_level_invalid_type(self
):
127 class MySink(bt2
._UserSinkComponent
):
131 def _graph_is_configured(self
):
135 def _query(cls
, query_exec
, obj
, params
, log_level
):
138 with self
.assertRaises(TypeError):
139 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23], 'yeah')
141 def test_query_logging_level_invalid_value(self
):
142 class MySink(bt2
._UserSinkComponent
):
146 def _graph_is_configured(self
):
150 def _query(cls
, query_exec
, obj
, params
, log_level
):
153 with self
.assertRaises(ValueError):
154 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23], 12345)
156 def test_query_invalid_params(self
):
157 class MySink(bt2
._UserSinkComponent
):
161 def _graph_is_configured(self
):
165 def _query(cls
, query_exec
, obj
, params
, log_level
):
166 raise bt2
.InvalidParams
168 with self
.assertRaises(bt2
.InvalidParams
):
169 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
171 def test_query_try_again(self
):
172 class MySink(bt2
._UserSinkComponent
):
176 def _graph_is_configured(self
):
180 def _query(cls
, query_exec
, obj
, params
, log_level
):
183 with self
.assertRaises(bt2
.TryAgain
):
184 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
186 def test_query_add_interrupter(self
):
187 class MySink(bt2
._UserSinkComponent
):
191 def _graph_is_configured(self
):
195 def _query(cls
, query_exec
, obj
, params
, log_level
):
196 nonlocal interrupter2
197 test_self
.assertFalse(query_exec
.is_interrupted
)
199 test_self
.assertTrue(query_exec
.is_interrupted
)
201 test_self
.assertFalse(query_exec
.is_interrupted
)
203 interrupter1
= bt2
.Interrupter()
204 interrupter2
= bt2
.Interrupter()
206 query_exec
= bt2
.QueryExecutor()
207 query_exec
.add_interrupter(interrupter1
)
208 query_exec
.add_interrupter(interrupter2
)
209 query_exec
.query(MySink
, 'obj', [17, 23])
211 def test_query_interrupt(self
):
212 class MySink(bt2
._UserSinkComponent
):
216 def _graph_is_configured(self
):
220 def _query(cls
, query_exec
, obj
, params
, log_level
):
221 test_self
.assertFalse(query_exec
.is_interrupted
)
222 query_exec
.interrupt()
223 test_self
.assertTrue(query_exec
.is_interrupted
)
226 query_exec
= bt2
.QueryExecutor()
227 query_exec
.query(MySink
, 'obj', [17, 23])
This page took 0.059758 seconds and 4 git commands to generate.