Merge r6160:6168 from jcorgan/fg into trunk.
[debian/gnuradio] / gnuradio-core / src / python / gnuradio / gr / qa_hier_block2.py
1 #!/usr/bin/env python
2
3 from gnuradio import gr, gr_unittest
4
5 class test_hier_block2(gr_unittest.TestCase):
6
7     def setUp(self):
8         pass
9
10     def tearDown(self):
11         pass
12
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))
22
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)
29
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))
39
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)
46
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))
56
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))
64
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))
72
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)))
81
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)))
90
91     def test_009_check_topology(self):
92         hblock = gr.top_block("test_block")
93         hblock.check_topology(0, 0)
94
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)
104         runtime.run()
105         actual1 = sink1.data()
106         actual2 = sink2.data()
107         self.assertEquals(expected, actual1)
108         self.assertEquals(expected, actual2)
109
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)
117    
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))
127    
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))
136
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))
145
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)
153    
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))
163    
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)))
172
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)))
181
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)
189         r.run()
190         self.assertEquals(data, dst.data())
191
192 if __name__ == "__main__":
193     gr_unittest.main()