Imported Upstream version 3.2.2
[debian/gnuradio] / gnuradio-core / src / python / gnuradio / gr / qa_kludge_copy.py
1 #!/usr/bin/env python
2 #
3 # Copyright 2006 Free Software Foundation, Inc.
4
5 # This file is part of GNU Radio
6
7 # GNU Radio is free software; you can redistribute it and/or modify
8 # it under the terms of the GNU General Public License as published by
9 # the Free Software Foundation; either version 3, or (at your option)
10 # any later version.
11
12 # GNU Radio is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15 # GNU General Public License for more details.
16
17 # You should have received a copy of the GNU General Public License
18 # along with GNU Radio; see the file COPYING.  If not, write to
19 # the Free Software Foundation, Inc., 51 Franklin Street,
20 # Boston, MA 02110-1301, USA.
21
22
23 from gnuradio import gr, gr_unittest
24 import math
25 import random
26
27
28 class test_kludge_copy(gr_unittest.TestCase):
29
30     def setUp(self):
31         self.tb = gr.top_block()
32         self.rng = random.Random()
33         self.rng.seed(0)
34
35     def tearDown(self):
36         del self.tb
37         del self.rng
38
39     def make_random_int_tuple(self, L):
40         result = []
41         for x in range(L):
42             result.append(self.rng.randint(int(-1e9), int(+1e9)))
43         return tuple(result)
44
45
46     def test_001(self):
47         # 1 input stream; 1 output stream
48         src0_data = self.make_random_int_tuple(16000)
49         src0 = gr.vector_source_i(src0_data)
50         op = gr.kludge_copy(gr.sizeof_int)
51         dst0 = gr.vector_sink_i()
52         self.tb.connect(src0, op, dst0)
53         self.tb.run()
54         dst0_data = dst0.data()
55         self.assertEqual(src0_data, dst0_data)
56         
57     def test_002(self):
58         # 2 input streams; 2 output streams
59         src0_data = self.make_random_int_tuple(16000)
60         src1_data = self.make_random_int_tuple(16000)
61         src0 = gr.vector_source_i(src0_data)
62         src1 = gr.vector_source_i(src1_data)
63         op = gr.kludge_copy(gr.sizeof_int)
64         dst0 = gr.vector_sink_i()
65         dst1 = gr.vector_sink_i()
66         self.tb.connect(src0, (op, 0), dst0)
67         self.tb.connect(src1, (op, 1), dst1)
68         self.tb.run()
69         dst0_data = dst0.data()
70         dst1_data = dst1.data()
71         self.assertEqual(src0_data, dst0_data)
72         self.assertEqual(src1_data, dst1_data)
73         
74     # Note: this is disabled due to triggering bug in ticket:181
75     # It only occurs with new top block code
76     def xtest_003(self):
77         # number of input streams != number of output streams
78         src0_data = self.make_random_int_tuple(16000)
79         src1_data = self.make_random_int_tuple(16000)
80         src0 = gr.vector_source_i(src0_data)
81         src1 = gr.vector_source_i(src1_data)
82         op = gr.kludge_copy(gr.sizeof_int)
83         dst0 = gr.vector_sink_i()
84         dst1 = gr.vector_sink_i()
85         self.tb.connect(src0, (op, 0), dst0)
86         self.tb.connect(src1, (op, 1))
87         self.assertRaises(ValueError, self.tb.run)
88
89 if __name__ == '__main__':
90     gr_unittest.main ()
91