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): |
f5567ea8 | 18 | self._add_output_port("out") |
811644b8 PP |
19 | |
20 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 21 | def __init__(self, config, params, obj): |
f5567ea8 | 22 | self._add_input_port("in") |
811644b8 | 23 | |
6a91742b | 24 | def _user_consume(self): |
811644b8 PP |
25 | raise bt2.Stop |
26 | ||
27 | graph = bt2.Graph() | |
f5567ea8 FD |
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"]) | |
c7e5224b | 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): |
f5567ea8 | 38 | self._add_output_port("out") |
811644b8 PP |
39 | |
40 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 41 | def __init__(self, config, params, obj): |
f5567ea8 | 42 | self._add_input_port("in") |
811644b8 | 43 | |
6a91742b | 44 | def _user_consume(self): |
811644b8 PP |
45 | raise bt2.Stop |
46 | ||
47 | graph = bt2.Graph() | |
f5567ea8 FD |
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) | |
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): |
f5567ea8 | 60 | self._add_output_port("out") |
811644b8 PP |
61 | |
62 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 63 | def __init__(self, config, params, obj): |
f5567ea8 | 64 | self._add_input_port("in") |
811644b8 | 65 | |
6a91742b | 66 | def _user_consume(self): |
811644b8 PP |
67 | raise bt2.Stop |
68 | ||
69 | graph = bt2.Graph() | |
f5567ea8 FD |
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) | |
5813b3a3 | 74 | self.assertIs(type(conn.upstream_port), bt2_port._OutputPortConst) |
d14ddbba SM |
75 | |
76 | ||
f5567ea8 | 77 | if __name__ == "__main__": |
d14ddbba | 78 | unittest.main() |