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_004_connect_output(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         hblock.connect(nop1, hblock)
  36
  37     def test_005_connect_output_in_use(self):
  38         hblock = gr.hier_block2("test_block",
  39                                 gr.io_signature(1,1,gr.sizeof_int),
  40                                 gr.io_signature(1,1,gr.sizeof_int))
  41         nop1 = gr.nop(gr.sizeof_int)
  42         nop2 = gr.nop(gr.sizeof_int)
  43         hblock.connect(nop1, hblock)
  44         self.assertRaises(ValueError,
  45             lambda: hblock.connect(nop2, hblock))
  46
  47     def test_006_connect_invalid_src_port_neg(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         self.assertRaises(ValueError,
  53             lambda: hblock.connect((hblock, -1), nop1))
  54
  55     def test_005_connect_invalid_src_port_exceeds(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         self.assertRaises(ValueError,
  61             lambda: hblock.connect((hblock, 1), nop1))
  62
  63     def test_007_connect_invalid_dst_port_neg(self):
  64         hblock = gr.hier_block2("test_block",
  65                                 gr.io_signature(1,1,gr.sizeof_int),
  66                                 gr.io_signature(1,1,gr.sizeof_int))
  67         nop1 = gr.nop(gr.sizeof_int)
  68         nop2 = gr.nop(gr.sizeof_int)
  69         self.assertRaises(ValueError,
  70             lambda: hblock.connect(nop1, (nop2, -1)))
  71
  72     def test_008_connect_invalid_dst_port_exceeds(self):
  73         hblock = gr.hier_block2("test_block",
  74                                 gr.io_signature(1,1,gr.sizeof_int),
  75                                 gr.io_signature(1,1,gr.sizeof_int))
  76         nop1 = gr.null_sink(gr.sizeof_int)
  77         nop2 = gr.null_sink(gr.sizeof_int)
  78         self.assertRaises(ValueError,
  79             lambda: hblock.connect(nop1, (nop2, 1)))
  80
  81     def test_009_check_topology(self):
  82         hblock = gr.top_block("test_block")
  83         hblock.check_topology(0, 0)
  84
  85     def test_010_run(self):
  86         expected = (1.0, 2.0, 3.0, 4.0)
  87         hblock = gr.top_block("test_block")
  88         src = gr.vector_source_f(expected, False)
  89         sink1 = gr.vector_sink_f()
  90         sink2 = gr.vector_sink_f()
  91         hblock.connect(src, sink1)
  92         hblock.connect(src, sink2)
  93         hblock.run()
  94         actual1 = sink1.data()
  95         actual2 = sink2.data()
  96         self.assertEquals(expected, actual1)
  97         self.assertEquals(expected, actual2)
  98
  99     def test_012_disconnect_input(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         hblock.connect(hblock, nop1)
 105         hblock.disconnect(hblock, nop1)
 106
 107     def test_013_disconnect_input_not_connected(self):
 108         hblock = gr.hier_block2("test_block",
 109                                 gr.io_signature(1,1,gr.sizeof_int),
 110                                 gr.io_signature(1,1,gr.sizeof_int))
 111         nop1 = gr.nop(gr.sizeof_int)
 112         nop2 = gr.nop(gr.sizeof_int)
 113         hblock.connect(hblock, nop1)
 114         self.assertRaises(ValueError,
 115             lambda: hblock.disconnect(hblock, nop2))
 116
 117     def test_014_disconnect_input_neg(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         hblock.connect(hblock, nop1)
 123         self.assertRaises(ValueError,
 124             lambda: hblock.disconnect((hblock, -1), nop1))
 125
 126     def test_015_disconnect_input_exceeds(self):
 127         hblock = gr.hier_block2("test_block",
 128                                 gr.io_signature(1,1,gr.sizeof_int),
 129                                 gr.io_signature(1,1,gr.sizeof_int))
 130         nop1 = gr.nop(gr.sizeof_int)
 131         hblock.connect(hblock, nop1)
 132         self.assertRaises(ValueError,
 133             lambda: hblock.disconnect((hblock, 1), nop1))
 134
 135     def test_016_disconnect_output(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         hblock.connect(nop1, hblock)
 141         hblock.disconnect(nop1, hblock)
 142
 143     def test_017_disconnect_output_not_connected(self):
 144         hblock = gr.hier_block2("test_block",
 145                                 gr.io_signature(1,1,gr.sizeof_int),
 146                                 gr.io_signature(1,1,gr.sizeof_int))
 147         nop1 = gr.nop(gr.sizeof_int)
 148         nop2 = gr.nop(gr.sizeof_int)
 149         hblock.connect(nop1, hblock)
 150         self.assertRaises(ValueError,
 151             lambda: hblock.disconnect(nop2, hblock))
 152
 153     def test_018_disconnect_output_neg(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         hblock.connect(hblock, nop1)
 159         self.assertRaises(ValueError,
 160             lambda: hblock.disconnect(nop1, (hblock, -1)))
 161
 162     def test_019_disconnect_output_exceeds(self):
 163         hblock = gr.hier_block2("test_block",
 164                                 gr.io_signature(1,1,gr.sizeof_int),
 165                                 gr.io_signature(1,1,gr.sizeof_int))
 166         nop1 = gr.nop(gr.sizeof_int)
 167         hblock.connect(nop1, hblock)
 168         self.assertRaises(ValueError,
 169             lambda: hblock.disconnect(nop1, (hblock, 1)))
 170
 171     def test_020_run(self):
 172         hblock = gr.top_block("test_block")
 173         data = (1.0, 2.0, 3.0, 4.0)
 174         src = gr.vector_source_f(data, False)
 175         dst = gr.vector_sink_f()
 176         hblock.connect(src, dst)
 177         hblock.run()
 178         self.assertEquals(data, dst.data())
 179
 180     def test_021_connect_single(self):
 181         hblock = gr.top_block("test_block")
 182         blk = gr.hier_block2("block",
 183                              gr.io_signature(0, 0, 0),
 184                              gr.io_signature(0, 0, 0))
 185         hblock.connect(blk)
 186
 187     def test_022_connect_single_with_ports(self):
 188         hblock = gr.top_block("test_block")
 189         blk = gr.hier_block2("block",
 190                              gr.io_signature(1, 1, 1),
 191                              gr.io_signature(1, 1, 1))
 192         self.assertRaises(ValueError,
 193                           lambda: hblock.connect(blk))
 194
 195     def test_023_connect_single_twice(self):
 196         hblock = gr.top_block("test_block")
 197         blk = gr.hier_block2("block",
 198                              gr.io_signature(0, 0, 0),
 199                              gr.io_signature(0, 0, 0))
 200         hblock.connect(blk)
 201         self.assertRaises(ValueError,
 202                           lambda: hblock.connect(blk))
 203
 204     def test_024_disconnect_single(self):
 205         hblock = gr.top_block("test_block")
 206         blk = gr.hier_block2("block",
 207                              gr.io_signature(0, 0, 0),
 208                              gr.io_signature(0, 0, 0))
 209         hblock.connect(blk)
 210         hblock.disconnect(blk)
 211
 212     def test_025_disconnect_single_not_connected(self):
 213         hblock = gr.top_block("test_block")
 214         blk = gr.hier_block2("block",
 215                              gr.io_signature(0, 0, 0),
 216                              gr.io_signature(0, 0, 0))
 217         self.assertRaises(ValueError,
 218                           lambda: hblock.disconnect(blk))
 219
 220     def test_026_run_single(self):
 221         expected_data = (1.0,)
 222         tb = gr.top_block("top_block")
 223         hb = gr.hier_block2("block",
 224                             gr.io_signature(0, 0, 0),
 225                             gr.io_signature(0, 0, 0))
 226         src = gr.vector_source_f(expected_data)
 227         dst = gr.vector_sink_f()
 228         hb.connect(src, dst)
 229         tb.connect(hb)
 230         tb.run()
 231         self.assertEquals(expected_data, dst.data())
 232
 233     def test_027_disconnected_internal(self):
 234         tb = gr.top_block()
 235         hb = gr.hier_block2("block",
 236                             gr.io_signature(1, 1, 1),
 237                             gr.io_signature(1, 1, 1))
 238         src = gr.vector_source_b([1, ])
 239         dst = gr.vector_sink_b()
 240         tb.connect(src, hb, dst) # hb is not connected internally
 241         self.assertRaises(RuntimeError,
 242                           lambda: tb.run())
 243
 244     def test_028_singleton_reconfigure(self):
 245         tb = gr.top_block()
 246         hb = gr.hier_block2("block",
 247                             gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0))
 248         src = gr.vector_source_b([1, ])
 249         dst = gr.vector_sink_b()
 250         hb.connect(src, dst)
 251         tb.connect(hb) # Singleton connect
 252         tb.lock()
 253         tb.disconnect_all()
 254         tb.connect(src, dst)
 255         tb.unlock()
 256
 257     def test_029_singleton_disconnect(self):
 258         tb = gr.top_block()
 259         src = gr.vector_source_b([1, ])
 260         dst = gr.vector_sink_b()
 261         tb.connect(src, dst)
 262         tb.disconnect(src)   # Singleton disconnect
 263         tb.connect(src, dst)
 264         tb.run()
 265         self.assertEquals(dst.data(), (1,))
 266
 267     def test_030_nested_input(self):
 268         tb = gr.top_block()
 269         src = gr.vector_source_b([1,])
 270         hb1 = gr.hier_block2("hb1",
 271                              gr.io_signature(1, 1, gr.sizeof_char),
 272                              gr.io_signature(0, 0, 0))
 273         hb2 = gr.hier_block2("hb2",
 274                              gr.io_signature(1, 1, gr.sizeof_char),
 275                              gr.io_signature(0, 0, 0))
 276         dst = gr.vector_sink_b()
 277         tb.connect(src, hb1)
 278         hb1.connect(hb1, hb2)
 279         hb2.connect(hb2, gr.kludge_copy(gr.sizeof_char), dst)
 280         tb.run()
 281         self.assertEquals(dst.data(), (1,))
 282
 283     def test_031_multiple_internal_inputs(self):
 284         tb = gr.top_block()
 285         src = gr.vector_source_f([1.0,])
 286         hb = gr.hier_block2("hb",
 287                             gr.io_signature(1, 1, gr.sizeof_float),
 288                             gr.io_signature(1, 1, gr.sizeof_float))
 289         m1 = gr.multiply_const_ff(1.0)
 290         m2 = gr.multiply_const_ff(2.0)
 291         add = gr.add_ff()
 292         hb.connect(hb, m1)       # m1 is connected to hb external input #0
 293         hb.connect(hb, m2)       # m2 is also connected to hb external input #0
 294         hb.connect(m1, (add, 0))
 295         hb.connect(m2, (add, 1))
 296         hb.connect(add, hb)      # add is connected to hb external output #0
 297         dst = gr.vector_sink_f()
 298         tb.connect(src, hb, dst)
 299         tb.run()
 300         self.assertEquals(dst.data(), (3.0,))
 301
 302     def test_032_nested_multiple_internal_inputs(self):
 303         tb = gr.top_block()
 304         src = gr.vector_source_f([1.0,])
 305         hb = gr.hier_block2("hb",
 306                             gr.io_signature(1, 1, gr.sizeof_float),
 307                             gr.io_signature(1, 1, gr.sizeof_float))
 308         hb2 = gr.hier_block2("hb",
 309                             gr.io_signature(1, 1, gr.sizeof_float),
 310                             gr.io_signature(1, 1, gr.sizeof_float))
 311
 312         m1 = gr.multiply_const_ff(1.0)
 313         m2 = gr.multiply_const_ff(2.0)
 314         add = gr.add_ff()
 315         hb2.connect(hb2, m1)       # m1 is connected to hb2 external input #0
 316         hb2.connect(hb2, m2)       # m2 is also connected to hb2 external input #0
 317         hb2.connect(m1, (add, 0))
 318         hb2.connect(m2, (add, 1))
 319         hb2.connect(add, hb2)      # add is connected to hb2 external output #0
 320         hb.connect(hb, hb2, hb)   # hb as hb2 as nested internal block
 321         dst = gr.vector_sink_f()
 322         tb.connect(src, hb, dst)
 323         tb.run()
 324         self.assertEquals(dst.data(), (3.0,))
 325
 326
 327 if __name__ == "__main__":
 328     gr_unittest.main()