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