class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut')
self.assertEqual(comp.name, 'salut')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut')
assert comp
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
params = {'hello': 23, 'path': '/path/to/stuff'}
comp = self._graph.add_component(MySink, 'salut', params)
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
with self.assertRaises(TypeError):
self._graph.add_component(MySink, 'salut', logging_level='yo')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
with self.assertRaises(ValueError):
self._graph.add_component(MySink, 'salut', logging_level=12345)
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut',
logging_level=bt2.LoggingLevel.DEBUG)
def _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 _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 _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def _port_connected(self, port, other_port):
self._add_input_port('taste')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def _port_connected(self, port, other_port):
self._add_input_port('taste')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
graph = bt2.Graph()
with self.assertRaises(bt2.Error):
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def port_added_listener(component, port):
raise ValueError('oh noes!')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def ports_connected_listener(upstream_component, upstream_port,
downstream_component, downstream_port):
raise ValueError('oh noes!')