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