8 class UserNotificationIteratorTestCase(unittest
.TestCase
):
10 def _create_graph(src_comp_cls
):
11 class MySink(bt2
._UserSinkComponent
):
12 def __init__(self
, params
):
13 self
._add
_input
_port
('in')
16 next(self
._notif
_iter
)
18 def _port_connected(self
, port
, other_port
):
19 self
._notif
_iter
= port
.connection
.create_notification_iterator()
22 src_comp
= graph
.add_component(src_comp_cls
, 'src')
23 sink_comp
= graph
.add_component(MySink
, 'sink')
24 graph
.connect_ports(src_comp
.output_ports
['out'],
25 sink_comp
.input_ports
['in'])
29 class MyIter(bt2
._UserNotificationIterator
):
34 class MySource(bt2
._UserSourceComponent
,
35 notification_iterator_class
=MyIter
):
36 def __init__(self
, params
):
37 self
._add
_output
_port
('out')
40 graph
= self
._create
_graph
(MySource
)
41 self
.assertTrue(initialized
)
43 def test_finalize(self
):
44 class MyIter(bt2
._UserNotificationIterator
):
49 class MySource(bt2
._UserSourceComponent
,
50 notification_iterator_class
=MyIter
):
51 def __init__(self
, params
):
52 self
._add
_output
_port
('out')
55 graph
= self
._create
_graph
(MySource
)
57 self
.assertTrue(finalized
)
59 def test_component(self
):
60 class MyIter(bt2
._UserNotificationIterator
):
63 salut
= self
._component
._salut
65 class MySource(bt2
._UserSourceComponent
,
66 notification_iterator_class
=MyIter
):
67 def __init__(self
, params
):
68 self
._add
_output
_port
('out')
72 graph
= self
._create
_graph
(MySource
)
73 self
.assertEqual(salut
, 23)
76 class MyIter(bt2
._UserNotificationIterator
):
81 class MySource(bt2
._UserSourceComponent
,
82 notification_iterator_class
=MyIter
):
83 def __init__(self
, params
):
84 self
._add
_output
_port
('out')
87 graph
= self
._create
_graph
(MySource
)
88 self
.assertIsNotNone(addr
)
89 self
.assertNotEqual(addr
, 0)
92 class PrivateConnectionNotificationIteratorTestCase(unittest
.TestCase
):
93 def test_component(self
):
94 class MyIter(bt2
._UserNotificationIterator
):
97 class MySource(bt2
._UserSourceComponent
,
98 notification_iterator_class
=MyIter
):
99 def __init__(self
, params
):
100 self
._add
_output
_port
('out')
102 class MySink(bt2
._UserSinkComponent
):
103 def __init__(self
, params
):
104 self
._add
_input
_port
('in')
107 next(self
._notif
_iter
)
109 def _port_connected(self
, port
, other_port
):
110 nonlocal upstream_comp
111 self
._notif
_iter
= port
.connection
.create_notification_iterator()
112 upstream_comp
= self
._notif
_iter
.component
116 src_comp
= graph
.add_component(MySource
, 'src')
117 sink_comp
= graph
.add_component(MySink
, 'sink')
118 graph
.connect_ports(src_comp
.output_ports
['out'],
119 sink_comp
.input_ports
['in'])
120 self
.assertEqual(src_comp
, upstream_comp
)
124 class OutputPortNotificationIteratorTestCase(unittest
.TestCase
):
125 def test_component(self
):
126 class MyIter(bt2
._UserNotificationIterator
):
131 def _build_meta(self
):
132 self
._trace
= bt2
.Trace()
133 self
._sc
= bt2
.StreamClass()
134 self
._ec
= bt2
.EventClass('salut')
135 self
._my
_int
_ft
= bt2
.IntegerFieldType(32)
136 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
137 self
._ec
.payload_field_type
+= collections
.OrderedDict([
138 ('my_int', self
._my
_int
_ft
),
140 self
._sc
.add_event_class(self
._ec
)
141 self
._trace
.add_stream_class(self
._sc
)
142 self
._stream
= self
._sc
()
143 self
._packet
= self
._stream
.create_packet()
145 def _create_event(self
, value
):
147 ev
.payload_field
['my_int'] = value
148 ev
.packet
= self
._packet
155 notif
= bt2
.EventNotification(self
._create
_event
(self
._at
* 3))
159 class MySource(bt2
._UserSourceComponent
,
160 notification_iterator_class
=MyIter
):
161 def __init__(self
, params
):
162 self
._add
_output
_port
('out')
165 src
= graph
.add_component(MySource
, 'src')
166 types
= [bt2
.EventNotification
]
167 notif_iter
= src
.output_ports
['out'].create_notification_iterator(types
)
169 for at
, notif
in enumerate(notif_iter
):
170 self
.assertIsInstance(notif
, bt2
.EventNotification
)
171 self
.assertEqual(notif
.event
.event_class
.name
, 'salut')
172 field
= notif
.event
.payload_field
['my_int']
173 self
.assertEqual(field
, at
* 3)