7 class UserNotificationIteratorTestCase(unittest
.TestCase
):
9 def _create_graph(src_comp_cls
):
10 class MySink(bt2
._UserSinkComponent
):
11 def __init__(self
, params
):
12 self
._add
_input
_port
('in')
15 next(self
._notif
_iter
)
17 def _port_connected(self
, port
, other_port
):
18 self
._notif
_iter
= port
.connection
.create_notification_iterator()
21 src_comp
= graph
.add_component(src_comp_cls
, 'src')
22 sink_comp
= graph
.add_component(MySink
, 'sink')
23 graph
.connect_ports(src_comp
.output_ports
['out'],
24 sink_comp
.input_ports
['in'])
28 class MyIter(bt2
._UserNotificationIterator
):
33 class MySource(bt2
._UserSourceComponent
,
34 notification_iterator_class
=MyIter
):
35 def __init__(self
, params
):
36 self
._add
_output
_port
('out')
39 graph
= self
._create
_graph
(MySource
)
40 self
.assertTrue(initialized
)
42 def test_finalize(self
):
43 class MyIter(bt2
._UserNotificationIterator
):
48 class MySource(bt2
._UserSourceComponent
,
49 notification_iterator_class
=MyIter
):
50 def __init__(self
, params
):
51 self
._add
_output
_port
('out')
54 graph
= self
._create
_graph
(MySource
)
56 self
.assertTrue(finalized
)
58 def test_component(self
):
59 class MyIter(bt2
._UserNotificationIterator
):
62 salut
= self
._component
._salut
64 class MySource(bt2
._UserSourceComponent
,
65 notification_iterator_class
=MyIter
):
66 def __init__(self
, params
):
67 self
._add
_output
_port
('out')
71 graph
= self
._create
_graph
(MySource
)
72 self
.assertEqual(salut
, 23)
75 class MyIter(bt2
._UserNotificationIterator
):
80 class MySource(bt2
._UserSourceComponent
,
81 notification_iterator_class
=MyIter
):
82 def __init__(self
, params
):
83 self
._add
_output
_port
('out')
86 graph
= self
._create
_graph
(MySource
)
87 self
.assertIsNotNone(addr
)
88 self
.assertNotEqual(addr
, 0)
91 class GenericNotificationIteratorTestCase(unittest
.TestCase
):
92 def test_component(self
):
93 class MyIter(bt2
._UserNotificationIterator
):
96 class MySource(bt2
._UserSourceComponent
,
97 notification_iterator_class
=MyIter
):
98 def __init__(self
, params
):
99 self
._add
_output
_port
('out')
101 class MySink(bt2
._UserSinkComponent
):
102 def __init__(self
, params
):
103 self
._add
_input
_port
('in')
106 next(self
._notif
_iter
)
108 def _port_connected(self
, port
, other_port
):
109 nonlocal upstream_comp
110 self
._notif
_iter
= port
.connection
.create_notification_iterator()
111 upstream_comp
= self
._notif
_iter
.component
115 src_comp
= graph
.add_component(MySource
, 'src')
116 sink_comp
= graph
.add_component(MySink
, 'sink')
117 graph
.connect_ports(src_comp
.output_ports
['out'],
118 sink_comp
.input_ports
['in'])
119 self
.assertEqual(src_comp
, upstream_comp
)