3 from gnuradio import gr, gr_unittest
6 # ----------------------------------------------------------------
9 class test_basic_flow_graph (gr_unittest.TestCase):
12 self.fg = gr.basic_flow_graph ()
17 def test_000_create_delete (self):
20 def test_001a_insert_1 (self):
22 src1 = gr.null_source (gr.sizeof_int)
23 dst1 = gr.null_sink (gr.sizeof_int)
24 fg.connect (gr.endpoint (src1, 0), gr.endpoint (dst1, 0))
26 def test_001b_insert_1 (self):
28 src1 = gr.null_source (gr.sizeof_int)
29 dst1 = gr.null_sink (gr.sizeof_int)
30 fg.connect (src1, gr.endpoint (dst1, 0))
32 def test_001c_insert_1 (self):
34 src1 = gr.null_source (gr.sizeof_int)
35 dst1 = gr.null_sink (gr.sizeof_int)
36 fg.connect (gr.endpoint (src1, 0), dst1)
38 def test_001d_insert_1 (self):
40 src1 = gr.null_source (gr.sizeof_int)
41 dst1 = gr.null_sink (gr.sizeof_int)
42 fg.connect (src1, dst1)
44 def test_001e_insert_1 (self):
46 src1 = gr.null_source (gr.sizeof_int)
47 dst1 = gr.null_sink (gr.sizeof_int)
48 fg.connect ((src1, 0), (dst1, 0))
50 def test_002_dst_in_use (self):
52 src1 = gr.null_source (gr.sizeof_int)
53 src2 = gr.null_source (gr.sizeof_int)
54 dst1 = gr.null_sink (gr.sizeof_int)
55 fg.connect ((src1, 0), (dst1, 0))
56 self.assertRaises (ValueError,
57 lambda : fg.connect ((src2, 0),
60 def test_003_no_such_src_port (self):
62 src1 = gr.null_source (gr.sizeof_int)
63 dst1 = gr.null_sink (gr.sizeof_int)
64 self.assertRaises (ValueError,
65 lambda : fg.connect ((src1, 1),
68 def test_004_no_such_dst_port (self):
70 src1 = gr.null_source (gr.sizeof_int)
71 dst1 = gr.null_sink (gr.sizeof_int)
72 self.assertRaises (ValueError,
73 lambda : fg.connect ((src1, 0),
76 def test_005_one_src_two_dst (self):
78 src1 = gr.null_source (gr.sizeof_int)
79 dst1 = gr.null_sink (gr.sizeof_int)
80 dst2 = gr.null_sink (gr.sizeof_int)
81 fg.connect ((src1, 0), (dst1, 0))
82 fg.connect ((src1, 0), (dst2, 0))
84 def test_006_check_item_sizes (self):
86 src1 = gr.null_source (gr.sizeof_int)
87 dst1 = gr.null_sink (gr.sizeof_char)
88 self.assertRaises (ValueError,
89 lambda : fg.connect ((src1, 0),
92 def test_007_validate (self):
94 src1 = gr.null_source (gr.sizeof_int)
95 dst1 = gr.null_sink (gr.sizeof_int)
96 dst2 = gr.null_sink (gr.sizeof_int)
97 fg.connect ((src1, 0), (dst1, 0))
98 fg.connect ((src1, 0), (dst2, 0))
101 def test_008_validate (self):
103 src1 = gr.null_source (gr.sizeof_int)
104 nop1 = gr.nop (gr.sizeof_int)
105 dst1 = gr.null_sink (gr.sizeof_int)
106 dst2 = gr.null_sink (gr.sizeof_int)
107 fg.connect ((src1, 0), (nop1, 0))
108 fg.connect ((src1, 0), (nop1, 1))
109 fg.connect ((nop1, 0), (dst1, 0))
110 fg.connect ((nop1, 1), (dst2, 0))
113 def test_009_validate (self):
115 src1 = gr.null_source (gr.sizeof_int)
116 nop1 = gr.nop (gr.sizeof_int)
117 dst1 = gr.null_sink (gr.sizeof_int)
118 dst2 = gr.null_sink (gr.sizeof_int)
119 fg.connect ((src1, 0), (nop1, 0))
120 fg.connect ((src1, 0), (nop1, 2))
121 fg.connect ((nop1, 0), (dst1, 0))
122 fg.connect ((nop1, 1), (dst2, 0))
123 self.assertRaises (ValueError,
124 lambda : fg.validate ())
127 def test_010_validate (self):
129 src1 = gr.null_source (gr.sizeof_int)
130 nop1 = gr.nop (gr.sizeof_int)
131 dst1 = gr.null_sink (gr.sizeof_int)
132 dst2 = gr.null_sink (gr.sizeof_int)
133 fg.connect ((src1, 0), (nop1, 0))
134 fg.connect ((src1, 0), (nop1, 1))
135 fg.connect ((nop1, 0), (dst1, 0))
136 fg.connect ((nop1, 2), (dst2, 0))
137 self.assertRaises (ValueError,
138 lambda : fg.validate ())
141 def test_011_disconnect (self):
143 src1 = gr.null_source (gr.sizeof_int)
144 nop1 = gr.nop (gr.sizeof_int)
145 dst1 = gr.null_sink (gr.sizeof_int)
146 dst2 = gr.null_sink (gr.sizeof_int)
147 fg.connect ((src1, 0), (nop1, 0))
148 fg.connect ((src1, 0), (nop1, 1))
149 fg.connect ((nop1, 0), (dst1, 0))
150 fg.connect ((nop1, 1), (dst2, 0))
152 fg.disconnect ((src1, 0), (nop1, 1))
154 self.assertRaises (ValueError,
155 lambda : fg.disconnect ((src1, 0),
158 def test_012_connect (self):
160 src_data = (0, 1, 2, 3)
161 expected_result = (2, 3, 4, 5)
162 src1 = gr.vector_source_i (src_data)
163 op = gr.add_const_ii (2)
164 dst1 = gr.vector_sink_i ()
165 fg.connect (src1, op)
166 fg.connect (op, dst1)
167 # print "edge_list:", fg.edge_list
170 def test_013_connect_varargs (self):
172 src1 = gr.null_source (gr.sizeof_int)
173 nop1 = gr.nop (gr.sizeof_int)
174 dst1 = gr.null_sink (gr.sizeof_int)
175 self.assertRaises (ValueError,
176 lambda : fg.connect ())
177 self.assertRaises (ValueError,
178 lambda : fg.connect (src1))
180 def test_014_connect_varargs (self):
182 src1 = gr.null_source (gr.sizeof_int)
183 nop1 = gr.nop (gr.sizeof_int)
184 dst1 = gr.null_sink (gr.sizeof_int)
185 fg.connect (src1, nop1, dst1)
188 if __name__ == '__main__':