Since a _graph_is_configured method is an essential method to implement
in any useful sink, check for the presence of that method when
instantiating a user sink component class in Python (that is, when the
Python class is created). We already do that for the _consume method,
and I think it greatly helps the user writing a sink by telling them
what not to forget.
Change-Id: Ic8c3741b121eccc2857afac809521dd7213aa679
Signed-off-by: Simon Marchi <simon.marchi@efficios.com>
Reviewed-on: https://review.lttng.org/c/babeltrace/+/1707
Tested-by: jenkins <jenkins@lttng.org>
Reviewed-by: Philippe Proulx <eeppeliteloop@gmail.com>
comp_cls_descr,
comp_cls_help)
elif _UserSinkComponent in bases:
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))
if not hasattr(cls, '_consume'):
raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _consume() method".format(class_name))
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes')
def test_logging_level(self):
comp = self._create_comp(MySink, 'yaes')
def test_logging_level(self):
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
def test_class(self):
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
def test_class(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_addr(self):
self._create_comp(MySink)
def test_addr(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_finalize(self):
self._create_comp(MySink)
def test_finalize(self):
+ def _graph_is_configured(self):
+ pass
+
def _finalize(comp_self):
nonlocal finalized
finalized = True
def _finalize(comp_self):
nonlocal finalized
finalized = True
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes')
self.assertEqual(comp.name, 'yaes')
comp = self._create_comp(MySink, 'yaes')
self.assertEqual(comp.name, 'yaes')
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.WARNING)
self.assertEqual(comp.logging_level, bt2.LoggingLevel.WARNING)
comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.WARNING)
self.assertEqual(comp.logging_level, bt2.LoggingLevel.WARNING)
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(comp.cls, MySink)
comp = self._create_comp(MySink)
self.assertEqual(comp.cls, MySink)
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertIsInstance(comp.addr, int)
self.assertNotEqual(comp.addr, 0)
comp = self._create_comp(MySink)
self.assertIsInstance(comp.addr, int)
self.assertNotEqual(comp.addr, 0)
+ def _graph_is_configured(self):
+ pass
+
self._test_no_init(MySink)
def test_incomplete_source_no_msg_iter_cls(self):
self._test_no_init(MySink)
def test_incomplete_source_no_msg_iter_cls(self):
+ def _graph_is_configured(self):
+ pass
+
def test_default_name(self):
class MySink(bt2._UserSinkComponent):
def _consume(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):
self.assertEqual(MySink.name, 'MySink')
def test_custom_name(self):
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.name, 'salut')
def test_invalid_custom_name(self):
self.assertEqual(MySink.name, 'salut')
def test_invalid_custom_name(self):
+ def _graph_is_configured(self):
+ pass
+
def test_description(self):
class MySink(bt2._UserSinkComponent):
"""
def test_description(self):
class MySink(bt2._UserSinkComponent):
"""
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.description, 'The description.')
def test_empty_description(self):
self.assertEqual(MySink.description, 'The description.')
def test_empty_description(self):
+ def _graph_is_configured(self):
+ pass
+
self.assertIsNone(MySink.description)
def test_help(self):
self.assertIsNone(MySink.description)
def test_help(self):
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
def test_addr(self):
self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
def test_addr(self):
+ def _graph_is_configured(self):
+ pass
+
self.assertIsInstance(MySink.addr, int)
self.assertNotEqual(MySink.addr, 0)
self.assertIsInstance(MySink.addr, int)
self.assertNotEqual(MySink.addr, 0)
+ def _graph_is_configured(self):
+ pass
+
with self.assertRaises(bt2.Error):
bt2.QueryExecutor().query(MySink, 'obj', 23)
with self.assertRaises(bt2.Error):
bt2.QueryExecutor().query(MySink, 'obj', 23)
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise ValueError
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise ValueError
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
return ...
@classmethod
def _query(cls, query_exec, obj, params, log_level):
return ...
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
+ def _graph_is_configured(self):
+ pass
+
@staticmethod
def _query(query_exec, obj, params, log_level):
return
@staticmethod
def _query(query_exec, obj, params, log_level):
return
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
+ def _graph_is_configured(self):
+ pass
+
self.assertEqual(MySink, MySink)
self.assertEqual(MySink, MySink)
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
return [obj, params, 23]
@classmethod
def _query(cls, query_exec, obj, params, log_level):
return [obj, params, 23]
def _consume(self):
raise bt2.Stop
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')
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
def _consume(self):
raise bt2.Stop
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')
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
def _consume(self):
raise bt2.Stop
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')
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
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')
comp = self._graph.add_component(MySink, 'salut')
self.assertEqual(comp.name, 'salut')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut')
assert comp
comp = self._graph.add_component(MySink, 'salut')
assert comp
+ def _graph_is_configured(self):
+ pass
params = {'hello': 23, 'path': '/path/to/stuff'}
comp = self._graph.add_component(MySink, 'salut', params)
params = {'hello': 23, 'path': '/path/to/stuff'}
comp = self._graph.add_component(MySink, 'salut', params)
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
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')
with self.assertRaises(TypeError):
self._graph.add_component(MySink, 'salut', logging_level='yo')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
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)
with self.assertRaises(ValueError):
self._graph.add_component(MySink, 'salut', logging_level=12345)
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
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)
comp = self._graph.add_component(MySink, 'salut',
logging_level=bt2.LoggingLevel.DEBUG)
def _consume(self):
raise bt2.Stop
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')
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
def _consume(self):
raise bt2.Stop
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')
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
def _consume(self):
raise bt2.Stop
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 _port_connected(self, port, other_port):
self._add_input_port('taste')
def _consume(self):
raise bt2.Stop
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 _port_connected(self, port, other_port):
self._add_input_port('taste')
def _consume(self):
raise bt2.Stop
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
graph = bt2.Graph()
with self.assertRaises(bt2.Error):
graph = bt2.Graph()
with self.assertRaises(bt2.Error):
def _consume(self):
raise bt2.Stop
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def port_added_listener(component, port):
raise ValueError('oh noes!')
def port_added_listener(component, port):
raise ValueError('oh noes!')
def _consume(self):
raise bt2.Stop
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 ports_connected_listener(upstream_component, upstream_port,
downstream_component, downstream_port):
raise ValueError('oh noes!')
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(len(comp.input_ports), 1)
comp = self._create_comp(MySink)
self.assertEqual(len(comp.input_ports), 1)
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_user_src_output_ports_getitem_invalid_key(self):
self._create_comp(MySink)
def test_user_src_output_ports_getitem_invalid_key(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_user_src_output_ports_len(self):
self._create_comp(MySink)
def test_user_src_output_ports_len(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
+ 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)
comp = self._create_comp(MySink)
self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
self.assertEqual(port2.addr, comp.input_ports['print'].addr)
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
with self.assertRaises(KeyError):
comp = self._create_comp(MySink)
with self.assertRaises(KeyError):
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(len(comp.input_ports), 3)
comp = self._create_comp(MySink)
self.assertEqual(len(comp.input_ports), 3)
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
ports = []
comp = self._create_comp(MySink)
ports = []
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertEqual(comp.input_ports['clear'].name, 'clear')
comp = self._create_comp(MySink)
self.assertEqual(comp.input_ports['clear'].name, 'clear')
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertIsNone(comp.input_ports['clear'].connection)
comp = self._create_comp(MySink)
self.assertIsNone(comp.input_ports['clear'].connection)
+ def _graph_is_configured(self):
+ pass
+
comp = self._create_comp(MySink)
self.assertFalse(comp.input_ports['clear'].is_connected)
comp = self._create_comp(MySink)
self.assertFalse(comp.input_ports['clear'].is_connected)
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_self_connection_none(self):
self._create_comp(MySink)
def test_self_connection_none(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_self_is_connected_false(self):
self._create_comp(MySink)
def test_self_is_connected_false(self):
+ def _graph_is_configured(self):
+ pass
+
self._create_comp(MySink)
def test_source_self_port_user_data(self):
self._create_comp(MySink)
def test_source_self_port_user_data(self):
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_params
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
@classmethod
def _query(cls, query_exec, obj, params, log_level):
nonlocal query_log_level
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise ValueError
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise ValueError
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidObject
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidObject
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
pass
@classmethod
def _query(cls, query_exec, obj, params, log_level):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
pass
@classmethod
def _query(cls, query_exec, obj, params, log_level):
pass
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidParams
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.InvalidParams
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
+ def _graph_is_configured(self):
+ pass
+
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
@classmethod
def _query(cls, query_exec, obj, params, log_level):
raise bt2.TryAgain
+ def _graph_is_configured(self):
+ pass
+
g = bt2.Graph()
res_bound = None
g.add_component(MySink, 'comp')
g = bt2.Graph()
res_bound = None
g.add_component(MySink, 'comp')
+ def _graph_is_configured(self):
+ pass
+
bt2.register_plugin(__name__, 'sparkling', author='Philippe Proulx',
description='A delicious plugin.',
bt2.register_plugin(__name__, 'sparkling', author='Philippe Proulx',
description='A delicious plugin.',