class SourceWithFailingIter(
bt2._UserSourceComponent, message_iterator_class=FailingIter
):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class SourceWithFailingInit(
bt2._UserSourceComponent, message_iterator_class=FailingIter
):
- def __init__(self, params):
+ def __init__(self, params, obj):
raise ValueError('Source is failing')
class WorkingSink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._in = self._add_input_port('in')
def _user_graph_is_configured(self):
class SinkWithExceptionChaining(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._in = self._add_input_port('in')
def _user_graph_is_configured(self):
pass
@staticmethod
- def _user_query(executor, obj, params, log_level):
+ def _user_query(priv_executor, obj, params, method_obj):
raise ValueError('Query is failing')
def test_current_thread_error_none(self):
# When a bt2._Error is raised, it steals the current thread's error.
# Verify that it is now NULL.
- exc = self._run_failing_graph(SourceWithFailingInit, WorkingSink)
+ self._run_failing_graph(SourceWithFailingInit, WorkingSink)
self.assertIsNone(native_bt.current_thread_take_error())
def test_len(self):
self.assertIsNone(cause.plugin_name)
def test_component_class_error_cause(self):
- q = bt2.QueryExecutor()
+ q = bt2.QueryExecutor(SinkWithFailingQuery, 'hello')
with self.assertRaises(bt2._Error) as ctx:
- q.query(SinkWithFailingQuery, 'hello')
+ q.query()
cause = ctx.exception[0]
self.assertIs(type(cause), bt2._ComponentClassErrorCause)
self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SOURCE)
self.assertEqual(cause.component_class_name, 'SourceWithFailingIter')
self.assertIsNone(cause.plugin_name)
+
+
+if __name__ == '__main__':
+ unittest.main()