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_001_connect_internal(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 nop2 = gr.nop(gr.sizeof_int)
29 hblock.connect(nop1, nop2)
31 def test_002_connect_input(self):
32 hblock = gr.hier_block2("test_block",
33 gr.io_signature(1,1,gr.sizeof_int),
34 gr.io_signature(1,1,gr.sizeof_int))
35 nop1 = gr.nop(gr.sizeof_int)
36 hblock.connect(hblock, nop1)
38 def test_002a_connect_input_in_use(self):
39 hblock = gr.hier_block2("test_block",
40 gr.io_signature(1,1,gr.sizeof_int),
41 gr.io_signature(1,1,gr.sizeof_int))
42 nop1 = gr.nop(gr.sizeof_int)
43 nop2 = gr.nop(gr.sizeof_int)
44 hblock.connect(hblock, nop1)
45 self.assertRaises(ValueError,
46 lambda: hblock.connect(hblock, nop2))
48 def test_003_connect_output(self):
49 hblock = gr.hier_block2("test_block",
50 gr.io_signature(1,1,gr.sizeof_int),
51 gr.io_signature(1,1,gr.sizeof_int))
52 nop1 = gr.nop(gr.sizeof_int)
53 hblock.connect(nop1, hblock)
55 def test_003a_connect_output_in_use(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 nop2 = gr.nop(gr.sizeof_int)
61 hblock.connect(nop1, hblock)
62 self.assertRaises(ValueError,
63 lambda: hblock.connect(nop2, hblock))
65 def test_004_connect_invalid_src_port_neg(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_005_connect_invalid_src_port_exceeds(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 self.assertRaises(ValueError,
79 lambda: hblock.connect((hblock, 1), nop1))
81 def test_006_connect_invalid_dst_port_neg(self):
82 hblock = gr.hier_block2("test_block",
83 gr.io_signature(1,1,gr.sizeof_int),
84 gr.io_signature(1,1,gr.sizeof_int))
85 nop1 = gr.nop(gr.sizeof_int)
86 nop2 = gr.nop(gr.sizeof_int)
87 self.assertRaises(ValueError,
88 lambda: hblock.connect(nop1, (nop2, -1)))
90 def test_007_connect_invalid_dst_port_exceeds(self):
91 hblock = gr.hier_block2("test_block",
92 gr.io_signature(1,1,gr.sizeof_int),
93 gr.io_signature(1,1,gr.sizeof_int))
94 nop1 = gr.null_sink(gr.sizeof_int)
95 nop2 = gr.null_sink(gr.sizeof_int)
96 self.assertRaises(ValueError,
97 lambda: hblock.connect(nop1, (nop2, 1)))
99 def test_008_connect_dst_port_in_use(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 nop2 = gr.nop(gr.sizeof_int)
105 hblock.connect(nop1, nop2);
106 self.assertRaises(ValueError,
107 lambda: hblock.connect(nop1, nop2))
109 def test_009_connect_one_src_two_dst(self):
110 hblock = gr.top_block("test_block")
111 src = gr.null_source(gr.sizeof_int)
112 dst1 = gr.null_sink(gr.sizeof_int)
113 dst2 = gr.null_sink(gr.sizeof_int)
114 hblock.connect(src, dst1)
115 hblock.connect(src, dst2)
117 def test_010_connect_type_mismatch(self):
118 hblock = gr.top_block("test_block")
119 nop1 = gr.nop(gr.sizeof_char)
120 nop2 = gr.nop(gr.sizeof_int)
121 self.assertRaises(ValueError,
122 lambda: hblock.connect(nop1, nop2))
124 def test_011_check_topology(self):
125 hblock = gr.top_block("test_block")
126 hblock.check_topology(0, 0)
128 def test_012_disconnect(self):
129 hblock = gr.top_block("test_block")
130 nop1 = gr.nop(gr.sizeof_int)
131 nop2 = gr.nop(gr.sizeof_int)
132 hblock.connect(nop1, nop2)
133 hblock.disconnect(nop1, nop2)
135 def test_013_disconnect_not_connected(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 nop2 = gr.nop(gr.sizeof_int)
141 nop3 = gr.nop(gr.sizeof_int)
142 hblock.connect(nop1, nop2)
143 self.assertRaises(ValueError,
144 lambda: hblock.disconnect(nop1, nop3))
146 def test_014_run(self):
147 expected = (1.0, 2.0, 3.0, 4.0)
148 hblock = gr.top_block("test_block")
149 src = gr.vector_source_f(expected, False)
150 sink1 = gr.vector_sink_f()
151 sink2 = gr.vector_sink_f()
152 hblock.connect(src, sink1)
153 hblock.connect(src, sink2)
154 runtime = gr.runtime(hblock)
156 actual1 = sink1.data()
157 actual2 = sink2.data()
158 self.assertEquals(expected, actual1)
159 self.assertEquals(expected, actual2)
161 def test_015_disconnect_input(self):
162 hblock = gr.hier_block2("test_block",
163 gr.io_signature(1,1,gr.sizeof_int),
164 gr.io_signature(1,1,gr.sizeof_int))
165 nop1 = gr.nop(gr.sizeof_int)
166 hblock.connect(hblock, nop1)
167 hblock.disconnect(hblock, nop1)
169 def test_016_disconnect_input_not_connected(self):
170 hblock = gr.hier_block2("test_block",
171 gr.io_signature(1,1,gr.sizeof_int),
172 gr.io_signature(1,1,gr.sizeof_int))
173 nop1 = gr.nop(gr.sizeof_int)
174 nop2 = gr.nop(gr.sizeof_int)
175 hblock.connect(hblock, nop1)
176 self.assertRaises(ValueError,
177 lambda: hblock.disconnect(hblock, nop2))
179 def test_017_disconnect_input_neg(self):
180 hblock = gr.hier_block2("test_block",
181 gr.io_signature(1,1,gr.sizeof_int),
182 gr.io_signature(1,1,gr.sizeof_int))
183 nop1 = gr.nop(gr.sizeof_int)
184 hblock.connect(hblock, nop1)
185 self.assertRaises(ValueError,
186 lambda: hblock.disconnect((hblock, -1), nop1))
188 def test_018_disconnect_input_exceeds(self):
189 hblock = gr.hier_block2("test_block",
190 gr.io_signature(1,1,gr.sizeof_int),
191 gr.io_signature(1,1,gr.sizeof_int))
192 nop1 = gr.nop(gr.sizeof_int)
193 hblock.connect(hblock, nop1)
194 self.assertRaises(ValueError,
195 lambda: hblock.disconnect((hblock, 1), nop1))
197 def test_019_disconnect_output(self):
198 hblock = gr.hier_block2("test_block",
199 gr.io_signature(1,1,gr.sizeof_int),
200 gr.io_signature(1,1,gr.sizeof_int))
201 nop1 = gr.nop(gr.sizeof_int)
202 hblock.connect(nop1, hblock)
203 hblock.disconnect(nop1, hblock)
205 def test_020_disconnect_output_not_connected(self):
206 hblock = gr.hier_block2("test_block",
207 gr.io_signature(1,1,gr.sizeof_int),
208 gr.io_signature(1,1,gr.sizeof_int))
209 nop1 = gr.nop(gr.sizeof_int)
210 nop2 = gr.nop(gr.sizeof_int)
211 hblock.connect(nop1, hblock)
212 self.assertRaises(ValueError,
213 lambda: hblock.disconnect(nop2, hblock))
215 def test_021_disconnect_output_neg(self):
216 hblock = gr.hier_block2("test_block",
217 gr.io_signature(1,1,gr.sizeof_int),
218 gr.io_signature(1,1,gr.sizeof_int))
219 nop1 = gr.nop(gr.sizeof_int)
220 hblock.connect(hblock, nop1)
221 self.assertRaises(ValueError,
222 lambda: hblock.disconnect(nop1, (hblock, -1)))
224 def test_022_disconnect_output_exceeds(self):
225 hblock = gr.hier_block2("test_block",
226 gr.io_signature(1,1,gr.sizeof_int),
227 gr.io_signature(1,1,gr.sizeof_int))
228 nop1 = gr.nop(gr.sizeof_int)
229 hblock.connect(nop1, hblock)
230 self.assertRaises(ValueError,
231 lambda: hblock.disconnect(nop1, (hblock, 1)))
233 def test_023_run(self):
234 hblock = gr.top_block("test_block")
235 data = (1.0, 2.0, 3.0, 4.0)
236 src = gr.vector_source_f(data, False)
237 dst = gr.vector_sink_f()
238 hblock.connect(src, dst)
239 r = gr.runtime(hblock)
241 self.assertEquals(data, dst.data())
243 if __name__ == "__main__":