tests/bindings/python/bt2/test_connection.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7 import bt2
   8 from bt2 import connection as bt2_connection
   9 from bt2 import port as bt2_port
  10
  11
  12 class ConnectionTestCase(unittest.TestCase):
  13     def test_create(self):
  14         class MySource(
  15             bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  16         ):
  17             def __init__(self, config, params, obj):
  18                 self._add_output_port('out')
  19
  20         class MySink(bt2._UserSinkComponent):
  21             def __init__(self, config, params, obj):
  22                 self._add_input_port('in')
  23
  24             def _user_consume(self):
  25                 raise bt2.Stop
  26
  27         graph = bt2.Graph()
  28         src = graph.add_component(MySource, 'src')
  29         sink = graph.add_component(MySink, 'sink')
  30         conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
  31         self.assertIs(type(conn), bt2_connection._ConnectionConst)
  32
  33     def test_downstream_port(self):
  34         class MySource(
  35             bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  36         ):
  37             def __init__(self, config, params, obj):
  38                 self._add_output_port('out')
  39
  40         class MySink(bt2._UserSinkComponent):
  41             def __init__(self, config, params, obj):
  42                 self._add_input_port('in')
  43
  44             def _user_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'], sink.input_ports['in'])
  51         self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
  52         self.assertIs(type(conn), bt2_connection._ConnectionConst)
  53         self.assertIs(type(conn.downstream_port), bt2_port._InputPortConst)
  54
  55     def test_upstream_port(self):
  56         class MySource(
  57             bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  58         ):
  59             def __init__(self, config, params, obj):
  60                 self._add_output_port('out')
  61
  62         class MySink(bt2._UserSinkComponent):
  63             def __init__(self, config, params, obj):
  64                 self._add_input_port('in')
  65
  66             def _user_consume(self):
  67                 raise bt2.Stop
  68
  69         graph = bt2.Graph()
  70         src = graph.add_component(MySource, 'src')
  71         sink = graph.add_component(MySink, 'sink')
  72         conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
  73         self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)
  74         self.assertIs(type(conn.upstream_port), bt2_port._OutputPortConst)
  75
  76
  77 if __name__ == '__main__':
  78     unittest.main()