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_027_disconnected_internal(self):
235 hb = gr.hier_block2("block",
236 gr.io_signature(1, 1, 1),
237 gr.io_signature(1, 1, 1))
238 src = gr.vector_source_b([1, ])
239 dst = gr.vector_sink_b()
240 tb.connect(src, hb, dst) # hb is not connected internally
241 self.assertRaises(RuntimeError,
244 def test_028_singleton_reconfigure(self):
246 hb = gr.hier_block2("block",
247 gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0))
248 src = gr.vector_source_b([1, ])
249 dst = gr.vector_sink_b()
251 tb.connect(hb) # Singleton connect
257 def test_029_singleton_disconnect(self):
259 src = gr.vector_source_b([1, ])
260 dst = gr.vector_sink_b()
262 tb.disconnect(src) # Singleton disconnect
265 self.assertEquals(dst.data(), (1,))
267 def test_030_nested_input(self):
269 src = gr.vector_source_b([1,])
270 hb1 = gr.hier_block2("hb1",
271 gr.io_signature(1, 1, gr.sizeof_char),
272 gr.io_signature(0, 0, 0))
273 hb2 = gr.hier_block2("hb2",
274 gr.io_signature(1, 1, gr.sizeof_char),
275 gr.io_signature(0, 0, 0))
276 dst = gr.vector_sink_b()
278 hb1.connect(hb1, hb2)
279 hb2.connect(hb2, gr.kludge_copy(gr.sizeof_char), dst)
281 self.assertEquals(dst.data(), (1,))
283 def test_031_multiple_internal_inputs(self):
285 src = gr.vector_source_f([1.0,])
286 hb = gr.hier_block2("hb",
287 gr.io_signature(1, 1, gr.sizeof_float),
288 gr.io_signature(1, 1, gr.sizeof_float))
289 m1 = gr.multiply_const_ff(1.0)
290 m2 = gr.multiply_const_ff(2.0)
292 hb.connect(hb, m1) # m1 is connected to hb external input #0
293 hb.connect(hb, m2) # m2 is also connected to hb external input #0
294 hb.connect(m1, (add, 0))
295 hb.connect(m2, (add, 1))
296 hb.connect(add, hb) # add is connected to hb external output #0
297 dst = gr.vector_sink_f()
298 tb.connect(src, hb, dst)
300 self.assertEquals(dst.data(), (3.0,))
302 def test_032_nested_multiple_internal_inputs(self):
304 src = gr.vector_source_f([1.0,])
305 hb = gr.hier_block2("hb",
306 gr.io_signature(1, 1, gr.sizeof_float),
307 gr.io_signature(1, 1, gr.sizeof_float))
308 hb2 = gr.hier_block2("hb",
309 gr.io_signature(1, 1, gr.sizeof_float),
310 gr.io_signature(1, 1, gr.sizeof_float))
312 m1 = gr.multiply_const_ff(1.0)
313 m2 = gr.multiply_const_ff(2.0)
315 hb2.connect(hb2, m1) # m1 is connected to hb2 external input #0
316 hb2.connect(hb2, m2) # m2 is also connected to hb2 external input #0
317 hb2.connect(m1, (add, 0))
318 hb2.connect(m2, (add, 1))
319 hb2.connect(add, hb2) # add is connected to hb2 external output #0
320 hb.connect(hb, hb2, hb) # hb as hb2 as nested internal block
321 dst = gr.vector_sink_f()
322 tb.connect(src, hb, dst)
324 self.assertEquals(dst.data(), (3.0,))
327 if __name__ == "__main__":