3 from gnuradio import gr, gr_unittest
5 class test_hier_block2(gr_unittest.TestCase):
13 def test_001_make(self):
14 hblock = gr.hier_block2("test_block",
15 gr.io_signature(1,1,gr.sizeof_int),
16 gr.io_signature(1,1,gr.sizeof_int))
17 self.assertEqual("test_block", hblock.name())
18 self.assertEqual(1, hblock.input_signature().max_streams())
19 self.assertEqual(1, hblock.output_signature().min_streams())
20 self.assertEqual(1, hblock.output_signature().max_streams())
21 self.assertEqual(gr.sizeof_int, hblock.output_signature().sizeof_stream_item(0))
23 def test_002_connect_input(self):
24 hblock = gr.hier_block2("test_block",
25 gr.io_signature(1,1,gr.sizeof_int),
26 gr.io_signature(1,1,gr.sizeof_int))
27 nop1 = gr.nop(gr.sizeof_int)
28 hblock.connect(hblock, nop1)
30 def test_004_connect_output(self):
31 hblock = gr.hier_block2("test_block",
32 gr.io_signature(1,1,gr.sizeof_int),
33 gr.io_signature(1,1,gr.sizeof_int))
34 nop1 = gr.nop(gr.sizeof_int)
35 hblock.connect(nop1, hblock)
37 def test_005_connect_output_in_use(self):
38 hblock = gr.hier_block2("test_block",
39 gr.io_signature(1,1,gr.sizeof_int),
40 gr.io_signature(1,1,gr.sizeof_int))
41 nop1 = gr.nop(gr.sizeof_int)
42 nop2 = gr.nop(gr.sizeof_int)
43 hblock.connect(nop1, hblock)
44 self.assertRaises(ValueError,
45 lambda: hblock.connect(nop2, hblock))
47 def test_006_connect_invalid_src_port_neg(self):
48 hblock = gr.hier_block2("test_block",
49 gr.io_signature(1,1,gr.sizeof_int),
50 gr.io_signature(1,1,gr.sizeof_int))
51 nop1 = gr.nop(gr.sizeof_int)
52 self.assertRaises(ValueError,
53 lambda: hblock.connect((hblock, -1), nop1))
55 def test_005_connect_invalid_src_port_exceeds(self):
56 hblock = gr.hier_block2("test_block",
57 gr.io_signature(1,1,gr.sizeof_int),
58 gr.io_signature(1,1,gr.sizeof_int))
59 nop1 = gr.nop(gr.sizeof_int)
60 self.assertRaises(ValueError,
61 lambda: hblock.connect((hblock, 1), nop1))
63 def test_007_connect_invalid_dst_port_neg(self):
64 hblock = gr.hier_block2("test_block",
65 gr.io_signature(1,1,gr.sizeof_int),
66 gr.io_signature(1,1,gr.sizeof_int))
67 nop1 = gr.nop(gr.sizeof_int)
68 nop2 = gr.nop(gr.sizeof_int)
69 self.assertRaises(ValueError,
70 lambda: hblock.connect(nop1, (nop2, -1)))
72 def test_008_connect_invalid_dst_port_exceeds(self):
73 hblock = gr.hier_block2("test_block",
74 gr.io_signature(1,1,gr.sizeof_int),
75 gr.io_signature(1,1,gr.sizeof_int))
76 nop1 = gr.null_sink(gr.sizeof_int)
77 nop2 = gr.null_sink(gr.sizeof_int)
78 self.assertRaises(ValueError,
79 lambda: hblock.connect(nop1, (nop2, 1)))
81 def test_009_check_topology(self):
82 hblock = gr.top_block("test_block")
83 hblock.check_topology(0, 0)
85 def test_010_run(self):
86 expected = (1.0, 2.0, 3.0, 4.0)
87 hblock = gr.top_block("test_block")
88 src = gr.vector_source_f(expected, False)
89 sink1 = gr.vector_sink_f()
90 sink2 = gr.vector_sink_f()
91 hblock.connect(src, sink1)
92 hblock.connect(src, sink2)
94 actual1 = sink1.data()
95 actual2 = sink2.data()
96 self.assertEquals(expected, actual1)
97 self.assertEquals(expected, actual2)
99 def test_012_disconnect_input(self):
100 hblock = gr.hier_block2("test_block",
101 gr.io_signature(1,1,gr.sizeof_int),
102 gr.io_signature(1,1,gr.sizeof_int))
103 nop1 = gr.nop(gr.sizeof_int)
104 hblock.connect(hblock, nop1)
105 hblock.disconnect(hblock, nop1)
107 def test_013_disconnect_input_not_connected(self):
108 hblock = gr.hier_block2("test_block",
109 gr.io_signature(1,1,gr.sizeof_int),
110 gr.io_signature(1,1,gr.sizeof_int))
111 nop1 = gr.nop(gr.sizeof_int)
112 nop2 = gr.nop(gr.sizeof_int)
113 hblock.connect(hblock, nop1)
114 self.assertRaises(ValueError,
115 lambda: hblock.disconnect(hblock, nop2))
117 def test_014_disconnect_input_neg(self):
118 hblock = gr.hier_block2("test_block",
119 gr.io_signature(1,1,gr.sizeof_int),
120 gr.io_signature(1,1,gr.sizeof_int))
121 nop1 = gr.nop(gr.sizeof_int)
122 hblock.connect(hblock, nop1)
123 self.assertRaises(ValueError,
124 lambda: hblock.disconnect((hblock, -1), nop1))
126 def test_015_disconnect_input_exceeds(self):
127 hblock = gr.hier_block2("test_block",
128 gr.io_signature(1,1,gr.sizeof_int),
129 gr.io_signature(1,1,gr.sizeof_int))
130 nop1 = gr.nop(gr.sizeof_int)
131 hblock.connect(hblock, nop1)
132 self.assertRaises(ValueError,
133 lambda: hblock.disconnect((hblock, 1), nop1))
135 def test_016_disconnect_output(self):
136 hblock = gr.hier_block2("test_block",
137 gr.io_signature(1,1,gr.sizeof_int),
138 gr.io_signature(1,1,gr.sizeof_int))
139 nop1 = gr.nop(gr.sizeof_int)
140 hblock.connect(nop1, hblock)
141 hblock.disconnect(nop1, hblock)
143 def test_017_disconnect_output_not_connected(self):
144 hblock = gr.hier_block2("test_block",
145 gr.io_signature(1,1,gr.sizeof_int),
146 gr.io_signature(1,1,gr.sizeof_int))
147 nop1 = gr.nop(gr.sizeof_int)
148 nop2 = gr.nop(gr.sizeof_int)
149 hblock.connect(nop1, hblock)
150 self.assertRaises(ValueError,
151 lambda: hblock.disconnect(nop2, hblock))
153 def test_018_disconnect_output_neg(self):
154 hblock = gr.hier_block2("test_block",
155 gr.io_signature(1,1,gr.sizeof_int),
156 gr.io_signature(1,1,gr.sizeof_int))
157 nop1 = gr.nop(gr.sizeof_int)
158 hblock.connect(hblock, nop1)
159 self.assertRaises(ValueError,
160 lambda: hblock.disconnect(nop1, (hblock, -1)))
162 def test_019_disconnect_output_exceeds(self):
163 hblock = gr.hier_block2("test_block",
164 gr.io_signature(1,1,gr.sizeof_int),
165 gr.io_signature(1,1,gr.sizeof_int))
166 nop1 = gr.nop(gr.sizeof_int)
167 hblock.connect(nop1, hblock)
168 self.assertRaises(ValueError,
169 lambda: hblock.disconnect(nop1, (hblock, 1)))
171 def test_020_run(self):
172 hblock = gr.top_block("test_block")
173 data = (1.0, 2.0, 3.0, 4.0)
174 src = gr.vector_source_f(data, False)
175 dst = gr.vector_sink_f()
176 hblock.connect(src, dst)
178 self.assertEquals(data, dst.data())
180 def test_021_connect_single(self):
181 hblock = gr.top_block("test_block")
182 blk = gr.hier_block2("block",
183 gr.io_signature(0, 0, 0),
184 gr.io_signature(0, 0, 0))
187 def test_022_connect_single_with_ports(self):
188 hblock = gr.top_block("test_block")
189 blk = gr.hier_block2("block",
190 gr.io_signature(1, 1, 1),
191 gr.io_signature(1, 1, 1))
192 self.assertRaises(ValueError,
193 lambda: hblock.connect(blk))
195 def test_023_connect_single_twice(self):
196 hblock = gr.top_block("test_block")
197 blk = gr.hier_block2("block",
198 gr.io_signature(0, 0, 0),
199 gr.io_signature(0, 0, 0))
201 self.assertRaises(ValueError,
202 lambda: hblock.connect(blk))
204 def test_024_disconnect_single(self):
205 hblock = gr.top_block("test_block")
206 blk = gr.hier_block2("block",
207 gr.io_signature(0, 0, 0),
208 gr.io_signature(0, 0, 0))
210 hblock.disconnect(blk)
212 def test_025_disconnect_single_not_connected(self):
213 hblock = gr.top_block("test_block")
214 blk = gr.hier_block2("block",
215 gr.io_signature(0, 0, 0),
216 gr.io_signature(0, 0, 0))
217 self.assertRaises(ValueError,
218 lambda: hblock.disconnect(blk))
220 def test_026_run_single(self):
221 expected_data = (1.0,)
222 tb = gr.top_block("top_block")
223 hb = gr.hier_block2("block",
224 gr.io_signature(0, 0, 0),
225 gr.io_signature(0, 0, 0))
226 src = gr.vector_source_f(expected_data)
227 dst = gr.vector_sink_f()
231 self.assertEquals(expected_data, dst.data())
233 def test_027a_internally_unconnected_input(self):
235 hb = gr.hier_block2("block",
236 gr.io_signature(1, 1, 1),
237 gr.io_signature(1, 1, 1))
238 hsrc = gr.vector_source_b([1,])
239 hb.connect(hsrc, hb) # wire output internally
240 src = gr.vector_source_b([1, ])
241 dst = gr.vector_sink_b()
242 tb.connect(src, hb, dst) # hb's input is not connected internally
243 self.assertRaises(RuntimeError,
246 def test_027b_internally_unconnected_output(self):
249 hb = gr.hier_block2("block",
250 gr.io_signature(1, 1, 1),
251 gr.io_signature(1, 1, 1))
252 hdst = gr.vector_sink_b()
253 hb.connect(hb, hdst) # wire input internally
254 src = gr.vector_source_b([1, ])
255 dst = gr.vector_sink_b()
256 tb.connect(src, hb, dst) # hb's output is not connected internally
257 self.assertRaises(RuntimeError,
260 def test_027c_fully_unconnected_output(self):
262 hb = gr.hier_block2("block",
263 gr.io_signature(1, 1, 1),
264 gr.io_signature(1, 1, 1))
265 hsrc = gr.vector_sink_b()
266 hb.connect(hb, hsrc) # wire input internally
267 src = gr.vector_source_b([1, ])
268 dst = gr.vector_sink_b()
269 tb.connect(src, hb) # hb's output is not connected internally or externally
270 self.assertRaises(RuntimeError,
273 def test_027d_fully_unconnected_input(self):
275 hb = gr.hier_block2("block",
276 gr.io_signature(1, 1, 1),
277 gr.io_signature(1, 1, 1))
278 hdst = gr.vector_source_b([1,])
279 hb.connect(hdst, hb) # wire output internally
280 dst = gr.vector_sink_b()
281 tb.connect(hb, dst) # hb's input is not connected internally or externally
282 self.assertRaises(RuntimeError,
285 def test_028_singleton_reconfigure(self):
287 hb = gr.hier_block2("block",
288 gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0))
289 src = gr.vector_source_b([1, ])
290 dst = gr.vector_sink_b()
292 tb.connect(hb) # Singleton connect
298 def test_029_singleton_disconnect(self):
300 src = gr.vector_source_b([1, ])
301 dst = gr.vector_sink_b()
303 tb.disconnect(src) # Singleton disconnect
306 self.assertEquals(dst.data(), (1,))
308 def test_030_nested_input(self):
310 src = gr.vector_source_b([1,])
311 hb1 = gr.hier_block2("hb1",
312 gr.io_signature(1, 1, gr.sizeof_char),
313 gr.io_signature(0, 0, 0))
314 hb2 = gr.hier_block2("hb2",
315 gr.io_signature(1, 1, gr.sizeof_char),
316 gr.io_signature(0, 0, 0))
317 dst = gr.vector_sink_b()
319 hb1.connect(hb1, hb2)
320 hb2.connect(hb2, gr.kludge_copy(gr.sizeof_char), dst)
322 self.assertEquals(dst.data(), (1,))
324 def test_031_multiple_internal_inputs(self):
326 src = gr.vector_source_f([1.0,])
327 hb = gr.hier_block2("hb",
328 gr.io_signature(1, 1, gr.sizeof_float),
329 gr.io_signature(1, 1, gr.sizeof_float))
330 m1 = gr.multiply_const_ff(1.0)
331 m2 = gr.multiply_const_ff(2.0)
333 hb.connect(hb, m1) # m1 is connected to hb external input #0
334 hb.connect(hb, m2) # m2 is also connected to hb external input #0
335 hb.connect(m1, (add, 0))
336 hb.connect(m2, (add, 1))
337 hb.connect(add, hb) # add is connected to hb external output #0
338 dst = gr.vector_sink_f()
339 tb.connect(src, hb, dst)
341 self.assertEquals(dst.data(), (3.0,))
343 def test_032_nested_multiple_internal_inputs(self):
345 src = gr.vector_source_f([1.0,])
346 hb = gr.hier_block2("hb",
347 gr.io_signature(1, 1, gr.sizeof_float),
348 gr.io_signature(1, 1, gr.sizeof_float))
349 hb2 = gr.hier_block2("hb",
350 gr.io_signature(1, 1, gr.sizeof_float),
351 gr.io_signature(1, 1, gr.sizeof_float))
353 m1 = gr.multiply_const_ff(1.0)
354 m2 = gr.multiply_const_ff(2.0)
356 hb2.connect(hb2, m1) # m1 is connected to hb2 external input #0
357 hb2.connect(hb2, m2) # m2 is also connected to hb2 external input #0
358 hb2.connect(m1, (add, 0))
359 hb2.connect(m2, (add, 1))
360 hb2.connect(add, hb2) # add is connected to hb2 external output #0
361 hb.connect(hb, hb2, hb) # hb as hb2 as nested internal block
362 dst = gr.vector_sink_f()
363 tb.connect(src, hb, dst)
365 self.assertEquals(dst.data(), (3.0,))
368 if __name__ == "__main__":