X-Git-Url: https://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_connection.py;h=d25e7178199032d67da9a672f3a44d019e1b625c;hb=a797a36fa286c4c16999dd9972a4e81338e3d7b8;hp=1ff932835bf332011e5939fce7a3154e42b06171;hpb=ae0bfae8ae6f4a54d7fcf55bf580a8e03d4f58ed;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_connection.py b/tests/bindings/python/bt2/test_connection.py index 1ff93283..d25e7178 100644 --- a/tests/bindings/python/bt2/test_connection.py +++ b/tests/bindings/python/bt2/test_connection.py @@ -1,394 +1,91 @@ -from bt2 import value +# +# 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 connection as bt2_connection +from bt2 import port as bt2_port -@unittest.skip("this is broken") 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): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertIsInstance(conn, bt2._Connection) - self.assertNotIsInstance(conn, bt2._PrivateConnection) - - def test_is_ended_false(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertFalse(conn.is_ended) - - def test_is_ended_true(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): + class MySource( + bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator + ): + def __init__(self, config, params, obj): self._add_output_port('out') class MySink(bt2._UserSinkComponent): - def __init__(self, params): + def __init__(self, config, params, obj): self._add_input_port('in') - def _consume(self): + def _user_consume(self): raise bt2.Stop graph = bt2.Graph() src = graph.add_component(MySource, 'src') sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - src.output_ports['out'].disconnect() - self.assertTrue(conn.is_ended) + conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in']) + 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): - def __init__(self, params): + class MySource( + bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator + ): + def __init__(self, config, params, obj): self._add_output_port('out') class MySink(bt2._UserSinkComponent): - def __init__(self, params): + def __init__(self, config, params, obj): self._add_input_port('in') - def _consume(self): + def _user_consume(self): raise bt2.Stop graph = bt2.Graph() src = graph.add_component(MySource, 'src') sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) + conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in']) self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr) + self.assertIs(type(conn), bt2_connection._ConnectionConst) + 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): - def __init__(self, params): + class MySource( + bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator + ): + def __init__(self, config, params, obj): self._add_output_port('out') class MySink(bt2._UserSinkComponent): - def __init__(self, params): + def __init__(self, config, params, obj): self._add_input_port('in') - def _consume(self): + def _user_consume(self): raise bt2.Stop graph = bt2.Graph() src = graph.add_component(MySource, 'src') sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) + conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in']) self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr) + self.assertIs(type(conn.upstream_port), bt2_port._OutputPortConst) - def test_eq(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertEqual(conn, conn) - - def test_eq_invalid(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertNotEqual(conn, 23) - - -@unittest.skip("this is broken") -class PrivateConnectionTestCase(unittest.TestCase): - def test_create(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_conn - priv_conn = port.connection - priv_conn = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertIsInstance(priv_conn, bt2._PrivateConnection) - self.assertEqual(conn, priv_conn) - del priv_conn - - def test_is_ended_false(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_conn - priv_conn = port.connection - - priv_conn = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertFalse(priv_conn.is_ended) - del priv_conn - - def test_is_ended_true(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_conn - priv_conn = port.connection - - priv_conn = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - sink.input_ports['in'].disconnect() - self.assertTrue(priv_conn.is_ended) - del priv_conn - - def test_downstream_port(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_port - priv_conn = port.connection - priv_port = priv_conn.downstream_port - - priv_port = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertEqual(priv_port.addr, sink.input_ports['in'].addr) - del priv_port - - def test_upstream_port(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_port - priv_conn = port.connection - priv_port = priv_conn.upstream_port - - priv_port = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertEqual(priv_port.addr, src.output_ports['out'].addr) - del priv_port - - def test_eq(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_conn - priv_conn = port.connection - - priv_conn = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertEqual(priv_conn, conn) - del priv_conn - - def test_eq_invalid(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop - - class MySource(bt2._UserSourceComponent, - message_iterator_class=MyIter): - def __init__(self, params): - self._add_output_port('out') - - class MySink(bt2._UserSinkComponent): - def __init__(self, params): - self._add_input_port('in') - - def _consume(self): - raise bt2.Stop - - def _port_connected(self, port, other_port): - nonlocal priv_conn - priv_conn = port.connection - - priv_conn = None - graph = bt2.Graph() - src = graph.add_component(MySource, 'src') - sink = graph.add_component(MySink, 'sink') - conn = graph.connect_ports(src.output_ports['out'], - sink.input_ports['in']) - self.assertNotEqual(priv_conn, 23) - del priv_conn +if __name__ == '__main__': + unittest.main()