Commit | Line | Data |
---|---|---|
f6a5e476 | 1 | from bt2 import values |
0d0edb5e | 2 | import collections |
f6a5e476 PP |
3 | import unittest |
4 | import copy | |
5 | import bt2 | |
6 | ||
7 | ||
8 | class UserNotificationIteratorTestCase(unittest.TestCase): | |
9 | @staticmethod | |
10 | def _create_graph(src_comp_cls): | |
11 | class MySink(bt2._UserSinkComponent): | |
12 | def __init__(self, params): | |
13 | self._add_input_port('in') | |
14 | ||
15 | def _consume(self): | |
16 | next(self._notif_iter) | |
17 | ||
18 | def _port_connected(self, port, other_port): | |
19 | self._notif_iter = port.connection.create_notification_iterator() | |
20 | ||
21 | graph = bt2.Graph() | |
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']) | |
26 | return graph | |
27 | ||
28 | def test_init(self): | |
29 | class MyIter(bt2._UserNotificationIterator): | |
30 | def __init__(self): | |
31 | nonlocal initialized | |
32 | initialized = True | |
33 | ||
34 | class MySource(bt2._UserSourceComponent, | |
35 | notification_iterator_class=MyIter): | |
36 | def __init__(self, params): | |
37 | self._add_output_port('out') | |
38 | ||
39 | initialized = False | |
40 | graph = self._create_graph(MySource) | |
41 | self.assertTrue(initialized) | |
42 | ||
43 | def test_finalize(self): | |
44 | class MyIter(bt2._UserNotificationIterator): | |
45 | def _finalize(self): | |
46 | nonlocal finalized | |
47 | finalized = True | |
48 | ||
49 | class MySource(bt2._UserSourceComponent, | |
50 | notification_iterator_class=MyIter): | |
51 | def __init__(self, params): | |
52 | self._add_output_port('out') | |
53 | ||
54 | finalized = False | |
55 | graph = self._create_graph(MySource) | |
56 | del graph | |
57 | self.assertTrue(finalized) | |
58 | ||
59 | def test_component(self): | |
60 | class MyIter(bt2._UserNotificationIterator): | |
61 | def __init__(self): | |
62 | nonlocal salut | |
63 | salut = self._component._salut | |
64 | ||
65 | class MySource(bt2._UserSourceComponent, | |
66 | notification_iterator_class=MyIter): | |
67 | def __init__(self, params): | |
68 | self._add_output_port('out') | |
69 | self._salut = 23 | |
70 | ||
71 | salut = None | |
72 | graph = self._create_graph(MySource) | |
73 | self.assertEqual(salut, 23) | |
74 | ||
75 | def test_addr(self): | |
76 | class MyIter(bt2._UserNotificationIterator): | |
77 | def __init__(self): | |
78 | nonlocal addr | |
79 | addr = self.addr | |
80 | ||
81 | class MySource(bt2._UserSourceComponent, | |
82 | notification_iterator_class=MyIter): | |
83 | def __init__(self, params): | |
84 | self._add_output_port('out') | |
85 | ||
86 | addr = None | |
87 | graph = self._create_graph(MySource) | |
88 | self.assertIsNotNone(addr) | |
89 | self.assertNotEqual(addr, 0) | |
90 | ||
91 | ||
0d0edb5e | 92 | class PrivateConnectionNotificationIteratorTestCase(unittest.TestCase): |
f6a5e476 PP |
93 | def test_component(self): |
94 | class MyIter(bt2._UserNotificationIterator): | |
95 | pass | |
96 | ||
97 | class MySource(bt2._UserSourceComponent, | |
98 | notification_iterator_class=MyIter): | |
99 | def __init__(self, params): | |
100 | self._add_output_port('out') | |
101 | ||
102 | class MySink(bt2._UserSinkComponent): | |
103 | def __init__(self, params): | |
104 | self._add_input_port('in') | |
105 | ||
106 | def _consume(self): | |
107 | next(self._notif_iter) | |
108 | ||
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 | |
113 | ||
114 | upstream_comp = None | |
115 | graph = bt2.Graph() | |
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) | |
121 | del upstream_comp | |
0d0edb5e PP |
122 | |
123 | ||
124 | class OutputPortNotificationIteratorTestCase(unittest.TestCase): | |
125 | def test_component(self): | |
126 | class MyIter(bt2._UserNotificationIterator): | |
127 | def __init__(self): | |
128 | self._build_meta() | |
129 | self._at = 0 | |
130 | ||
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), | |
139 | ]) | |
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() | |
144 | ||
145 | def _create_event(self, value): | |
146 | ev = self._ec() | |
147 | ev.payload_field['my_int'] = value | |
148 | ev.packet = self._packet | |
149 | return ev | |
150 | ||
151 | def __next__(self): | |
152 | if self._at == 5: | |
153 | raise bt2.Stop | |
154 | ||
155 | notif = bt2.EventNotification(self._create_event(self._at * 3)) | |
156 | self._at += 1 | |
157 | return notif | |
158 | ||
159 | class MySource(bt2._UserSourceComponent, | |
160 | notification_iterator_class=MyIter): | |
161 | def __init__(self, params): | |
162 | self._add_output_port('out') | |
163 | ||
164 | graph = bt2.Graph() | |
165 | src = graph.add_component(MySource, 'src') | |
166 | types = [bt2.EventNotification] | |
167 | notif_iter = src.output_ports['out'].create_notification_iterator(types) | |
168 | ||
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) |