bt2: remove native_bt.d in `make clean`
[babeltrace.git] / tests / bindings / python / bt2 / test_connection.py
1 #
2 # Copyright (C) 2019 EfficiOS Inc.
3 #
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
7 # of the License.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
13 #
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
17 #
18
19 import unittest
20 import bt2
21 from bt2 import connection as bt2_connection
22 from bt2 import port as bt2_port
23
24
25 class ConnectionTestCase(unittest.TestCase):
26 def test_create(self):
27 class MyIter(bt2._UserMessageIterator):
28 def __next__(self):
29 raise bt2.Stop
30
31 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
32 def __init__(self, params, obj):
33 self._add_output_port('out')
34
35 class MySink(bt2._UserSinkComponent):
36 def __init__(self, params, obj):
37 self._add_input_port('in')
38
39 def _user_consume(self):
40 raise bt2.Stop
41
42 graph = bt2.Graph()
43 src = graph.add_component(MySource, 'src')
44 sink = graph.add_component(MySink, 'sink')
45 conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
46 self.assertIs(type(conn), bt2_connection._ConnectionConst)
47
48 def test_downstream_port(self):
49 class MyIter(bt2._UserMessageIterator):
50 def __next__(self):
51 raise bt2.Stop
52
53 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
54 def __init__(self, params, obj):
55 self._add_output_port('out')
56
57 class MySink(bt2._UserSinkComponent):
58 def __init__(self, params, obj):
59 self._add_input_port('in')
60
61 def _user_consume(self):
62 raise bt2.Stop
63
64 graph = bt2.Graph()
65 src = graph.add_component(MySource, 'src')
66 sink = graph.add_component(MySink, 'sink')
67 conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
68 self.assertEqual(conn.downstream_port.addr, sink.input_ports['in'].addr)
69 self.assertIs(type(conn), bt2_connection._ConnectionConst)
70 self.assertIs(type(conn.downstream_port), bt2_port._InputPortConst)
71
72 def test_upstream_port(self):
73 class MyIter(bt2._UserMessageIterator):
74 def __next__(self):
75 raise bt2.Stop
76
77 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
78 def __init__(self, params, obj):
79 self._add_output_port('out')
80
81 class MySink(bt2._UserSinkComponent):
82 def __init__(self, params, obj):
83 self._add_input_port('in')
84
85 def _user_consume(self):
86 raise bt2.Stop
87
88 graph = bt2.Graph()
89 src = graph.add_component(MySource, 'src')
90 sink = graph.add_component(MySink, 'sink')
91 conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
92 self.assertEqual(conn.upstream_port.addr, src.output_ports['out'].addr)
93 self.assertIs(type(conn.upstream_port), bt2_port._OutputPortConst)
This page took 0.035416 seconds and 5 git commands to generate.