projects / debian / gnuradio / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
Merged r6285:6297 from features/deb into trunk. Nearing completion on Debian packagi...
[debian/gnuradio] / gnuradio-core / src / python / gnuradio / gr / qa_hier_block2.py
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py b/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py
index 9253b892af8fca7b94bb8589a49d263cd2455b13..d07d4cb3894ccb08be362bde038e4b20407fbb23 100755 (executable)
--- a/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py
@@ -6,10 +6,10 @@ class test_hier_block2(gr_unittest.TestCase):
 
     def setUp(self):
        pass
-       
+
     def tearDown(self):
        pass
-       
+
     def test_001_make(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
@@ -20,219 +20,172 @@ class test_hier_block2(gr_unittest.TestCase):
        self.assertEqual(1, hblock.output_signature().max_streams())
        self.assertEqual(gr.sizeof_int, hblock.output_signature().sizeof_stream_item(0))
 
-    def test_002_define_component(self):
+    def test_002_connect_input(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("src", gr.null_source(gr.sizeof_int))
-       hblock.define_component("dst", gr.null_sink(gr.sizeof_int))
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
 
-    def test_003_define_component_reserved_input(self):
+    def test_003_connect_input_in_use(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       self.assertRaises(ValueError, 
-           lambda: hblock.define_component("self", gr.nop(gr.sizeof_int)))
+       nop1 = gr.nop(gr.sizeof_int)
+       nop2 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
+       self.assertRaises(ValueError,
+           lambda: hblock.connect(hblock, nop2))
 
-    def test_004_define_component_name_in_use(self):
+    def test_004_connect_output(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("src", gr.null_source(gr.sizeof_int))
-       self.assertRaises(ValueError, 
-           lambda: hblock.define_component("src", gr.null_sink(gr.sizeof_int)))
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(nop1, hblock)
 
-    def test_006_connect_internal(self):
+    def test_005_connect_output_in_use(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.define_component("nop2", gr.nop(gr.sizeof_int))
-       hblock.connect("nop1", 0, "nop2", 0)
+       nop1 = gr.nop(gr.sizeof_int)
+       nop2 = gr.nop(gr.sizeof_int)
+       hblock.connect(nop1, hblock)
+       self.assertRaises(ValueError,
+           lambda: hblock.connect(nop2, hblock))
 
-    def test_007_connect_input(self):
+    def test_006_connect_invalid_src_port_neg(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.connect("self", 0, "nop1", 0)
+       nop1 = gr.nop(gr.sizeof_int)
+       self.assertRaises(ValueError, 
+           lambda: hblock.connect((hblock, -1), nop1))
 
-    def test_008_connect_output(self):
+    def test_005_connect_invalid_src_port_exceeds(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.connect("nop1", 0, "self", 0)
+       nop1 = gr.nop(gr.sizeof_int)
+       self.assertRaises(ValueError, 
+           lambda: hblock.connect((hblock, 1), nop1))
 
-    def test_009_connect_unknown_src(self):
+    def test_007_connect_invalid_dst_port_neg(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
        nop1 = gr.nop(gr.sizeof_int)
-       hblock.define_component("nop1", nop1)
+       nop2 = gr.nop(gr.sizeof_int)
        self.assertRaises(ValueError, 
-           lambda: hblock.connect("nop2", 0, "self", 0))
-    
-    def test_010_connect_unknown_dst(self):
+           lambda: hblock.connect(nop1, (nop2, -1)))
+
+    def test_008_connect_invalid_dst_port_exceeds(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
+       nop1 = gr.null_sink(gr.sizeof_int)
+       nop2 = gr.null_sink(gr.sizeof_int)
        self.assertRaises(ValueError, 
-           lambda: hblock.connect("self", 0, "nop2", 0))
+           lambda: hblock.connect(nop1, (nop2, 1)))
 
-    def test_011_connect_invalid_src_port_neg(self):
+    def test_009_check_topology(self):
+       hblock = gr.top_block("test_block")
+       hblock.check_topology(0, 0)
+
+    def test_010_run(self):
+        expected = (1.0, 2.0, 3.0, 4.0)
+        hblock = gr.top_block("test_block")
+        src = gr.vector_source_f(expected, False)
+        sink1 = gr.vector_sink_f()
+        sink2 = gr.vector_sink_f()
+        hblock.connect(src, sink1)
+        hblock.connect(src, sink2)
+        hblock.run()
+        actual1 = sink1.data()
+        actual2 = sink2.data()
+        self.assertEquals(expected, actual1)
+        self.assertEquals(expected, actual2)
+
+    def test_012_disconnect_input(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       self.assertRaises(ValueError, 
-           lambda: hblock.connect("self", -1, "nop1", 0))
-
-    def test_012_connect_invalid_src_port_exceeds(self):
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
+        hblock.disconnect(hblock, nop1)
+   
+    def test_013_disconnect_input_not_connected(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       self.assertRaises(ValueError, 
-           lambda: hblock.connect("self", 1, "nop1", 0))
-    
-    def test_013_connect_invalid_dst_port_neg(self):
+       nop1 = gr.nop(gr.sizeof_int)
+        nop2 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(hblock, nop2))
+   
+    def test_014_disconnect_input_neg(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       self.assertRaises(ValueError, 
-           lambda: hblock.connect("self", -1, "nop1", 0))
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect((hblock, -1), nop1))
 
-    def test_014_connect_invalid_dst_port_exceeds(self):
+    def test_015_disconnect_input_exceeds(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       self.assertRaises(ValueError, 
-           lambda: hblock.connect("self", 1, "nop1", 0))
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect((hblock, 1), nop1))
 
-    def test_015_connect_dst_port_in_use(self):
+    def test_016_disconnect_output(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
        nop1 = gr.nop(gr.sizeof_int)
-       hblock.define_component("nop1", nop1)
-       hblock.connect("nop1", 0, "self", 0);
-       self.assertRaises(ValueError, 
-           lambda: hblock.connect("nop1", 0, "self", 0))
-    
-    def test_016_connect_one_src_two_dst(self):
+       hblock.connect(nop1, hblock)
+        hblock.disconnect(nop1, hblock)
+   
+    def test_017_disconnect_output_not_connected(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("src", gr.null_source(gr.sizeof_int))
-       hblock.define_component("dst1", gr.null_sink(gr.sizeof_int))
-       hblock.define_component("dst2", gr.null_sink(gr.sizeof_int))
-       hblock.connect("src", 0, "dst1", 0)
-       hblock.connect("src", 0, "dst2", 0)
+       nop1 = gr.nop(gr.sizeof_int)
+        nop2 = gr.nop(gr.sizeof_int)
+       hblock.connect(nop1, hblock)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(nop2, hblock))
+   
+    def test_018_disconnect_output_neg(self):
+       hblock = gr.hier_block2("test_block", 
+                               gr.io_signature(1,1,gr.sizeof_int), 
+                               gr.io_signature(1,1,gr.sizeof_int))
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(nop1, (hblock, -1)))
 
-    def test_017_connect_type_mismatch(self):
+    def test_019_disconnect_output_exceeds(self):
        hblock = gr.hier_block2("test_block", 
                                gr.io_signature(1,1,gr.sizeof_int), 
                                gr.io_signature(1,1,gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_char))
-       self.assertRaises(ValueError, 
-           lambda: hblock.connect("nop1", 0, "self", 0))
-
-    def test_018_check_topology(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       hblock.check_topology(0, 0);
-    """
-    def test_019_validate(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       runtime = gr.runtime(hblock)
-       runtime.validate()
-    
-    def test_020_validate_1(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       hblock.define_component("src", gr.null_source(gr.sizeof_int))
-       hblock.define_component("dst1", gr.null_sink(gr.sizeof_int))
-       hblock.define_component("dst2", gr.null_sink(gr.sizeof_int))
-       hblock.connect("src", 0, "dst1", 0)
-       hblock.connect("src", 0, "dst2", 0)                     
-       runtime = gr.runtime(hblock)
-       runtime.validate()
-
-    def test_021_validate_2(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       hblock.define_component("src1", gr.null_source(gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.define_component("dst1", gr.null_sink(gr.sizeof_int))
-       hblock.define_component("dst2", gr.null_sink(gr.sizeof_int))
-       hblock.connect("src1", 0, "nop1", 0)
-       hblock.connect("src1", 0, "nop1", 1)
-       hblock.connect("nop1", 0, "dst1", 0)
-       hblock.connect("nop1", 1, "dst2", 0)
-       runtime = gr.runtime(hblock)
-       runtime.validate()
-        
-    def test_022_validate_3(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       hblock.define_component("src1", gr.null_source(gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.define_component("dst1", gr.null_sink(gr.sizeof_int))
-       hblock.define_component("dst2", gr.null_sink(gr.sizeof_int))
-       hblock.connect("src1", 0, "nop1", 0)
-       hblock.connect("src1", 0, "nop1", 2)
-       hblock.connect("nop1", 0, "dst1", 0)
-       hblock.connect("nop1", 1, "dst2", 0)
-       runtime = gr.runtime(hblock)
-       self.assertRaises(RuntimeError,
-           lambda: runtime.validate())
-        
-    def test_023_validate_4(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       hblock.define_component("src1", gr.null_source(gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.define_component("dst1", gr.null_sink(gr.sizeof_int))
-       hblock.define_component("dst2", gr.null_sink(gr.sizeof_int))
-       hblock.connect("src1", 0, "nop1", 0)
-       hblock.connect("src1", 0, "nop1", 1)
-       hblock.connect("nop1", 0, "dst1", 0)
-       hblock.connect("nop1", 2, "dst2", 0)
-       runtime = gr.runtime(hblock)
-       self.assertRaises(RuntimeError,
-           lambda: runtime.validate())
-        
-    def test_024_validate_5(self):
-       hblock = gr.hier_block2("test_block", 
-                               gr.io_signature(0,0,gr.sizeof_int), 
-                               gr.io_signature(0,0,gr.sizeof_int))
-       hblock.define_component("src1", gr.null_source(gr.sizeof_int))
-       hblock.define_component("nop1", gr.nop(gr.sizeof_int))
-       hblock.define_component("dst1", gr.null_sink(gr.sizeof_int))
-       hblock.define_component("dst2", gr.null_sink(gr.sizeof_int))
-       hblock.connect("src1", 0, "nop1", 0)
-       hblock.connect("src1", 0, "nop1", 1)
-       hblock.connect("nop1", 0, "dst1", 0)
-       hblock.connect("nop1", 1, "dst2", 0)
-       runtime = gr.runtime(hblock)
-       runtime.validate()
-       # Pending implementation of disconnect
-       # hblock.disconnect("src1", 0, "nop1", 1)
-       # runtime.validate()    
-       # self.assertRaises(ValueError,
-       #     lambda: hblock.disconnect("src1", 0, "nop1", 1))
-       """
-               
+       nop1 = gr.nop(gr.sizeof_int)
+       hblock.connect(nop1, hblock)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(nop1, (hblock, 1)))
+
+    def test_020_run(self):
+       hblock = gr.top_block("test_block")
+       data = (1.0, 2.0, 3.0, 4.0)
+       src = gr.vector_source_f(data, False)
+       dst = gr.vector_sink_f()
+       hblock.connect(src, dst)
+       hblock.run()
+       self.assertEquals(data, dst.data())
+
 if __name__ == "__main__":
     gr_unittest.main()
Debian packaging of gnuradio