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 | ||
90cfc012 | 8 | @unittest.skip("this is broken") |
f6a5e476 PP |
9 | class UserNotificationIteratorTestCase(unittest.TestCase): |
10 | @staticmethod | |
11 | def _create_graph(src_comp_cls): | |
12 | class MySink(bt2._UserSinkComponent): | |
13 | def __init__(self, params): | |
14 | self._add_input_port('in') | |
15 | ||
16 | def _consume(self): | |
17 | next(self._notif_iter) | |
18 | ||
19 | def _port_connected(self, port, other_port): | |
20 | self._notif_iter = port.connection.create_notification_iterator() | |
21 | ||
22 | graph = bt2.Graph() | |
23 | src_comp = graph.add_component(src_comp_cls, 'src') | |
24 | sink_comp = graph.add_component(MySink, 'sink') | |
25 | graph.connect_ports(src_comp.output_ports['out'], | |
26 | sink_comp.input_ports['in']) | |
27 | return graph | |
28 | ||
29 | def test_init(self): | |
30 | class MyIter(bt2._UserNotificationIterator): | |
31 | def __init__(self): | |
32 | nonlocal initialized | |
33 | initialized = True | |
34 | ||
35 | class MySource(bt2._UserSourceComponent, | |
36 | notification_iterator_class=MyIter): | |
37 | def __init__(self, params): | |
38 | self._add_output_port('out') | |
39 | ||
40 | initialized = False | |
41 | graph = self._create_graph(MySource) | |
42 | self.assertTrue(initialized) | |
43 | ||
44 | def test_finalize(self): | |
45 | class MyIter(bt2._UserNotificationIterator): | |
46 | def _finalize(self): | |
47 | nonlocal finalized | |
48 | finalized = True | |
49 | ||
50 | class MySource(bt2._UserSourceComponent, | |
51 | notification_iterator_class=MyIter): | |
52 | def __init__(self, params): | |
53 | self._add_output_port('out') | |
54 | ||
55 | finalized = False | |
56 | graph = self._create_graph(MySource) | |
57 | del graph | |
58 | self.assertTrue(finalized) | |
59 | ||
60 | def test_component(self): | |
61 | class MyIter(bt2._UserNotificationIterator): | |
62 | def __init__(self): | |
63 | nonlocal salut | |
64 | salut = self._component._salut | |
65 | ||
66 | class MySource(bt2._UserSourceComponent, | |
67 | notification_iterator_class=MyIter): | |
68 | def __init__(self, params): | |
69 | self._add_output_port('out') | |
70 | self._salut = 23 | |
71 | ||
72 | salut = None | |
73 | graph = self._create_graph(MySource) | |
74 | self.assertEqual(salut, 23) | |
75 | ||
76 | def test_addr(self): | |
77 | class MyIter(bt2._UserNotificationIterator): | |
78 | def __init__(self): | |
79 | nonlocal addr | |
80 | addr = self.addr | |
81 | ||
82 | class MySource(bt2._UserSourceComponent, | |
83 | notification_iterator_class=MyIter): | |
84 | def __init__(self, params): | |
85 | self._add_output_port('out') | |
86 | ||
87 | addr = None | |
88 | graph = self._create_graph(MySource) | |
89 | self.assertIsNotNone(addr) | |
90 | self.assertNotEqual(addr, 0) | |
91 | ||
92 | ||
90cfc012 | 93 | @unittest.skip("this is broken") |
0d0edb5e | 94 | class PrivateConnectionNotificationIteratorTestCase(unittest.TestCase): |
f6a5e476 PP |
95 | def test_component(self): |
96 | class MyIter(bt2._UserNotificationIterator): | |
97 | pass | |
98 | ||
99 | class MySource(bt2._UserSourceComponent, | |
100 | notification_iterator_class=MyIter): | |
101 | def __init__(self, params): | |
102 | self._add_output_port('out') | |
103 | ||
104 | class MySink(bt2._UserSinkComponent): | |
105 | def __init__(self, params): | |
106 | self._add_input_port('in') | |
107 | ||
108 | def _consume(self): | |
109 | next(self._notif_iter) | |
110 | ||
111 | def _port_connected(self, port, other_port): | |
112 | nonlocal upstream_comp | |
113 | self._notif_iter = port.connection.create_notification_iterator() | |
114 | upstream_comp = self._notif_iter.component | |
115 | ||
116 | upstream_comp = None | |
117 | graph = bt2.Graph() | |
118 | src_comp = graph.add_component(MySource, 'src') | |
119 | sink_comp = graph.add_component(MySink, 'sink') | |
120 | graph.connect_ports(src_comp.output_ports['out'], | |
121 | sink_comp.input_ports['in']) | |
122 | self.assertEqual(src_comp, upstream_comp) | |
123 | del upstream_comp | |
0d0edb5e PP |
124 | |
125 | ||
90cfc012 | 126 | @unittest.skip("this is broken") |
0d0edb5e PP |
127 | class OutputPortNotificationIteratorTestCase(unittest.TestCase): |
128 | def test_component(self): | |
129 | class MyIter(bt2._UserNotificationIterator): | |
130 | def __init__(self): | |
131 | self._build_meta() | |
132 | self._at = 0 | |
133 | ||
134 | def _build_meta(self): | |
135 | self._trace = bt2.Trace() | |
136 | self._sc = bt2.StreamClass() | |
137 | self._ec = bt2.EventClass('salut') | |
138 | self._my_int_ft = bt2.IntegerFieldType(32) | |
139 | self._ec.payload_field_type = bt2.StructureFieldType() | |
140 | self._ec.payload_field_type += collections.OrderedDict([ | |
141 | ('my_int', self._my_int_ft), | |
142 | ]) | |
143 | self._sc.add_event_class(self._ec) | |
144 | self._trace.add_stream_class(self._sc) | |
145 | self._stream = self._sc() | |
146 | self._packet = self._stream.create_packet() | |
147 | ||
148 | def _create_event(self, value): | |
149 | ev = self._ec() | |
150 | ev.payload_field['my_int'] = value | |
151 | ev.packet = self._packet | |
152 | return ev | |
153 | ||
154 | def __next__(self): | |
155 | if self._at == 5: | |
156 | raise bt2.Stop | |
157 | ||
158 | notif = bt2.EventNotification(self._create_event(self._at * 3)) | |
159 | self._at += 1 | |
160 | return notif | |
161 | ||
162 | class MySource(bt2._UserSourceComponent, | |
163 | notification_iterator_class=MyIter): | |
164 | def __init__(self, params): | |
165 | self._add_output_port('out') | |
166 | ||
167 | graph = bt2.Graph() | |
168 | src = graph.add_component(MySource, 'src') | |
169 | types = [bt2.EventNotification] | |
170 | notif_iter = src.output_ports['out'].create_notification_iterator(types) | |
171 | ||
172 | for at, notif in enumerate(notif_iter): | |
173 | self.assertIsInstance(notif, bt2.EventNotification) | |
174 | self.assertEqual(notif.event.event_class.name, 'salut') | |
175 | field = notif.event.payload_field['my_int'] | |
176 | self.assertEqual(field, at * 3) |