git.gag.com Git - debian/gnuradio/blob - 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_001_connect_internal(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         nop2 = gr.nop(gr.sizeof_int)
  29         hblock.connect(nop1, nop2)
  30
  31     def test_002_connect_input(self):
  32         hblock = gr.hier_block2("test_block",
  33                                 gr.io_signature(1,1,gr.sizeof_int),
  34                                 gr.io_signature(1,1,gr.sizeof_int))
  35         nop1 = gr.nop(gr.sizeof_int)
  36         hblock.connect(hblock, nop1)
  37
  38     def test_002a_connect_input_in_use(self):
  39         hblock = gr.hier_block2("test_block",
  40                                 gr.io_signature(1,1,gr.sizeof_int),
  41                                 gr.io_signature(1,1,gr.sizeof_int))
  42         nop1 = gr.nop(gr.sizeof_int)
  43         nop2 = gr.nop(gr.sizeof_int)
  44         hblock.connect(hblock, nop1)
  45         self.assertRaises(ValueError,
  46             lambda: hblock.connect(hblock, nop2))
  47
  48     def test_003_connect_output(self):
  49         hblock = gr.hier_block2("test_block",
  50                                 gr.io_signature(1,1,gr.sizeof_int),
  51                                 gr.io_signature(1,1,gr.sizeof_int))
  52         nop1 = gr.nop(gr.sizeof_int)
  53         hblock.connect(nop1, hblock)
  54
  55     def test_003a_connect_output_in_use(self):
  56         hblock = gr.hier_block2("test_block",
  57                                 gr.io_signature(1,1,gr.sizeof_int),
  58                                 gr.io_signature(1,1,gr.sizeof_int))
  59         nop1 = gr.nop(gr.sizeof_int)
  60         nop2 = gr.nop(gr.sizeof_int)
  61         hblock.connect(nop1, hblock)
  62         self.assertRaises(ValueError,
  63             lambda: hblock.connect(nop2, hblock))
  64
  65     def test_004_connect_invalid_src_port_neg(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_005_connect_invalid_src_port_exceeds(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         self.assertRaises(ValueError,
  79             lambda: hblock.connect((hblock, 1), nop1))
  80
  81     def test_006_connect_invalid_dst_port_neg(self):
  82         hblock = gr.hier_block2("test_block",
  83                                 gr.io_signature(1,1,gr.sizeof_int),
  84                                 gr.io_signature(1,1,gr.sizeof_int))
  85         nop1 = gr.nop(gr.sizeof_int)
  86         nop2 = gr.nop(gr.sizeof_int)
  87         self.assertRaises(ValueError,
  88             lambda: hblock.connect(nop1, (nop2, -1)))
  89
  90     def test_007_connect_invalid_dst_port_exceeds(self):
  91         hblock = gr.hier_block2("test_block",
  92                                 gr.io_signature(1,1,gr.sizeof_int),
  93                                 gr.io_signature(1,1,gr.sizeof_int))
  94         nop1 = gr.null_sink(gr.sizeof_int)
  95         nop2 = gr.null_sink(gr.sizeof_int)
  96         self.assertRaises(ValueError,
  97             lambda: hblock.connect(nop1, (nop2, 1)))
  98
  99     def test_008_connect_dst_port_in_use(self):
 100         hblock = gr.hier_block2("test_block",
 101                                 gr.io_signature(1,1,gr.sizeof_int),
 102                                 gr.io_signature(1,1,gr.sizeof_int))
 103         nop1 = gr.nop(gr.sizeof_int)
 104         nop2 = gr.nop(gr.sizeof_int)
 105         hblock.connect(nop1, nop2);
 106         self.assertRaises(ValueError,
 107             lambda: hblock.connect(nop1, nop2))
 108
 109     def test_009_connect_one_src_two_dst(self):
 110         hblock = gr.top_block("test_block")
 111         src = gr.null_source(gr.sizeof_int)
 112         dst1 = gr.null_sink(gr.sizeof_int)
 113         dst2 = gr.null_sink(gr.sizeof_int)
 114         hblock.connect(src, dst1)
 115         hblock.connect(src, dst2)
 116
 117     def test_010_connect_type_mismatch(self):
 118         hblock = gr.top_block("test_block")
 119         nop1 = gr.nop(gr.sizeof_char)
 120         nop2 = gr.nop(gr.sizeof_int)
 121         self.assertRaises(ValueError,
 122             lambda: hblock.connect(nop1, nop2))
 123
 124     def test_011_check_topology(self):
 125         hblock = gr.top_block("test_block")
 126         hblock.check_topology(0, 0)
 127
 128     def test_012_disconnect(self):
 129         hblock = gr.top_block("test_block")
 130         nop1 = gr.nop(gr.sizeof_int)
 131         nop2 = gr.nop(gr.sizeof_int)
 132         hblock.connect(nop1, nop2)
 133         hblock.disconnect(nop1, nop2)
 134
 135     def test_013_disconnect_not_connected(self):
 136         hblock = gr.hier_block2("test_block",
 137                                 gr.io_signature(1,1,gr.sizeof_int),
 138                                 gr.io_signature(1,1,gr.sizeof_int))
 139         nop1 = gr.nop(gr.sizeof_int)
 140         nop2 = gr.nop(gr.sizeof_int)
 141         nop3 = gr.nop(gr.sizeof_int)
 142         hblock.connect(nop1, nop2)
 143         self.assertRaises(ValueError,
 144             lambda: hblock.disconnect(nop1, nop3))
 145
 146     def test_014_run(self):
 147         expected = (1.0, 2.0, 3.0, 4.0)
 148         hblock = gr.top_block("test_block")
 149         src = gr.vector_source_f(expected, False)
 150         sink1 = gr.vector_sink_f()
 151         sink2 = gr.vector_sink_f()
 152         hblock.connect(src, sink1)
 153         hblock.connect(src, sink2)
 154         runtime = gr.runtime(hblock)
 155         runtime.run()
 156         actual1 = sink1.data()
 157         actual2 = sink2.data()
 158         self.assertEquals(expected, actual1)
 159         self.assertEquals(expected, actual2)
 160
 161     def test_015_connect_disconnect(self):
 162         expected = (1.0, 2.0, 3.0, 4.0)
 163         hblock = gr.top_block("test_block")
 164         src = gr.vector_source_f(expected, False)
 165         sink1 = gr.vector_sink_f()
 166         sink2 = gr.vector_sink_f()
 167         hblock.connect(src, sink1)
 168         hblock.connect(src, sink2)
 169         hblock.disconnect(src, sink2)
 170         runtime = gr.runtime(hblock)
 171         runtime.run()
 172
 173 if __name__ == "__main__":
 174     gr_unittest.main()