These user message iterator classes don't do anything special, we can
remove them and pass bt2._UserMessageIterator directly to the components
instead, reducing a little bit the noise in the tests.
Some classes in test_component_class.py were simply unused, so I removed
them.
Change-Id: I79f787b9d121a3dc6456f81090ebf51d088d2c73
Signed-off-by: Simon Marchi <simon.marchi@efficios.com>
Reviewed-on: https://review.lttng.org/c/babeltrace/+/2736
Tested-by: jenkins <jenkins@lttng.org>
cls()
def test_no_init_source(self):
cls()
def test_no_init_source(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
pass
self._test_no_init(MySource)
def test_no_init_filter(self):
pass
self._test_no_init(MySource)
def test_no_init_filter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
pass
self._test_no_init(MyFilter)
pass
self._test_no_init(MyFilter)
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):
- class MyIter(bt2._UserMessageIterator):
- pass
-
with self.assertRaises(bt2._IncompleteUserClass):
class MySource(bt2._UserSourceComponent):
pass
def test_incomplete_source_wrong_msg_iter_cls_type(self):
with self.assertRaises(bt2._IncompleteUserClass):
class MySource(bt2._UserSourceComponent):
pass
def test_incomplete_source_wrong_msg_iter_cls_type(self):
- class MyIter(bt2._UserMessageIterator):
- pass
-
with self.assertRaises(bt2._IncompleteUserClass):
class MySource(bt2._UserSourceComponent, message_iterator_class=int):
pass
def test_incomplete_filter_no_msg_iter_cls(self):
with self.assertRaises(bt2._IncompleteUserClass):
class MySource(bt2._UserSourceComponent, message_iterator_class=int):
pass
def test_incomplete_filter_no_msg_iter_cls(self):
- class MyIter(bt2._UserMessageIterator):
- pass
-
with self.assertRaises(bt2._IncompleteUserClass):
class MyFilter(bt2._UserFilterComponent):
pass
def test_incomplete_sink_no_consume_method(self):
with self.assertRaises(bt2._IncompleteUserClass):
class MyFilter(bt2._UserFilterComponent):
pass
def test_incomplete_sink_no_consume_method(self):
- class MyIter(bt2._UserMessageIterator):
- pass
-
with self.assertRaises(bt2._IncompleteUserClass):
class MySink(bt2._UserSinkComponent):
pass
def test_minimal_source(self):
with self.assertRaises(bt2._IncompleteUserClass):
class MySink(bt2._UserSinkComponent):
pass
def test_minimal_source(self):
- class MyIter(bt2._UserMessageIterator):
- pass
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
pass
def test_minimal_filter(self):
pass
def test_minimal_filter(self):
- class MyIter(bt2._UserMessageIterator):
- pass
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
pass
def test_minimal_sink(self):
pass
def test_minimal_sink(self):
class ConnectionTestCase(unittest.TestCase):
def test_create(self):
class ConnectionTestCase(unittest.TestCase):
def test_create(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
def __init__(self, config, params, obj):
self._add_output_port('out')
self.assertIs(type(conn), bt2_connection._ConnectionConst)
def test_downstream_port(self):
self.assertIs(type(conn), bt2_connection._ConnectionConst)
def test_downstream_port(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
def __init__(self, config, params, obj):
self._add_output_port('out')
self.assertIs(type(conn.downstream_port), bt2_port._InputPortConst)
def test_upstream_port(self):
self.assertIs(type(conn.downstream_port), bt2_port._InputPortConst)
def test_upstream_port(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
def __init__(self, config, params, obj):
self._add_output_port('out')
self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)
def test_connect_ports(self):
self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)
def test_connect_ports(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
def __init__(self, config, params, obj):
self._add_output_port('out')
self.assertEqual(sink.input_ports['in'].connection.addr, conn.addr)
def test_connect_ports_invalid_direction(self):
self.assertEqual(sink.input_ports['in'].connection.addr, conn.addr)
def test_connect_ports_invalid_direction(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
def __init__(self, config, params, obj):
self._add_output_port('out')
self._graph.run()
def test_listeners(self):
self._graph.run()
def test_listeners(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
self._add_output_port('zero')
def __init__(self, config, params, obj):
self._add_output_port('out')
self._add_output_port('zero')
self.assertEqual(calls[3][2].name, 'taste')
def test_invalid_listeners(self):
self.assertEqual(calls[3][2].name, 'taste')
def test_invalid_listeners(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(self, config, params, obj):
self._add_output_port('out')
self._add_output_port('zero')
def __init__(self, config, params, obj):
self._add_output_port('out')
self._add_output_port('zero')
return graph.add_component(comp_cls, name)
def test_src_add_output_port(self):
return graph.add_component(comp_cls, name)
def test_src_add_output_port(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
def __init__(comp_self, config, params, obj):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
self.assertIs(type(comp.output_ports['out']), bt2_port._OutputPortConst)
def test_flt_add_output_port(self):
self.assertIs(type(comp.output_ports['out']), bt2_port._OutputPortConst)
def test_flt_add_output_port(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
def __init__(comp_self, config, params, obj):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
self.assertEqual(len(comp.output_ports), 1)
def test_flt_add_input_port(self):
self.assertEqual(len(comp.output_ports), 1)
def test_flt_add_input_port(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
def __init__(comp_self, config, params, obj):
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
self.assertEqual(len(comp.input_ports), 1)
def test_user_src_output_ports_getitem(self):
self.assertEqual(len(comp.input_ports), 1)
def test_user_src_output_ports_getitem(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
self._create_comp(MySource)
def test_user_flt_output_ports_getitem(self):
self._create_comp(MySource)
def test_user_flt_output_ports_getitem(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem(self):
self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
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):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self._create_comp(MySource)
def test_user_flt_output_ports_getitem_invalid_key(self):
self._create_comp(MySource)
def test_user_flt_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem_invalid_key(self):
self._create_comp(MyFilter)
def test_user_flt_input_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
self._create_comp(MySink)
def test_user_src_output_ports_len(self):
self._create_comp(MySink)
def test_user_src_output_ports_len(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self._create_comp(MySource)
def test_user_flt_output_ports_len(self):
self._create_comp(MySource)
def test_user_flt_output_ports_len(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self._create_comp(MyFilter)
def test_user_flt_input_ports_len(self):
self._create_comp(MyFilter)
def test_user_flt_input_ports_len(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
self._create_comp(MySink)
def test_user_src_output_ports_iter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
self._create_comp(MySource)
def test_user_flt_output_ports_iter(self):
self._create_comp(MySource)
def test_user_flt_output_ports_iter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
self._create_comp(MyFilter)
def test_user_flt_input_ports_iter(self):
self._create_comp(MyFilter)
def test_user_flt_input_ports_iter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
def __init__(comp_self, config, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
self._create_comp(MySink)
def test_gen_src_output_ports_getitem(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
port1 = None
port2 = None
port3 = None
port1 = None
port2 = None
port3 = None
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_output_ports_getitem(self):
del port3
def test_gen_flt_output_ports_getitem(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
port1 = None
port2 = None
port3 = None
port1 = None
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_input_ports_getitem(self):
del port3
def test_gen_flt_input_ports_getitem(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
port1 = None
port2 = None
port3 = None
port1 = None
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
del port3
def test_gen_src_output_ports_getitem_invalid_key(self):
del port3
def test_gen_src_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp.output_ports['hello']
def test_gen_flt_output_ports_getitem_invalid_key(self):
comp.output_ports['hello']
def test_gen_flt_output_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp.output_ports['hello']
def test_gen_flt_input_ports_getitem_invalid_key(self):
comp.output_ports['hello']
def test_gen_flt_input_ports_getitem_invalid_key(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp.input_ports['hello']
def test_gen_src_output_ports_len(self):
comp.input_ports['hello']
def test_gen_src_output_ports_len(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self.assertEqual(len(comp.output_ports), 3)
def test_gen_flt_output_ports_len(self):
self.assertEqual(len(comp.output_ports), 3)
def test_gen_flt_output_ports_len(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
self.assertEqual(len(comp.output_ports), 3)
def test_gen_flt_input_ports_len(self):
self.assertEqual(len(comp.output_ports), 3)
def test_gen_flt_input_ports_len(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
def __init__(comp_self, config, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
self.assertEqual(len(comp.input_ports), 3)
def test_gen_src_output_ports_iter(self):
self.assertEqual(len(comp.input_ports), 3)
def test_gen_src_output_ports_iter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
port1 = None
port2 = None
port3 = None
port1 = None
port2 = None
port3 = None
- class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_output_ports_iter(self):
del port3
def test_gen_flt_output_ports_iter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
port1 = None
port2 = None
port3 = None
port1 = None
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
del port3
def test_gen_flt_input_ports_iter(self):
del port3
def test_gen_flt_input_ports_iter(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
port1 = None
port2 = None
port3 = None
port1 = None
port2 = None
port3 = None
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
def __init__(comp_self, config, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
self._create_comp(MySink)
def test_source_self_port_user_data(self):
self._create_comp(MySink)
def test_source_self_port_user_data(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MySource(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal user_datas
def __init__(comp_self, config, params, obj):
nonlocal user_datas
self.assertEqual(user_datas, [None, 2])
def test_filter_self_port_user_data(self):
self.assertEqual(user_datas, [None, 2])
def test_filter_self_port_user_data(self):
- class MyIter(bt2._UserMessageIterator):
- def __next__(self):
- raise bt2.Stop
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
+ class MyFilter(
+ bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
+ ):
def __init__(comp_self, config, params, obj):
nonlocal user_datas
def __init__(comp_self, config, params, obj):
nonlocal user_datas