X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_port.py;h=53fdf9c79168a81d99cdcd4c65f602c9e62a7c01;hb=30b828d2858e5b3823f0b36a01563855564d8c19;hp=592a5b6fcd915b5f7beea939dc582042dc6e049c;hpb=811644b8fe5fb9946972a7ace9df02ed872f448a;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_port.py b/tests/bindings/python/bt2/test_port.py index 592a5b6f..53fdf9c7 100644 --- a/tests/bindings/python/bt2/test_port.py +++ b/tests/bindings/python/bt2/test_port.py @@ -1,7 +1,24 @@ -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 class PortTestCase(unittest.TestCase): @@ -15,28 +32,26 @@ class PortTestCase(unittest.TestCase): return graph.add_component(comp_cls, name) def test_src_add_output_port(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, 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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port = comp_self._add_output_port('out') self.assertEqual(port.name, 'out') @@ -44,39 +59,38 @@ class PortTestCase(unittest.TestCase): self.assertEqual(len(comp.output_ports), 1) def test_flt_add_input_port(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, 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, config, 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) self.assertEqual(len(comp.input_ports), 1) def test_user_src_output_ports_getitem(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -84,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -101,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -118,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, config, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -130,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -150,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -167,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -184,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, config, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -196,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, 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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, 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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, 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, config, 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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -282,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') port3 = comp_self._add_output_port('insert') @@ -307,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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -332,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, config, params, obj): port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') port3 = comp_self._add_input_port('insert') @@ -352,13 +355,13 @@ 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._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop @@ -366,9 +369,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -383,7 +385,7 @@ class PortTestCase(unittest.TestCase): del port3 def test_gen_flt_output_ports_getitem(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop @@ -391,9 +393,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -408,7 +409,7 @@ class PortTestCase(unittest.TestCase): del port3 def test_gen_flt_input_ports_getitem(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop @@ -416,9 +417,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') @@ -438,13 +438,13 @@ class PortTestCase(unittest.TestCase): port3 = None class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, config, 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) @@ -456,13 +456,12 @@ class PortTestCase(unittest.TestCase): del port3 def test_gen_src_output_ports_getitem_invalid_key(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -473,13 +472,12 @@ class PortTestCase(unittest.TestCase): comp.output_ports['hello'] def test_gen_flt_output_ports_getitem_invalid_key(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -490,13 +488,12 @@ class PortTestCase(unittest.TestCase): comp.output_ports['hello'] def test_gen_flt_input_ports_getitem_invalid_key(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -508,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, config, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -516,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) @@ -525,13 +522,12 @@ class PortTestCase(unittest.TestCase): comp.input_ports['hello'] def test_gen_src_output_ports_len(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -540,13 +536,12 @@ class PortTestCase(unittest.TestCase): self.assertEqual(len(comp.output_ports), 3) def test_gen_flt_output_ports_len(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_output_port('clear') comp_self._add_output_port('print') comp_self._add_output_port('insert') @@ -555,13 +550,12 @@ class PortTestCase(unittest.TestCase): self.assertEqual(len(comp.output_ports), 3) def test_gen_flt_input_ports_len(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): comp_self._add_input_port('clear') comp_self._add_input_port('print') comp_self._add_input_port('insert') @@ -571,19 +565,19 @@ 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, config, 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) self.assertEqual(len(comp.input_ports), 3) def test_gen_src_output_ports_iter(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop @@ -591,9 +585,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MySource(bt2._UserSourceComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -616,7 +609,7 @@ class PortTestCase(unittest.TestCase): del port3 def test_gen_flt_output_ports_iter(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop @@ -624,9 +617,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_output_port('clear') port2 = comp_self._add_output_port('print') @@ -649,7 +641,7 @@ class PortTestCase(unittest.TestCase): del port3 def test_gen_flt_input_ports_iter(self): - class MyIter(bt2._UserNotificationIterator): + class MyIter(bt2._UserMessageIterator): def __next__(self): raise bt2.Stop @@ -657,9 +649,8 @@ class PortTestCase(unittest.TestCase): port2 = None port3 = None - class MyFilter(bt2._UserFilterComponent, - notification_iterator_class=MyIter): - def __init__(comp_self, params): + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): nonlocal port1, port2, port3 port1 = comp_self._add_input_port('clear') port2 = comp_self._add_input_port('print') @@ -687,13 +678,13 @@ class PortTestCase(unittest.TestCase): port3 = None class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, config, 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) @@ -714,32 +705,21 @@ class PortTestCase(unittest.TestCase): def test_name(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, config, 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, config, params, obj): comp_self._add_input_port('clear') - def _consume(self): + def _user_consume(self): pass comp = self._create_comp(MySink) @@ -747,142 +727,112 @@ class PortTestCase(unittest.TestCase): def test_is_connected_false(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, config, 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): - 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): + def test_self_name(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, config, 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): + def __init__(comp_self, config, params, obj): port = comp_self._add_input_port('clear') - self.assertEqual(port.component, comp_self) + self.assertIsNone(port.connection) - def _consume(self): + def _user_consume(self): pass - comp = self._create_comp(MySink) + self._create_comp(MySink) - def test_priv_connection_none(self): + def test_self_is_connected_false(self): class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): + def __init__(comp_self, config, params, obj): port = comp_self._add_input_port('clear') - self.assertIsNone(port.connection) + 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_is_connected_false(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertFalse(port.is_connected) + 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, config, 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(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertEqual(port, port) + 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_eq_invalid(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - self.assertNotEqual(port, 23) + class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, 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_disconnect_no_connection(self): - class MySink(bt2._UserSinkComponent): - def __init__(comp_self, params): - port = comp_self._add_input_port('clear') - port.disconnect() + user_datas = [] - def _consume(self): - pass + self._create_comp(MyFilter) + self.assertEqual( + user_datas, [None, 'user data string', None, {'user data': 'dict'}] + ) - comp = self._create_comp(MySink) + def test_sink_self_port_user_data(self): + class MyIter(bt2._UserMessageIterator): + def __next__(self): + raise bt2.Stop - 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) + class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter): + def __init__(comp_self, config, params, obj): + nonlocal user_datas - try: - port.remove_from_component() - except: - import traceback - traceback.print_exc() + 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) - self.assertEqual(len(comp_self._input_ports), 0) - self.assertIsNone(port.component) + user_datas = [] - def _consume(self): - pass + self._create_comp(MySink) + self.assertEqual(user_datas, [None, set()]) - comp = self._create_comp(MySink) + +if __name__ == '__main__': + unittest.main()