tests/bindings/python/bt2/test_connection.py

   1 import unittest
   2 import bt2
   3
   4
   5 class ConnectionTestCase(unittest.TestCase):
   6     def test_create(self):
   7         class MyIter(bt2._UserMessageIterator):
   8             def __next__(self):
   9                 raise bt2.Stop
  10
  11         class MySource(bt2._UserSourceComponent,
  12                        message_iterator_class=MyIter):
  13             def __init__(self, params):
  14                 self._add_output_port('out')
  15
  16         class MySink(bt2._UserSinkComponent):
  17             def __init__(self, params):
  18                 self._add_input_port('in')
  19
  20             def _consume(self):
  21                 raise bt2.Stop
  22
  23         graph = bt2.Graph()
  24         src = graph.add_component(MySource, 'src')
  25         sink = graph.add_component(MySink, 'sink')
  26         conn = graph.connect_ports(src.output_ports['out'],
  27                                    sink.input_ports['in'])
  28         self.assertIsInstance(conn, bt2._Connection)
  29
  30     def test_downstream_port(self):
  31         class MyIter(bt2._UserMessageIterator):
  32             def __next__(self):
  33                 raise bt2.Stop
  34
  35         class MySource(bt2._UserSourceComponent,
  36                        message_iterator_class=MyIter):
  37             def __init__(self, params):
  38                 self._add_output_port('out')
  39
  40         class MySink(bt2._UserSinkComponent):
  41             def __init__(self, params):
  42                 self._add_input_port('in')
  43
  44             def _consume(self):
  45                 raise bt2.Stop
  46
  47         graph = bt2.Graph()
  48         src = graph.add_component(MySource, 'src')
  49         sink = graph.add_component(MySink, 'sink')
  50         conn = graph.connect_ports(src.output_ports['out'],
  51                                    sink.input_ports['in'])
  52         self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
  53
  54     def test_upstream_port(self):
  55         class MyIter(bt2._UserMessageIterator):
  56             def __next__(self):
  57                 raise bt2.Stop
  58
  59         class MySource(bt2._UserSourceComponent,
  60                        message_iterator_class=MyIter):
  61             def __init__(self, params):
  62                 self._add_output_port('out')
  63
  64         class MySink(bt2._UserSinkComponent):
  65             def __init__(self, params):
  66                 self._add_input_port('in')
  67
  68             def _consume(self):
  69                 raise bt2.Stop
  70
  71         graph = bt2.Graph()
  72         src = graph.add_component(MySource, 'src')
  73         sink = graph.add_component(MySink, 'sink')
  74         conn = graph.connect_ports(src.output_ports['out'],
  75                                    sink.input_ports['in'])
  76         self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)