cls, comp_cls_name, comp_cls_descr, comp_cls_help
)
elif _UserSinkComponent in bases:
- if not hasattr(cls, '_consume'):
+ if not hasattr(cls, '_user_consume'):
raise bt2._IncompleteUserClass(
- "cannot create component class '{}': missing a _consume() method".format(
+ "cannot create component class '{}': missing a _user_consume() method".format(
class_name
)
)
)
# this can raise, but the native side checks the exception
- results = cls._query(query_exec, obj, params, log_level)
+ results = cls._user_query(query_exec, obj, params, log_level)
# this can raise, but the native side checks the exception
results = bt2.create_value(results)
return int(results_ptr)
- def _query(cls, query_executor, obj, params, log_level):
+ def _user_query(cls, query_executor, obj, params, log_level):
raise NotImplementedError
def _bt_component_class_ptr(self):
def __init__(self, params=None):
pass
- def _finalize(self):
+ def _user_finalize(self):
pass
- def _port_connected(self, port, other_port):
+ def _user_port_connected(self, port, other_port):
pass
def _bt_port_connected_from_native(
other_port = bt2_port._create_from_ptr_and_get_ref(
other_port_ptr, other_port_type
)
- self._port_connected(port, other_port)
+ self._user_port_connected(port, other_port)
def _create_trace_class(self, assigns_automatic_stream_class_id=True):
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
)
def _bt_graph_is_configured_from_native(self):
- self._graph_is_configured()
+ self._user_graph_is_configured()
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
pass
@property
def _is_interrupted(self):
return bool(native_bt.self_message_iterator_is_interrupted(self._bt_ptr))
- def _finalize(self):
+ def _user_finalize(self):
pass
def __next__(self):
def _bt_can_seek_beginning_from_native(self):
# Here, we mimic the behavior of the C API:
#
- # - If the iterator has a _can_seek_beginning attribute, read it and use
- # that result.
+ # - If the iterator has a _user_can_seek_beginning attribute,
+ # read it and use that result.
# - Otherwise, the presence or absence of a `_seek_beginning`
# method indicates whether the iterator can seek beginning.
- if hasattr(self, '_can_seek_beginning'):
- can_seek_beginning = self._can_seek_beginning
+ if hasattr(self, '_user_can_seek_beginning'):
+ can_seek_beginning = self._user_can_seek_beginning
utils._check_bool(can_seek_beginning)
return can_seek_beginning
else:
- return hasattr(self, '_seek_beginning')
+ return hasattr(self, '_user_seek_beginning')
def _bt_seek_beginning_from_native(self):
- self._seek_beginning()
+ self._user_seek_beginning()
def _create_input_port_message_iterator(self, input_port):
utils._check_type(input_port, bt2_port._UserComponentInputPort)
PyObject *py_comp = bt_self_component_get_data(self_component);
BT_ASSERT(py_comp);
- /* Call user's _finalize() method */
+ /* Call user's _user_finalize() method */
PyObject *py_method_result = PyObject_CallMethod(py_comp,
- "_finalize", NULL);
+ "_user_finalize", NULL);
if (PyErr_Occurred()) {
- BT_LOGW("User component's _finalize() method raised an exception: ignoring:");
+ BT_LOGW("User component's _user_finalize() method raised an exception: ignoring:");
logw_exception();
}
/*
- * Ignore any exception raised by the _finalize() method because
- * it won't change anything at this point: the component is
- * being destroyed anyway.
+ * Ignore any exception raised by the _user_finalize() method
+ * because it won't change anything at this point: the component
+ * is being destroyed anyway.
*/
PyErr_Clear();
Py_XDECREF(py_method_result);
BT_ASSERT(py_message_iter);
- /* Call user's _finalize() method */
+ /* Call user's _user_finalize() method */
py_method_result = PyObject_CallMethod(py_message_iter,
- "_finalize", NULL);
+ "_user_finalize", NULL);
if (PyErr_Occurred()) {
- BT_LOGW("User's _finalize() method raised an exception: ignoring:");
+ BT_LOGW("User's _user_finalize() method raised an exception: ignoring:");
logw_exception();
}
/*
- * Ignore any exception raised by the _finalize() method because
- * it won't change anything at this point: the component is
- * being destroyed anyway.
+ * Ignore any exception raised by the _user_finalize() method
+ * because it won't change anything at this point: the component
+ * is being destroyed anyway.
*/
PyErr_Clear();
Py_XDECREF(py_method_result);
BT_ASSERT(py_comp);
py_method_result = PyObject_CallMethod(py_comp,
- "_consume", NULL);
+ "_user_consume", NULL);
status = py_exc_to_status_component(self_component);
if (!py_method_result && status == __BT_FUNC_STATUS_OK) {
/* Pretty sure this should never happen, but just in case */
- BT_LOGE("User's _consume() method failed without raising an exception: "
+ BT_LOGE("User's _user_consume() method failed without raising an exception: "
"status=%d", status);
status = __BT_FUNC_STATUS_ERROR;
}
def __init__(comp_self, params):
self.assertEqual(comp_self.name, 'yaes')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink, 'yaes')
def __init__(comp_self, params):
self.assertEqual(comp_self.logging_level, bt2.LoggingLevel.INFO)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
def __init__(comp_self, params):
self.assertEqual(comp_self.cls, MySink)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
self.assertIsInstance(comp_self.addr, int)
self.assertNotEqual(comp_self.addr, 0)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
finalized = False
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
- def _finalize(comp_self):
+ def _user_finalize(comp_self):
nonlocal finalized
finalized = True
def test_name(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink, 'yaes')
def test_logging_level(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.WARNING)
def test_class(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
def test_addr(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
def test_no_init_sink(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._test_no_init(MySink)
def test_minimal_sink(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
def test_default_name(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertEqual(MySink.name, 'MySink')
def test_custom_name(self):
class MySink(bt2._UserSinkComponent, name='salut'):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertEqual(MySink.name, 'salut')
with self.assertRaises(TypeError):
class MySink(bt2._UserSinkComponent, name=23):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
def test_description(self):
cupim flank tenderloin.
"""
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertEqual(MySink.description, 'The description.')
"""
"""
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertIsNone(MySink.description)
here.
"""
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
def test_addr(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertIsInstance(MySink.addr, int)
def test_query_not_implemented(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
with self.assertRaises(bt2._Error):
def test_query_raises(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
raise ValueError
with self.assertRaises(bt2._Error):
def test_query_wrong_return_type(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
return ...
with self.assertRaises(bt2._Error):
def test_query_params_none(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
return None
def test_query_logging_level(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
query_log_level = log_level
def test_query_returns_none(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@staticmethod
- def _query(query_exec, obj, params, log_level):
+ def _user_query(query_exec, obj, params, log_level):
return
res = bt2.QueryExecutor().query(MySink, 'obj', None)
def test_query_simple(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
return 17.5
def test_query_complex(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
return {'null': None, 'bt2': 'BT2'}
def test_eq(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self.assertEqual(MySink, MySink)
The help.
'''
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
return [obj, params, 23]
self._py_comp_cls = MySink
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
def __init__(self, params):
self._in = self._add_input_port('in')
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._iter = self._create_input_port_message_iterator(self._in)
- def _consume(self):
+ def _user_consume(self):
next(self._iter)
def __init__(self, params):
self._in = self._add_input_port('in')
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._iter = self._create_input_port_message_iterator(self._in)
- def _consume(self):
+ def _user_consume(self):
try:
next(self._iter)
except bt2._Error as e:
class SinkWithFailingQuery(bt2._UserSinkComponent):
- def _graph_is_configured(self):
- pass
-
- def _consume(self):
+ def _user_consume(self):
pass
@staticmethod
- def _query(executor, obj, params, log_level):
+ def _user_query(executor, obj, params, log_level):
raise ValueError('Query is failing')
def test_add_component_user_cls(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._graph.add_component(MySink, 'salut')
def test_add_component_gen_cls(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._graph.add_component(MySink, 'salut')
nonlocal comp_params
comp_params = params
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
params = {'hello': 23, 'path': '/path/to/stuff'}
def test_add_component_invalid_logging_level_type(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
with self.assertRaises(TypeError):
def test_add_component_invalid_logging_level_value(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
with self.assertRaises(ValueError):
def test_add_component_logging_level(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._graph.add_component(
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
next(self._msg_iter)
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._msg_iter = self._create_input_port_message_iterator(
self._input_ports['in']
)
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
# Pretend that somebody asynchronously interrupted the graph.
nonlocal graph
graph.interrupt()
return next(self._msg_iter)
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._msg_iter = self._create_input_port_message_iterator(
self._input_ports['in']
)
self._input_port = self._add_input_port('in')
self._at = 0
- def _consume(comp_self):
+ def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
comp_self._at += 1
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._msg_iter = self._create_input_port_message_iterator(
self._input_port
)
self._input_port = self._add_input_port('in')
self._at = 0
- def _consume(comp_self):
+ def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
self.assertIsInstance(msg, bt2._StreamBeginningMessage)
comp_self._at += 1
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._msg_iter = self._create_input_port_message_iterator(
self._input_port
)
self._input_port = self._add_input_port('in')
self._at = 0
- def _consume(comp_self):
+ def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
self.assertIsInstance(msg, bt2._StreamBeginningMessage)
comp_self._at += 1
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._msg_iter = self._create_input_port_message_iterator(
self._input_port
)
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
- def _port_connected(self, port, other_port):
+ def _user_port_connected(self, port, other_port):
self._add_input_port('taste')
def port_added_listener(component, port):
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
- def _port_connected(self, port, other_port):
+ def _user_port_connected(self, port, other_port):
self._add_input_port('taste')
with self.assertRaises(TypeError):
def __init__(self, params):
raise ValueError('oops!')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
graph = bt2.Graph()
with self.assertRaises(bt2._Error):
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
def port_added_listener(component, port):
raise ValueError('oh noes!')
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
raise bt2.Stop
- def _graph_is_configured(self):
- pass
-
def ports_connected_listener(
upstream_component, upstream_port, downstream_component, downstream_port
):
def __init__(self, params):
self._add_input_port('in')
- def _consume(self):
+ def _user_consume(self):
next(self._msg_iter)
- def _graph_is_configured(self):
+ def _user_graph_is_configured(self):
self._msg_iter = self._create_input_port_message_iterator(
self._input_ports['in']
)
def test_finalize(self):
class MyIter(bt2._UserMessageIterator):
- def _finalize(self):
+ def _user_finalize(self):
nonlocal finalized
finalized = True
]
self._at = 0
- def _seek_beginning(self):
+ def _user_seek_beginning(self):
self._at = 0
def __next__(self):
def __next__(self):
return next(self._upstream_iter)
- def _seek_beginning(self):
+ def _user_seek_beginning(self):
self._upstream_iter.seek_beginning()
@property
- def _can_seek_beginning(self):
+ def _user_can_seek_beginning(self):
return self._upstream_iter.can_seek_beginning
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
def test_can_seek_beginning(self):
it, MySourceIter = self._setup_seek_beginning_test()
- def _can_seek_beginning(self):
+ def _user_can_seek_beginning(self):
nonlocal can_seek_beginning
return can_seek_beginning
- MySourceIter._can_seek_beginning = property(_can_seek_beginning)
+ MySourceIter._user_can_seek_beginning = property(_user_can_seek_beginning)
can_seek_beginning = True
self.assertTrue(it.can_seek_beginning)
# Remove the _can_seek_beginning method, we now rely on the presence of
# a _seek_beginning method to know whether the iterator can seek to
# beginning or not.
- del MySourceIter._can_seek_beginning
+ del MySourceIter._user_can_seek_beginning
self.assertTrue(it.can_seek_beginning)
- del MySourceIter._seek_beginning
+ del MySourceIter._user_seek_beginning
self.assertFalse(it.can_seek_beginning)
def test_seek_beginning(self):
def test_seek_beginning_user_error(self):
it, MySourceIter = self._setup_seek_beginning_test()
- def _seek_beginning_error(self):
+ def _user_seek_beginning_error(self):
raise ValueError('ouch')
- MySourceIter._seek_beginning = _seek_beginning_error
+ MySourceIter._user_seek_beginning = _user_seek_beginning_error
with self.assertRaises(bt2._Error):
it.seek_beginning()
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
self.assertEqual(port2.addr, comp_self._input_ports['print'].addr)
self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
with self.assertRaises(KeyError):
comp_self._input_ports['hello']
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
comp_self._add_input_port('insert')
self.assertEqual(len(comp_self._input_ports), 3)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
self.assertEqual(ports[2][0], 'insert')
self.assertEqual(ports[2][1].addr, port3.addr)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
port2 = comp_self._add_input_port('print')
port3 = comp_self._add_input_port('insert')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
with self.assertRaises(KeyError):
comp_self._input_ports['hello']
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
port2 = comp_self._add_input_port('print')
port3 = comp_self._add_input_port('insert')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
def __init__(comp_self, params):
comp_self._add_input_port('clear')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
def __init__(comp_self, params):
comp_self._add_input_port('clear')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
def __init__(comp_self, params):
comp_self._add_input_port('clear')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
comp = self._create_comp(MySink)
port = comp_self._add_input_port('clear')
self.assertEqual(port.name, 'clear')
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
port = comp_self._add_input_port('clear')
self.assertIsNone(port.connection)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
port = comp_self._add_input_port('clear')
self.assertFalse(port.is_connected)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
self._create_comp(MySink)
class QueryExecutorTestCase(unittest.TestCase):
def test_query(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
return {'null': None, 'bt2': 'BT2'}
def test_query_params_none(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_params
query_params = params
def test_query_logging_level(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
query_log_level = log_level
def test_query_gen_error(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
raise ValueError
with self.assertRaises(bt2._Error) as ctx:
def test_query_invalid_object(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidObject
with self.assertRaises(bt2.InvalidObject):
def test_query_logging_level_invalid_type(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
pass
with self.assertRaises(TypeError):
def test_query_logging_level_invalid_value(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
pass
with self.assertRaises(ValueError):
def test_query_try_again(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
with self.assertRaises(bt2.TryAgain):
def test_query_add_interrupter(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
nonlocal interrupter2
test_self.assertFalse(query_exec.is_interrupted)
interrupter2.set()
def test_query_interrupt(self):
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
@classmethod
- def _query(cls, query_exec, obj, params, log_level):
+ def _user_query(cls, query_exec, obj, params, log_level):
test_self.assertFalse(query_exec.is_interrupted)
query_exec.interrupt()
test_self.assertTrue(query_exec.is_interrupted)
nonlocal res_bound
res_bound = func(self)
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
g = bt2.Graph()
self._print_params(params)
@staticmethod
- def _query(query_exec, obj, params, log_level):
+ def _user_query(query_exec, obj, params, log_level):
if obj == 'support-info':
if params['type'] == 'file':
name = os.path.basename(str(params['input']))
self._print_params(params)
@staticmethod
- def _query(query_exec, obj, params, log_level):
+ def _user_query(query_exec, obj, params, log_level):
if obj == 'support-info':
if params['type'] == 'directory':
name = os.path.basename(str(params['input']))
self._print_params(params)
@staticmethod
- def _query(query_exec, obj, params, log_level):
+ def _user_query(query_exec, obj, params, log_level):
if obj == 'support-info':
return (
1.0
]
self._at = 0
- def _seek_beginning(self):
+ def _user_seek_beginning(self):
self._at = 0
def __next__(self):
@bt2.plugin_component_class
class MySink(bt2._UserSinkComponent):
- def _consume(self):
- pass
-
- def _graph_is_configured(self):
+ def _user_consume(self):
pass
projects / babeltrace.git / commitdiff
