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_003_connect_input_in_use(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 nop2 = gr.nop(gr.sizeof_int)
36 hblock.connect(hblock, nop1)
37 self.assertRaises(ValueError,
38 lambda: hblock.connect(hblock, nop2))
40 def test_004_connect_output(self):
41 hblock = gr.hier_block2("test_block",
42 gr.io_signature(1,1,gr.sizeof_int),
43 gr.io_signature(1,1,gr.sizeof_int))
44 nop1 = gr.nop(gr.sizeof_int)
45 hblock.connect(nop1, hblock)
47 def test_005_connect_output_in_use(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 nop2 = gr.nop(gr.sizeof_int)
53 hblock.connect(nop1, hblock)
54 self.assertRaises(ValueError,
55 lambda: hblock.connect(nop2, hblock))
57 def test_006_connect_invalid_src_port_neg(self):
58 hblock = gr.hier_block2("test_block",
59 gr.io_signature(1,1,gr.sizeof_int),
60 gr.io_signature(1,1,gr.sizeof_int))
61 nop1 = gr.nop(gr.sizeof_int)
62 self.assertRaises(ValueError,
63 lambda: hblock.connect((hblock, -1), nop1))
65 def test_005_connect_invalid_src_port_exceeds(self):
66 hblock = gr.hier_block2("test_block",
67 gr.io_signature(1,1,gr.sizeof_int),
68 gr.io_signature(1,1,gr.sizeof_int))
69 nop1 = gr.nop(gr.sizeof_int)
70 self.assertRaises(ValueError,
71 lambda: hblock.connect((hblock, 1), nop1))
73 def test_007_connect_invalid_dst_port_neg(self):
74 hblock = gr.hier_block2("test_block",
75 gr.io_signature(1,1,gr.sizeof_int),
76 gr.io_signature(1,1,gr.sizeof_int))
77 nop1 = gr.nop(gr.sizeof_int)
78 nop2 = gr.nop(gr.sizeof_int)
79 self.assertRaises(ValueError,
80 lambda: hblock.connect(nop1, (nop2, -1)))
82 def test_008_connect_invalid_dst_port_exceeds(self):
83 hblock = gr.hier_block2("test_block",
84 gr.io_signature(1,1,gr.sizeof_int),
85 gr.io_signature(1,1,gr.sizeof_int))
86 nop1 = gr.null_sink(gr.sizeof_int)
87 nop2 = gr.null_sink(gr.sizeof_int)
88 self.assertRaises(ValueError,
89 lambda: hblock.connect(nop1, (nop2, 1)))
91 def test_009_check_topology(self):
92 hblock = gr.top_block("test_block")
93 hblock.check_topology(0, 0)
95 def test_010_run(self):
96 expected = (1.0, 2.0, 3.0, 4.0)
97 hblock = gr.top_block("test_block")
98 src = gr.vector_source_f(expected, False)
99 sink1 = gr.vector_sink_f()
100 sink2 = gr.vector_sink_f()
101 hblock.connect(src, sink1)
102 hblock.connect(src, sink2)
104 actual1 = sink1.data()
105 actual2 = sink2.data()
106 self.assertEquals(expected, actual1)
107 self.assertEquals(expected, actual2)
109 def test_012_disconnect_input(self):
110 hblock = gr.hier_block2("test_block",
111 gr.io_signature(1,1,gr.sizeof_int),
112 gr.io_signature(1,1,gr.sizeof_int))
113 nop1 = gr.nop(gr.sizeof_int)
114 hblock.connect(hblock, nop1)
115 hblock.disconnect(hblock, nop1)
117 def test_013_disconnect_input_not_connected(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 nop2 = gr.nop(gr.sizeof_int)
123 hblock.connect(hblock, nop1)
124 self.assertRaises(ValueError,
125 lambda: hblock.disconnect(hblock, nop2))
127 def test_014_disconnect_input_neg(self):
128 hblock = gr.hier_block2("test_block",
129 gr.io_signature(1,1,gr.sizeof_int),
130 gr.io_signature(1,1,gr.sizeof_int))
131 nop1 = gr.nop(gr.sizeof_int)
132 hblock.connect(hblock, nop1)
133 self.assertRaises(ValueError,
134 lambda: hblock.disconnect((hblock, -1), nop1))
136 def test_015_disconnect_input_exceeds(self):
137 hblock = gr.hier_block2("test_block",
138 gr.io_signature(1,1,gr.sizeof_int),
139 gr.io_signature(1,1,gr.sizeof_int))
140 nop1 = gr.nop(gr.sizeof_int)
141 hblock.connect(hblock, nop1)
142 self.assertRaises(ValueError,
143 lambda: hblock.disconnect((hblock, 1), nop1))
145 def test_016_disconnect_output(self):
146 hblock = gr.hier_block2("test_block",
147 gr.io_signature(1,1,gr.sizeof_int),
148 gr.io_signature(1,1,gr.sizeof_int))
149 nop1 = gr.nop(gr.sizeof_int)
150 hblock.connect(nop1, hblock)
151 hblock.disconnect(nop1, hblock)
153 def test_017_disconnect_output_not_connected(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 nop2 = gr.nop(gr.sizeof_int)
159 hblock.connect(nop1, hblock)
160 self.assertRaises(ValueError,
161 lambda: hblock.disconnect(nop2, hblock))
163 def test_018_disconnect_output_neg(self):
164 hblock = gr.hier_block2("test_block",
165 gr.io_signature(1,1,gr.sizeof_int),
166 gr.io_signature(1,1,gr.sizeof_int))
167 nop1 = gr.nop(gr.sizeof_int)
168 hblock.connect(hblock, nop1)
169 self.assertRaises(ValueError,
170 lambda: hblock.disconnect(nop1, (hblock, -1)))
172 def test_019_disconnect_output_exceeds(self):
173 hblock = gr.hier_block2("test_block",
174 gr.io_signature(1,1,gr.sizeof_int),
175 gr.io_signature(1,1,gr.sizeof_int))
176 nop1 = gr.nop(gr.sizeof_int)
177 hblock.connect(nop1, hblock)
178 self.assertRaises(ValueError,
179 lambda: hblock.disconnect(nop1, (hblock, 1)))
181 def test_020_run(self):
182 hblock = gr.top_block("test_block")
183 data = (1.0, 2.0, 3.0, 4.0)
184 src = gr.vector_source_f(data, False)
185 dst = gr.vector_sink_f()
186 hblock.connect(src, dst)
188 self.assertEquals(data, dst.data())
190 def test_021_connect_single(self):
191 hblock = gr.top_block("test_block")
192 blk = gr.hier_block2("block",
193 gr.io_signature(0, 0, 0),
194 gr.io_signature(0, 0, 0))
197 def test_022_connect_single_with_ports(self):
198 hblock = gr.top_block("test_block")
199 blk = gr.hier_block2("block",
200 gr.io_signature(1, 1, 1),
201 gr.io_signature(1, 1, 1))
202 self.assertRaises(ValueError,
203 lambda: hblock.connect(blk))
205 def test_023_connect_single_twice(self):
206 hblock = gr.top_block("test_block")
207 blk = gr.hier_block2("block",
208 gr.io_signature(0, 0, 0),
209 gr.io_signature(0, 0, 0))
211 self.assertRaises(ValueError,
212 lambda: hblock.connect(blk))
214 def test_024_disconnect_single(self):
215 hblock = gr.top_block("test_block")
216 blk = gr.hier_block2("block",
217 gr.io_signature(0, 0, 0),
218 gr.io_signature(0, 0, 0))
220 hblock.disconnect(blk)
222 def test_025_disconnect_single_not_connected(self):
223 hblock = gr.top_block("test_block")
224 blk = gr.hier_block2("block",
225 gr.io_signature(0, 0, 0),
226 gr.io_signature(0, 0, 0))
227 self.assertRaises(ValueError,
228 lambda: hblock.disconnect(blk))
230 def test_026_run_single(self):
231 expected_data = (1.0,)
232 tb = gr.top_block("top_block")
233 hb = gr.hier_block2("block",
234 gr.io_signature(0, 0, 0),
235 gr.io_signature(0, 0, 0))
236 src = gr.vector_source_f(expected_data)
237 dst = gr.vector_sink_f()
241 self.assertEquals(expected_data, dst.data())
243 if __name__ == "__main__":