Imported Upstream version 3.0
[debian/gnuradio] / gnuradio-core / src / python / gnuradio / gr / qa_basic_flow_graph.py
1 #!/usr/bin/env python
2
3 from gnuradio import gr, gr_unittest
4
5
6 # ----------------------------------------------------------------                         
7
8
9 class test_basic_flow_graph (gr_unittest.TestCase):
10
11     def setUp (self):
12         self.fg = gr.basic_flow_graph ()
13         
14     def tearDown (self):
15         self.fg = None
16
17     def test_000_create_delete (self):
18         pass
19
20     def test_001a_insert_1 (self):
21         fg = self.fg
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))
25
26     def test_001b_insert_1 (self):
27         fg = self.fg
28         src1 = gr.null_source (gr.sizeof_int)
29         dst1 = gr.null_sink (gr.sizeof_int)
30         fg.connect (src1, gr.endpoint (dst1, 0))
31
32     def test_001c_insert_1 (self):
33         fg = self.fg
34         src1 = gr.null_source (gr.sizeof_int)
35         dst1 = gr.null_sink (gr.sizeof_int)
36         fg.connect (gr.endpoint (src1, 0), dst1)
37
38     def test_001d_insert_1 (self):
39         fg = self.fg
40         src1 = gr.null_source (gr.sizeof_int)
41         dst1 = gr.null_sink (gr.sizeof_int)
42         fg.connect (src1, dst1)
43
44     def test_001e_insert_1 (self):
45         fg = self.fg
46         src1 = gr.null_source (gr.sizeof_int)
47         dst1 = gr.null_sink (gr.sizeof_int)
48         fg.connect ((src1, 0), (dst1, 0))
49
50     def test_002_dst_in_use (self):
51         fg = self.fg
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),
58                                                 (dst1, 0)))
59
60     def test_003_no_such_src_port (self):
61         fg = self.fg
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),
66                                                 (dst1, 0)))
67
68     def test_004_no_such_dst_port (self):
69         fg = self.fg
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),
74                                                 (dst1, 1)))
75
76     def test_005_one_src_two_dst (self):
77         fg = self.fg
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))
83
84     def test_006_check_item_sizes (self):
85         fg = self.fg
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),
90                                                 (dst1, 0)))
91         
92     def test_007_validate (self):
93         fg = self.fg
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))
99         fg.validate ()
100
101     def test_008_validate (self):
102         fg = self.fg
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))
111         fg.validate ()
112
113     def test_009_validate (self):
114         fg = self.fg
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 ())
125
126
127     def test_010_validate (self):
128         fg = self.fg
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 ())
139
140         
141     def test_011_disconnect (self):
142         fg = self.fg
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))
151         fg.validate ()
152         fg.disconnect ((src1, 0), (nop1, 1))
153         fg.validate ()
154         self.assertRaises (ValueError,
155                            lambda : fg.disconnect ((src1, 0),
156                                                    (nop1, 1)))
157
158     def test_012_connect (self):
159         fg = self.fg
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
168         fg.validate ()
169
170     def test_013_connect_varargs (self):
171         fg = self.fg
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))
179
180     def test_014_connect_varargs (self):
181         fg = self.fg
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)
186         fg.validate ()
187
188 if __name__ == '__main__':
189     gr_unittest.main ()
190