X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_connection.py;fp=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_connection.py;h=490c984b20b70eba45e224a4c86df388e0cbdab8;hb=2c67587ab59a15b012008ca38cc13fb77f0eb776;hp=1ff932835bf332011e5939fce7a3154e42b06171;hpb=894a8df566ca63e6b53c831d245569d8598c5889;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_connection.py b/tests/bindings/python/bt2/test_connection.py index 1ff93283..490c984b 100644 --- a/tests/bindings/python/bt2/test_connection.py +++ b/tests/bindings/python/bt2/test_connection.py @@ -1,10 +1,7 @@ -from bt2 import value import unittest -import copy import bt2 -@unittest.skip("this is broken") class ConnectionTestCase(unittest.TestCase): def test_create(self): class MyIter(bt2._UserMessageIterator): @@ -29,56 +26,6 @@ class ConnectionTestCase(unittest.TestCase): 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): - 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']) - src.output_ports['out'].disconnect() - self.assertTrue(conn.is_ended) def test_downstream_port(self): class MyIter(bt2._UserMessageIterator): @@ -127,268 +74,3 @@ class ConnectionTestCase(unittest.TestCase): conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in']) self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr) - - 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