7 class ConnectionTestCase(unittest
.TestCase
):
9 class MyIter(bt2
._UserNotificationIterator
):
13 class MySource(bt2
._UserSourceComponent
,
14 notification_iterator_class
=MyIter
):
15 def __init__(self
, params
):
16 self
._add
_output
_port
('out')
18 class MySink(bt2
._UserSinkComponent
):
19 def __init__(self
, params
):
20 self
._add
_input
_port
('in')
26 src
= graph
.add_component(MySource
, 'src')
27 sink
= graph
.add_component(MySink
, 'sink')
28 conn
= graph
.connect_ports(src
.output_ports
['out'],
29 sink
.input_ports
['in'])
30 self
.assertIsInstance(conn
, bt2
._Connection
)
31 self
.assertNotIsInstance(conn
, bt2
._PrivateConnection
)
33 def test_is_ended_false(self
):
34 class MyIter(bt2
._UserNotificationIterator
):
38 class MySource(bt2
._UserSourceComponent
,
39 notification_iterator_class
=MyIter
):
40 def __init__(self
, params
):
41 self
._add
_output
_port
('out')
43 class MySink(bt2
._UserSinkComponent
):
44 def __init__(self
, params
):
45 self
._add
_input
_port
('in')
51 src
= graph
.add_component(MySource
, 'src')
52 sink
= graph
.add_component(MySink
, 'sink')
53 conn
= graph
.connect_ports(src
.output_ports
['out'],
54 sink
.input_ports
['in'])
55 self
.assertFalse(conn
.is_ended
)
57 def test_is_ended_true(self
):
58 class MyIter(bt2
._UserNotificationIterator
):
62 class MySource(bt2
._UserSourceComponent
,
63 notification_iterator_class
=MyIter
):
64 def __init__(self
, params
):
65 self
._add
_output
_port
('out')
67 class MySink(bt2
._UserSinkComponent
):
68 def __init__(self
, params
):
69 self
._add
_input
_port
('in')
75 src
= graph
.add_component(MySource
, 'src')
76 sink
= graph
.add_component(MySink
, 'sink')
77 conn
= graph
.connect_ports(src
.output_ports
['out'],
78 sink
.input_ports
['in'])
79 src
.output_ports
['out'].disconnect()
80 self
.assertTrue(conn
.is_ended
)
82 def test_downstream_port(self
):
83 class MyIter(bt2
._UserNotificationIterator
):
87 class MySource(bt2
._UserSourceComponent
,
88 notification_iterator_class
=MyIter
):
89 def __init__(self
, params
):
90 self
._add
_output
_port
('out')
92 class MySink(bt2
._UserSinkComponent
):
93 def __init__(self
, params
):
94 self
._add
_input
_port
('in')
100 src
= graph
.add_component(MySource
, 'src')
101 sink
= graph
.add_component(MySink
, 'sink')
102 conn
= graph
.connect_ports(src
.output_ports
['out'],
103 sink
.input_ports
['in'])
104 self
.assertEqual(conn
.downstream_port
.addr
, sink
.input_ports
['in'].addr
)
106 def test_upstream_port(self
):
107 class MyIter(bt2
._UserNotificationIterator
):
111 class MySource(bt2
._UserSourceComponent
,
112 notification_iterator_class
=MyIter
):
113 def __init__(self
, params
):
114 self
._add
_output
_port
('out')
116 class MySink(bt2
._UserSinkComponent
):
117 def __init__(self
, params
):
118 self
._add
_input
_port
('in')
124 src
= graph
.add_component(MySource
, 'src')
125 sink
= graph
.add_component(MySink
, 'sink')
126 conn
= graph
.connect_ports(src
.output_ports
['out'],
127 sink
.input_ports
['in'])
128 self
.assertEqual(conn
.upstream_port
.addr
, src
.output_ports
['out'].addr
)
131 class MyIter(bt2
._UserNotificationIterator
):
135 class MySource(bt2
._UserSourceComponent
,
136 notification_iterator_class
=MyIter
):
137 def __init__(self
, params
):
138 self
._add
_output
_port
('out')
140 class MySink(bt2
._UserSinkComponent
):
141 def __init__(self
, params
):
142 self
._add
_input
_port
('in')
148 src
= graph
.add_component(MySource
, 'src')
149 sink
= graph
.add_component(MySink
, 'sink')
150 conn
= graph
.connect_ports(src
.output_ports
['out'],
151 sink
.input_ports
['in'])
152 self
.assertEqual(conn
, conn
)
154 def test_eq_invalid(self
):
155 class MyIter(bt2
._UserNotificationIterator
):
159 class MySource(bt2
._UserSourceComponent
,
160 notification_iterator_class
=MyIter
):
161 def __init__(self
, params
):
162 self
._add
_output
_port
('out')
164 class MySink(bt2
._UserSinkComponent
):
165 def __init__(self
, params
):
166 self
._add
_input
_port
('in')
172 src
= graph
.add_component(MySource
, 'src')
173 sink
= graph
.add_component(MySink
, 'sink')
174 conn
= graph
.connect_ports(src
.output_ports
['out'],
175 sink
.input_ports
['in'])
176 self
.assertNotEqual(conn
, 23)
179 class PrivateConnectionTestCase(unittest
.TestCase
):
180 def test_create(self
):
181 class MyIter(bt2
._UserNotificationIterator
):
185 class MySource(bt2
._UserSourceComponent
,
186 notification_iterator_class
=MyIter
):
187 def __init__(self
, params
):
188 self
._add
_output
_port
('out')
190 class MySink(bt2
._UserSinkComponent
):
191 def __init__(self
, params
):
192 self
._add
_input
_port
('in')
197 def _port_connected(self
, port
, other_port
):
199 priv_conn
= port
.connection
203 src
= graph
.add_component(MySource
, 'src')
204 sink
= graph
.add_component(MySink
, 'sink')
205 conn
= graph
.connect_ports(src
.output_ports
['out'],
206 sink
.input_ports
['in'])
207 self
.assertIsInstance(priv_conn
, bt2
._PrivateConnection
)
208 self
.assertEqual(conn
, priv_conn
)
211 def test_is_ended_false(self
):
212 class MyIter(bt2
._UserNotificationIterator
):
216 class MySource(bt2
._UserSourceComponent
,
217 notification_iterator_class
=MyIter
):
218 def __init__(self
, params
):
219 self
._add
_output
_port
('out')
221 class MySink(bt2
._UserSinkComponent
):
222 def __init__(self
, params
):
223 self
._add
_input
_port
('in')
228 def _port_connected(self
, port
, other_port
):
230 priv_conn
= port
.connection
234 src
= graph
.add_component(MySource
, 'src')
235 sink
= graph
.add_component(MySink
, 'sink')
236 conn
= graph
.connect_ports(src
.output_ports
['out'],
237 sink
.input_ports
['in'])
238 self
.assertFalse(priv_conn
.is_ended
)
241 def test_is_ended_true(self
):
242 class MyIter(bt2
._UserNotificationIterator
):
246 class MySource(bt2
._UserSourceComponent
,
247 notification_iterator_class
=MyIter
):
248 def __init__(self
, params
):
249 self
._add
_output
_port
('out')
251 class MySink(bt2
._UserSinkComponent
):
252 def __init__(self
, params
):
253 self
._add
_input
_port
('in')
258 def _port_connected(self
, port
, other_port
):
260 priv_conn
= port
.connection
264 src
= graph
.add_component(MySource
, 'src')
265 sink
= graph
.add_component(MySink
, 'sink')
266 conn
= graph
.connect_ports(src
.output_ports
['out'],
267 sink
.input_ports
['in'])
268 sink
.input_ports
['in'].disconnect()
269 self
.assertTrue(priv_conn
.is_ended
)
272 def test_downstream_port(self
):
273 class MyIter(bt2
._UserNotificationIterator
):
277 class MySource(bt2
._UserSourceComponent
,
278 notification_iterator_class
=MyIter
):
279 def __init__(self
, params
):
280 self
._add
_output
_port
('out')
282 class MySink(bt2
._UserSinkComponent
):
283 def __init__(self
, params
):
284 self
._add
_input
_port
('in')
289 def _port_connected(self
, port
, other_port
):
291 priv_conn
= port
.connection
292 priv_port
= priv_conn
.downstream_port
296 src
= graph
.add_component(MySource
, 'src')
297 sink
= graph
.add_component(MySink
, 'sink')
298 conn
= graph
.connect_ports(src
.output_ports
['out'],
299 sink
.input_ports
['in'])
300 self
.assertEqual(priv_port
.addr
, sink
.input_ports
['in'].addr
)
303 def test_upstream_port(self
):
304 class MyIter(bt2
._UserNotificationIterator
):
308 class MySource(bt2
._UserSourceComponent
,
309 notification_iterator_class
=MyIter
):
310 def __init__(self
, params
):
311 self
._add
_output
_port
('out')
313 class MySink(bt2
._UserSinkComponent
):
314 def __init__(self
, params
):
315 self
._add
_input
_port
('in')
320 def _port_connected(self
, port
, other_port
):
322 priv_conn
= port
.connection
323 priv_port
= priv_conn
.upstream_port
327 src
= graph
.add_component(MySource
, 'src')
328 sink
= graph
.add_component(MySink
, 'sink')
329 conn
= graph
.connect_ports(src
.output_ports
['out'],
330 sink
.input_ports
['in'])
331 self
.assertEqual(priv_port
.addr
, src
.output_ports
['out'].addr
)
335 class MyIter(bt2
._UserNotificationIterator
):
339 class MySource(bt2
._UserSourceComponent
,
340 notification_iterator_class
=MyIter
):
341 def __init__(self
, params
):
342 self
._add
_output
_port
('out')
344 class MySink(bt2
._UserSinkComponent
):
345 def __init__(self
, params
):
346 self
._add
_input
_port
('in')
351 def _port_connected(self
, port
, other_port
):
353 priv_conn
= port
.connection
357 src
= graph
.add_component(MySource
, 'src')
358 sink
= graph
.add_component(MySink
, 'sink')
359 conn
= graph
.connect_ports(src
.output_ports
['out'],
360 sink
.input_ports
['in'])
361 self
.assertEqual(priv_conn
, conn
)
364 def test_eq_invalid(self
):
365 class MyIter(bt2
._UserNotificationIterator
):
369 class MySource(bt2
._UserSourceComponent
,
370 notification_iterator_class
=MyIter
):
371 def __init__(self
, params
):
372 self
._add
_output
_port
('out')
374 class MySink(bt2
._UserSinkComponent
):
375 def __init__(self
, params
):
376 self
._add
_input
_port
('in')
381 def _port_connected(self
, port
, other_port
):
383 priv_conn
= port
.connection
387 src
= graph
.add_component(MySource
, 'src')
388 sink
= graph
.add_component(MySink
, 'sink')
389 conn
= graph
.connect_ports(src
.output_ports
['out'],
390 sink
.input_ports
['in'])
391 self
.assertNotEqual(priv_conn
, 23)