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)
103 runtime = gr.runtime(hblock)
105 actual1 = sink1.data()
106 actual2 = sink2.data()
107 self.assertEquals(expected, actual1)
108 self.assertEquals(expected, actual2)
110 def test_012_disconnect_input(self):
111 hblock = gr.hier_block2("test_block",
112 gr.io_signature(1,1,gr.sizeof_int),
113 gr.io_signature(1,1,gr.sizeof_int))
114 nop1 = gr.nop(gr.sizeof_int)
115 hblock.connect(hblock, nop1)
116 hblock.disconnect(hblock, nop1)
118 def test_013_disconnect_input_not_connected(self):
119 hblock = gr.hier_block2("test_block",
120 gr.io_signature(1,1,gr.sizeof_int),
121 gr.io_signature(1,1,gr.sizeof_int))
122 nop1 = gr.nop(gr.sizeof_int)
123 nop2 = gr.nop(gr.sizeof_int)
124 hblock.connect(hblock, nop1)
125 self.assertRaises(ValueError,
126 lambda: hblock.disconnect(hblock, nop2))
128 def test_014_disconnect_input_neg(self):
129 hblock = gr.hier_block2("test_block",
130 gr.io_signature(1,1,gr.sizeof_int),
131 gr.io_signature(1,1,gr.sizeof_int))
132 nop1 = gr.nop(gr.sizeof_int)
133 hblock.connect(hblock, nop1)
134 self.assertRaises(ValueError,
135 lambda: hblock.disconnect((hblock, -1), nop1))
137 def test_015_disconnect_input_exceeds(self):
138 hblock = gr.hier_block2("test_block",
139 gr.io_signature(1,1,gr.sizeof_int),
140 gr.io_signature(1,1,gr.sizeof_int))
141 nop1 = gr.nop(gr.sizeof_int)
142 hblock.connect(hblock, nop1)
143 self.assertRaises(ValueError,
144 lambda: hblock.disconnect((hblock, 1), nop1))
146 def test_016_disconnect_output(self):
147 hblock = gr.hier_block2("test_block",
148 gr.io_signature(1,1,gr.sizeof_int),
149 gr.io_signature(1,1,gr.sizeof_int))
150 nop1 = gr.nop(gr.sizeof_int)
151 hblock.connect(nop1, hblock)
152 hblock.disconnect(nop1, hblock)
154 def test_017_disconnect_output_not_connected(self):
155 hblock = gr.hier_block2("test_block",
156 gr.io_signature(1,1,gr.sizeof_int),
157 gr.io_signature(1,1,gr.sizeof_int))
158 nop1 = gr.nop(gr.sizeof_int)
159 nop2 = gr.nop(gr.sizeof_int)
160 hblock.connect(nop1, hblock)
161 self.assertRaises(ValueError,
162 lambda: hblock.disconnect(nop2, hblock))
164 def test_018_disconnect_output_neg(self):
165 hblock = gr.hier_block2("test_block",
166 gr.io_signature(1,1,gr.sizeof_int),
167 gr.io_signature(1,1,gr.sizeof_int))
168 nop1 = gr.nop(gr.sizeof_int)
169 hblock.connect(hblock, nop1)
170 self.assertRaises(ValueError,
171 lambda: hblock.disconnect(nop1, (hblock, -1)))
173 def test_019_disconnect_output_exceeds(self):
174 hblock = gr.hier_block2("test_block",
175 gr.io_signature(1,1,gr.sizeof_int),
176 gr.io_signature(1,1,gr.sizeof_int))
177 nop1 = gr.nop(gr.sizeof_int)
178 hblock.connect(nop1, hblock)
179 self.assertRaises(ValueError,
180 lambda: hblock.disconnect(nop1, (hblock, 1)))
182 def test_020_run(self):
183 hblock = gr.top_block("test_block")
184 data = (1.0, 2.0, 3.0, 4.0)
185 src = gr.vector_source_f(data, False)
186 dst = gr.vector_sink_f()
187 hblock.connect(src, dst)
188 r = gr.runtime(hblock)
190 self.assertEquals(data, dst.data())
192 if __name__ == "__main__":