7 @unittest.skip("this is broken")
8 class QueryExecutorTestCase(unittest
.TestCase
):
10 class MySink(bt2
._UserSinkComponent
):
15 def _query(cls
, query_exec
, obj
, params
):
25 'array': ['coucou', 23, None],
34 res
= bt2
.QueryExecutor().query(MySink
, 'obj', params
)
35 self
.assertEqual(query_params
, params
)
36 self
.assertEqual(res
, {
42 def test_query_params_none(self
):
43 class MySink(bt2
._UserSinkComponent
):
48 def _query(cls
, query_exec
, obj
, params
):
53 res
= bt2
.QueryExecutor().query(MySink
, 'obj', None)
54 self
.assertEqual(query_params
, None)
57 def test_query_gen_error(self
):
58 class MySink(bt2
._UserSinkComponent
):
63 def _query(cls
, query_exec
, obj
, params
):
66 with self
.assertRaises(bt2
.Error
):
67 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
69 def test_query_invalid_object(self
):
70 class MySink(bt2
._UserSinkComponent
):
75 def _query(cls
, query_exec
, obj
, params
):
76 raise bt2
.InvalidQueryObject
78 with self
.assertRaises(bt2
.InvalidQueryObject
):
79 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
81 def test_query_invalid_params(self
):
82 class MySink(bt2
._UserSinkComponent
):
87 def _query(cls
, query_exec
, obj
, params
):
88 raise bt2
.InvalidQueryParams
90 with self
.assertRaises(bt2
.InvalidQueryParams
):
91 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
93 def test_query_try_again(self
):
94 class MySink(bt2
._UserSinkComponent
):
99 def _query(cls
, query_exec
, obj
, params
):
102 with self
.assertRaises(bt2
.TryAgain
):
103 res
= bt2
.QueryExecutor().query(MySink
, 'obj', [17, 23])
105 def test_cancel_no_query(self
):
106 query_exec
= bt2
.QueryExecutor()
107 self
.assertFalse(query_exec
.is_canceled
)
109 self
.assertTrue(query_exec
.is_canceled
)
111 def test_query_canceled(self
):
112 class MySink(bt2
._UserSinkComponent
):
117 def _query(cls
, query_exec
, obj
, params
):
120 query_exec
= bt2
.QueryExecutor()
123 with self
.assertRaises(bt2
.QueryExecutorCanceled
):
124 res
= query_exec
.query(MySink
, 'obj', [17, 23])
127 query_exec
= bt2
.QueryExecutor()
128 self
.assertEqual(query_exec
, query_exec
)
130 def test_eq_invalid(self
):
131 query_exec
= bt2
.QueryExecutor()
132 self
.assertNotEqual(query_exec
, 23)
This page took 0.044182 seconds and 4 git commands to generate.