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_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         hblock.run()
 104         actual1 = sink1.data()
 105         actual2 = sink2.data()
 106         self.assertEquals(expected, actual1)
 107         self.assertEquals(expected, actual2)
 108
 109     def test_012_disconnect_input(self):
 110         hblock = gr.hier_block2("test_block",
 111                                 gr.io_signature(1,1,gr.sizeof_int),
 112                                 gr.io_signature(1,1,gr.sizeof_int))
 113         nop1 = gr.nop(gr.sizeof_int)
 114         hblock.connect(hblock, nop1)
 115         hblock.disconnect(hblock, nop1)
 116
 117     def test_013_disconnect_input_not_connected(self):
 118         hblock = gr.hier_block2("test_block",
 119                                 gr.io_signature(1,1,gr.sizeof_int),
 120                                 gr.io_signature(1,1,gr.sizeof_int))
 121         nop1 = gr.nop(gr.sizeof_int)
 122         nop2 = gr.nop(gr.sizeof_int)
 123         hblock.connect(hblock, nop1)
 124         self.assertRaises(ValueError,
 125             lambda: hblock.disconnect(hblock, nop2))
 126
 127     def test_014_disconnect_input_neg(self):
 128         hblock = gr.hier_block2("test_block",
 129                                 gr.io_signature(1,1,gr.sizeof_int),
 130                                 gr.io_signature(1,1,gr.sizeof_int))
 131         nop1 = gr.nop(gr.sizeof_int)
 132         hblock.connect(hblock, nop1)
 133         self.assertRaises(ValueError,
 134             lambda: hblock.disconnect((hblock, -1), nop1))
 135
 136     def test_015_disconnect_input_exceeds(self):
 137         hblock = gr.hier_block2("test_block",
 138                                 gr.io_signature(1,1,gr.sizeof_int),
 139                                 gr.io_signature(1,1,gr.sizeof_int))
 140         nop1 = gr.nop(gr.sizeof_int)
 141         hblock.connect(hblock, nop1)
 142         self.assertRaises(ValueError,
 143             lambda: hblock.disconnect((hblock, 1), nop1))
 144
 145     def test_016_disconnect_output(self):
 146         hblock = gr.hier_block2("test_block",
 147                                 gr.io_signature(1,1,gr.sizeof_int),
 148                                 gr.io_signature(1,1,gr.sizeof_int))
 149         nop1 = gr.nop(gr.sizeof_int)
 150         hblock.connect(nop1, hblock)
 151         hblock.disconnect(nop1, hblock)
 152
 153     def test_017_disconnect_output_not_connected(self):
 154         hblock = gr.hier_block2("test_block",
 155                                 gr.io_signature(1,1,gr.sizeof_int),
 156                                 gr.io_signature(1,1,gr.sizeof_int))
 157         nop1 = gr.nop(gr.sizeof_int)
 158         nop2 = gr.nop(gr.sizeof_int)
 159         hblock.connect(nop1, hblock)
 160         self.assertRaises(ValueError,
 161             lambda: hblock.disconnect(nop2, hblock))
 162
 163     def test_018_disconnect_output_neg(self):
 164         hblock = gr.hier_block2("test_block",
 165                                 gr.io_signature(1,1,gr.sizeof_int),
 166                                 gr.io_signature(1,1,gr.sizeof_int))
 167         nop1 = gr.nop(gr.sizeof_int)
 168         hblock.connect(hblock, nop1)
 169         self.assertRaises(ValueError,
 170             lambda: hblock.disconnect(nop1, (hblock, -1)))
 171
 172     def test_019_disconnect_output_exceeds(self):
 173         hblock = gr.hier_block2("test_block",
 174                                 gr.io_signature(1,1,gr.sizeof_int),
 175                                 gr.io_signature(1,1,gr.sizeof_int))
 176         nop1 = gr.nop(gr.sizeof_int)
 177         hblock.connect(nop1, hblock)
 178         self.assertRaises(ValueError,
 179             lambda: hblock.disconnect(nop1, (hblock, 1)))
 180
 181     def test_020_run(self):
 182         hblock = gr.top_block("test_block")
 183         data = (1.0, 2.0, 3.0, 4.0)
 184         src = gr.vector_source_f(data, False)
 185         dst = gr.vector_sink_f()
 186         hblock.connect(src, dst)
 187         hblock.run()
 188         self.assertEquals(data, dst.data())
 189
 190     def test_021_connect_single(self):
 191         hblock = gr.top_block("test_block")
 192         blk = gr.hier_block2("block",
 193                              gr.io_signature(0, 0, 0),
 194                              gr.io_signature(0, 0, 0))
 195         hblock.connect(blk)
 196
 197     def test_022_connect_single_with_ports(self):
 198         hblock = gr.top_block("test_block")
 199         blk = gr.hier_block2("block",
 200                              gr.io_signature(1, 1, 1),
 201                              gr.io_signature(1, 1, 1))
 202         self.assertRaises(ValueError,
 203                           lambda: hblock.connect(blk))
 204
 205     def test_023_connect_single_twice(self):
 206         hblock = gr.top_block("test_block")
 207         blk = gr.hier_block2("block",
 208                              gr.io_signature(0, 0, 0),
 209                              gr.io_signature(0, 0, 0))
 210         hblock.connect(blk)
 211         self.assertRaises(ValueError,
 212                           lambda: hblock.connect(blk))
 213
 214     def test_024_disconnect_single(self):
 215         hblock = gr.top_block("test_block")
 216         blk = gr.hier_block2("block",
 217                              gr.io_signature(0, 0, 0),
 218                              gr.io_signature(0, 0, 0))
 219         hblock.connect(blk)
 220         hblock.disconnect(blk)
 221
 222     def test_025_disconnect_single_not_connected(self):
 223         hblock = gr.top_block("test_block")
 224         blk = gr.hier_block2("block",
 225                              gr.io_signature(0, 0, 0),
 226                              gr.io_signature(0, 0, 0))
 227         self.assertRaises(ValueError,
 228                           lambda: hblock.disconnect(blk))
 229
 230     def test_026_run_single(self):
 231         expected_data = (1.0,)
 232         tb = gr.top_block("top_block")
 233         hb = gr.hier_block2("block",
 234                             gr.io_signature(0, 0, 0),
 235                             gr.io_signature(0, 0, 0))
 236         src = gr.vector_source_f(expected_data)
 237         dst = gr.vector_sink_f()
 238         hb.connect(src, dst)
 239         tb.connect(hb)
 240         tb.run()
 241         self.assertEquals(expected_data, dst.data())
 242
 243 if __name__ == "__main__":
 244     gr_unittest.main()