5 class ConnectionTestCase(unittest
.TestCase
):
7 class MyIter(bt2
._UserMessageIterator
):
11 class MySource(bt2
._UserSourceComponent
,
12 message_iterator_class
=MyIter
):
13 def __init__(self
, params
):
14 self
._add
_output
_port
('out')
16 class MySink(bt2
._UserSinkComponent
):
17 def __init__(self
, params
):
18 self
._add
_input
_port
('in')
24 src
= graph
.add_component(MySource
, 'src')
25 sink
= graph
.add_component(MySink
, 'sink')
26 conn
= graph
.connect_ports(src
.output_ports
['out'],
27 sink
.input_ports
['in'])
28 self
.assertIsInstance(conn
, bt2
._Connection
)
30 def test_downstream_port(self
):
31 class MyIter(bt2
._UserMessageIterator
):
35 class MySource(bt2
._UserSourceComponent
,
36 message_iterator_class
=MyIter
):
37 def __init__(self
, params
):
38 self
._add
_output
_port
('out')
40 class MySink(bt2
._UserSinkComponent
):
41 def __init__(self
, params
):
42 self
._add
_input
_port
('in')
48 src
= graph
.add_component(MySource
, 'src')
49 sink
= graph
.add_component(MySink
, 'sink')
50 conn
= graph
.connect_ports(src
.output_ports
['out'],
51 sink
.input_ports
['in'])
52 self
.assertEqual(conn
.downstream_port
.addr
, sink
.input_ports
['in'].addr
)
54 def test_upstream_port(self
):
55 class MyIter(bt2
._UserMessageIterator
):
59 class MySource(bt2
._UserSourceComponent
,
60 message_iterator_class
=MyIter
):
61 def __init__(self
, params
):
62 self
._add
_output
_port
('out')
64 class MySink(bt2
._UserSinkComponent
):
65 def __init__(self
, params
):
66 self
._add
_input
_port
('in')
72 src
= graph
.add_component(MySource
, 'src')
73 sink
= graph
.add_component(MySink
, 'sink')
74 conn
= graph
.connect_ports(src
.output_ports
['out'],
75 sink
.input_ports
['in'])
76 self
.assertEqual(conn
.upstream_port
.addr
, src
.output_ports
['out'].addr
)
This page took 0.037284 seconds and 4 git commands to generate.