X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_port.py;h=5c1a8271fef73a9dc1cc826bc6fc853d21c11e92;hb=5813b3a3ffb8ff44d54b134e26d604af0b5e36db;hp=93b9ab12199fe4d9563e81fe4d6e05adb29929d5;hpb=5602ef8155de326402dcb33f40ee4c7d5d693ca5;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_port.py b/tests/bindings/python/bt2/test_port.py index 93b9ab12..5c1a8271 100644 --- a/tests/bindings/python/bt2/test_port.py +++ b/tests/bindings/python/bt2/test_port.py @@ -1,10 +1,26 @@ -from bt2 import values +# +# Copyright (C) 2019 EfficiOS Inc. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; only version 2 +# of the License. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. +# + import unittest -import copy import bt2 +from bt2 import port as bt2_port -@unittest.skip("this is broken") class PortTestCase(unittest.TestCase): @staticmethod def _create_comp(comp_cls, name=None): @@ -20,24 +36,22 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port = comp_self._add_output_port('out') self.assertEqual(port.name, 'out') comp = self._create_comp(MySource) self.assertEqual(len(comp.output_ports), 1) - + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port = comp_self._add_output_port('out') self.assertEqual(port.name, 'out') @@ -49,22 +63,22 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port = comp_self._add_input_port('in') self.assertEqual(port.name, 'in') comp = self._create_comp(MyFilter) self.assertEqual(len(comp.input_ports), 1) + self.assertIs(type(comp.input_ports['in']), bt2_port._InputPortConst) def test_sink_add_input_port(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): port = comp_self._add_input_port('in') self.assertEqual(port.name, 'in') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -75,9 +89,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -85,16 +98,15 @@ class PortTestCase(unittest.TestCase): self.assertEqual(port2.addr, comp_self._output_ports['print'].addr) self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr) - comp = self._create_comp(MySource) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -102,16 +114,15 @@ class PortTestCase(unittest.TestCase): self.assertEqual(port2.addr, comp_self._output_ports['print'].addr) self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr) - comp = self._create_comp(MyFilter) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -119,11 +130,11 @@ class PortTestCase(unittest.TestCase): self.assertEqual(port2.addr, comp_self._input_ports['print'].addr) self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr) - comp = self._create_comp(MyFilter) + self._create_comp(MyFilter) def test_user_sink_input_ports_getitem(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -131,19 +142,18 @@ class PortTestCase(unittest.TestCase): self.assertEqual(port2.addr, comp_self._input_ports['print'].addr) self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr) - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + 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): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -151,16 +161,15 @@ class PortTestCase(unittest.TestCase): with self.assertRaises(KeyError): comp_self._output_ports['hello'] - comp = self._create_comp(MySource) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -168,16 +177,15 @@ class PortTestCase(unittest.TestCase): with self.assertRaises(KeyError): comp_self._output_ports['hello'] - comp = self._create_comp(MyFilter) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -185,11 +193,11 @@ class PortTestCase(unittest.TestCase): with self.assertRaises(KeyError): comp_self._input_ports['hello'] - comp = self._create_comp(MyFilter) + self._create_comp(MyFilter) def test_user_sink_input_ports_getitem_invalid_key(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -197,77 +205,73 @@ class PortTestCase(unittest.TestCase): with self.assertRaises(KeyError): comp_self._input_ports['hello'] - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + 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): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') self.assertEqual(len(comp_self._output_ports), 3) - comp = self._create_comp(MySource) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') self.assertEqual(len(comp_self._output_ports), 3) - comp = self._create_comp(MyFilter) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') self.assertEqual(len(comp_self._input_ports), 3) - comp = self._create_comp(MyFilter) + self._create_comp(MyFilter) def test_user_sink_input_ports_len(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') self.assertEqual(len(comp_self._input_ports), 3) - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + 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): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -283,16 +287,15 @@ class PortTestCase(unittest.TestCase): self.assertEqual(ports[2][0], 'insert') self.assertEqual(ports[2][1].addr, port3.addr) - comp = self._create_comp(MySource) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -308,16 +311,15 @@ class PortTestCase(unittest.TestCase): self.assertEqual(ports[2][0], 'insert') self.assertEqual(ports[2][1].addr, port3.addr) - comp = self._create_comp(MyFilter) + 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): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -333,11 +335,11 @@ class PortTestCase(unittest.TestCase): self.assertEqual(ports[2][0], 'insert') self.assertEqual(ports[2][1].addr, port3.addr) - comp = self._create_comp(MyFilter) + self._create_comp(MyFilter) def test_user_sink_input_ports_iter(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -353,10 +355,10 @@ class PortTestCase(unittest.TestCase): self.assertEqual(ports[2][0], 'insert') self.assertEqual(ports[2][1].addr, port3.addr) - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + self._create_comp(MySink) def test_gen_src_output_ports_getitem(self): class MyIter(bt2._UserMessageIterator): @@ -367,9 +369,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -392,9 +393,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -417,9 +417,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') @@ -439,13 +438,13 @@ class PortTestCase(unittest.TestCase): port3 = None class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -461,9 +460,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -478,9 +476,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -495,9 +492,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -509,7 +505,7 @@ class PortTestCase(unittest.TestCase): def test_gen_sink_input_ports_getitem_invalid_key(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -517,7 +513,7 @@ class PortTestCase(unittest.TestCase): with self.assertRaises(KeyError): comp_self._input_ports['hello'] - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -530,9 +526,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -545,9 +540,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -560,9 +554,8 @@ class PortTestCase(unittest.TestCase): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -572,12 +565,12 @@ class PortTestCase(unittest.TestCase): def test_gen_sink_input_ports_len(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -592,9 +585,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -625,9 +617,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -658,9 +649,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - message_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') @@ -688,13 +678,13 @@ class PortTestCase(unittest.TestCase): port3 = None class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -715,32 +705,21 @@ class PortTestCase(unittest.TestCase): def test_name(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) self.assertEqual(comp.input_ports['clear'].name, 'clear') - def test_component(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - comp_self._add_input_port('clear') - - def _consume(self): - pass - - comp = self._create_comp(MySink) - self.assertEqual(comp.input_ports['clear'].component.addr, comp.addr) - def test_connection_none(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -748,142 +727,108 @@ class PortTestCase(unittest.TestCase): def test_is_connected_false(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): comp_self._add_input_port('clear') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) self.assertFalse(comp.input_ports['clear'].is_connected) - def test_eq(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - comp_self._add_input_port('clear') - - def _consume(self): - pass - - comp = self._create_comp(MySink) - self.assertEqual(comp.input_ports['clear'], - comp.input_ports['clear']) - - def test_eq_invalid(self): + def test_self_name(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - comp_self._add_input_port('clear') - - def _consume(self): - pass - - comp = self._create_comp(MySink) - self.assertNotEqual(comp.input_ports['clear'], 23) - - def test_disconnect_no_connection(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - - def _consume(self): - pass - - comp = self._create_comp(MySink) - comp.input_ports['clear'].disconnect() - - def test_priv_name(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): port = comp_self._add_input_port('clear') self.assertEqual(port.name, 'clear') - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + self._create_comp(MySink) - def test_priv_component(self): + def test_self_connection_none(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertEqual(port.component, comp_self) - - def _consume(self): - pass - - comp = self._create_comp(MySink) - - def test_priv_connection_none(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): port = comp_self._add_input_port('clear') self.assertIsNone(port.connection) - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + self._create_comp(MySink) - def test_priv_is_connected_false(self): + def test_self_is_connected_false(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, params, obj): port = comp_self._add_input_port('clear') self.assertFalse(port.is_connected) - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + self._create_comp(MySink) - def test_priv_eq(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertEqual(port, port) + def test_source_self_port_user_data(self): + class MyIter(bt2._UserMessageIterator): + def __next__(self): + raise bt2.Stop - def _consume(self): - pass + class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): + nonlocal user_datas - comp = self._create_comp(MySink) + p = comp_self._add_output_port('port1') + user_datas.append(p.user_data) + p = comp_self._add_output_port('port2', 2) + user_datas.append(p.user_data) - def test_priv_eq_invalid(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertNotEqual(port, 23) + user_datas = [] - def _consume(self): - pass + self._create_comp(MySource) + self.assertEqual(user_datas, [None, 2]) - comp = self._create_comp(MySink) + def test_filter_self_port_user_data(self): + class MyIter(bt2._UserMessageIterator): + def __next__(self): + raise bt2.Stop - def test_priv_disconnect_no_connection(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - port.disconnect() + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): + nonlocal user_datas - def _consume(self): - pass + p = comp_self._add_output_port('port1') + user_datas.append(p.user_data) + p = comp_self._add_output_port('port2', 'user data string') + user_datas.append(p.user_data) - comp = self._create_comp(MySink) + p = comp_self._add_input_port('port3') + user_datas.append(p.user_data) + p = comp_self._add_input_port('port4', user_data={'user data': 'dict'}) + user_datas.append(p.user_data) - def test_priv_remove_from_component(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertEqual(len(comp_self._input_ports), 1) + user_datas = [] + + self._create_comp(MyFilter) + self.assertEqual( + user_datas, [None, 'user data string', None, {'user data': 'dict'}] + ) + + def test_sink_self_port_user_data(self): + class MyIter(bt2._UserMessageIterator): + def __next__(self): + raise bt2.Stop - try: - port.remove_from_component() - except: - import traceback - traceback.print_exc() + class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, params, obj): + nonlocal user_datas - self.assertEqual(len(comp_self._input_ports), 0) - self.assertIsNone(port.component) + p = comp_self._add_input_port('port1') + user_datas.append(p.user_data) + p = comp_self._add_input_port('port2', set()) + user_datas.append(p.user_data) - def _consume(self): - pass + user_datas = [] - comp = self._create_comp(MySink) + self._create_comp(MySink) + self.assertEqual(user_datas, [None, set()])