configure: disable -Wmaybe-uninitialized
[babeltrace.git] / 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()
This page took 0.032211 seconds and 5 git commands to generate.