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')
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)