import bt2
-@unittest.skip("this is broken")
class QueryExecutorTestCase(unittest.TestCase):
def test_query(self):
class MySink(bt2._UserSinkComponent):
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
return {
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
raise ValueError
with self.assertRaises(bt2.Error):
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidQueryObject
with self.assertRaises(bt2.InvalidQueryObject):
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidQueryParams
with self.assertRaises(bt2.InvalidQueryParams):
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
with self.assertRaises(bt2.TryAgain):
pass
@classmethod
- def _query(cls, query_exec, obj, params):
+ def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
query_exec = bt2.QueryExecutor()
with self.assertRaises(bt2.QueryExecutorCanceled):
res = query_exec.query(MySink, 'obj', [17, 23])
-
- def test_eq(self):
- query_exec = bt2.QueryExecutor()
- self.assertEqual(query_exec, query_exec)
-
- def test_eq_invalid(self):
- query_exec = bt2.QueryExecutor()
- self.assertNotEqual(query_exec, 23)