comp_cls_descr,
comp_cls_help)
elif _UserSinkComponent in bases:
+ if not hasattr(cls, '_graph_is_configured'):
+ raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _graph_is_configured() method".format(class_name))
+
if not hasattr(cls, '_consume'):
raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _consume() method".format(class_name))
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes')
def test_logging_level(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
def test_class(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_addr(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_finalize(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
def _finalize(comp_self):
nonlocal finalized
finalized = True
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes')
self.assertEqual(comp.name, 'yaes')
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.WARNING)
self.assertEqual(comp.logging_level, bt2.LoggingLevel.WARNING)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(comp.cls, MySink)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertIsInstance(comp.addr, int)
self.assertNotEqual(comp.addr, 0)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._test_no_init(MySink)
def test_incomplete_source_no_msg_iter_cls(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
def test_default_name(self):
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.name, 'MySink')
def test_custom_name(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.name, 'salut')
def test_invalid_custom_name(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
def test_description(self):
class MySink(bt2._UserSinkComponent):
"""
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.description, 'The description.')
def test_empty_description(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertIsNone(MySink.description)
def test_help(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
def test_addr(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertIsInstance(MySink.addr, int)
self.assertNotEqual(MySink.addr, 0)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
with self.assertRaises(bt2.Error):
bt2.QueryExecutor().query(MySink, 'obj', 23)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise ValueError
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
return ...
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@staticmethod
def _query(query_exec, obj, params, log_level):
return
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink, MySink)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
return [obj, params, 23]
def _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 _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 _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')
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!')
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(len(comp.input_ports), 1)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_user_src_output_ports_getitem_invalid_key(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_user_src_output_ports_len(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
self.assertEqual(port2.addr, comp.input_ports['print'].addr)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
with self.assertRaises(KeyError):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(len(comp.input_ports), 3)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
ports = []
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(comp.input_ports['clear'].name, 'clear')
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertIsNone(comp.input_ports['clear'].connection)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertFalse(comp.input_ports['clear'].is_connected)
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_self_connection_none(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_self_is_connected_false(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_source_self_port_user_data(self):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise ValueError
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidObject
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
pass
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
pass
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidParams
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
g = bt2.Graph()
res_bound = None
g.add_component(MySink, 'comp')
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
+
bt2.register_plugin(__name__, 'sparkling', author='Philippe Proulx',
description='A delicious plugin.',