X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_connection.py;h=7ec6bb01188980c04bc22a0cfa1673b3bd67603a;hb=f5567ea88d172767b34373bc6e402da8bfd85ef8;hp=1ff932835bf332011e5939fce7a3154e42b06171;hpb=c4239792c6758f579fb9482ddccf336f1bbf26c4;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_connection.py b/tests/bindings/python/bt2/test_connection.py index 1ff93283..7ec6bb01 100644 --- a/tests/bindings/python/bt2/test_connection.py +++ b/tests/bindings/python/bt2/test_connection.py @@ -1,394 +1,78 @@ -from bt2 import value +# SPDX-License-Identifier: GPL-2.0-only +# +# Copyright (C) 2019 EfficiOS Inc. +# + 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 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): - self._add_input_port('in') + 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']) - 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) + 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.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): - self._add_output_port('out') + 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): - self._add_input_port('in') + 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']) - self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr) + 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.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): - 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.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 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): - self._add_input_port('in') + 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']) - self.assertEqual(conn, conn) - - def test_eq_invalid(self): - class MyIter(bt2._UserMessageIterator): - def __next__(self): - raise bt2.Stop + 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.upstream_port.addr, src.output_ports["out"].addr) + self.assertIs(type(conn.upstream_port), bt2_port._OutputPortConst) - 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()