diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2017-07-13 13:59:16 -0700 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2017-07-13 13:59:16 -0700 |
| commit | 7d29847824e68966de9dbeada02fa1d9a22ee8da (patch) | |
| tree | d20e3b8490402cf20f3633707794d2ef38a94242 /gr-blocks/python | |
| parent | b629e3faa419a2fbd1d4f0a1563a28748dfc9403 (diff) | |
| parent | 19c0aeea4320e40aea11564879e3d2b9c0ca0d21 (diff) | |
Merge branch 'next' into python3
Diffstat (limited to 'gr-blocks/python')
| -rw-r--r-- | gr-blocks/python/blocks/qa_file_source_sink.py | 62 |
1 files changed, 62 insertions, 0 deletions
diff --git a/gr-blocks/python/blocks/qa_file_source_sink.py b/gr-blocks/python/blocks/qa_file_source_sink.py index 30c7708a3c..e39a34ad82 100644 --- a/gr-blocks/python/blocks/qa_file_source_sink.py +++ b/gr-blocks/python/blocks/qa_file_source_sink.py @@ -24,6 +24,7 @@ from gnuradio import gr, gr_unittest, blocks import os import tempfile +import pmt class test_file_source_sink(gr_unittest.TestCase): @@ -56,6 +57,7 @@ class test_file_source_sink(gr_unittest.TestCase): result_data = snk2.data() self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk2.tags()), 0) def test_descriptor_001(self): src_data = list(range(1000)) @@ -87,6 +89,7 @@ class test_file_source_sink(gr_unittest.TestCase): result_data = snk2.data() self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk2.tags()), 0) def test_file_source_can_seek_after_open(self): src_data = list(range(1000)) @@ -102,6 +105,65 @@ class test_file_source_sink(gr_unittest.TestCase): source = blocks.file_source(gr.sizeof_float, temp.name) self.assertTrue(source.seek(0, os.SEEK_SET)) + def test_begin_tag(self): + src_data = range(1000) + expected_result = range(1000) + + snk2 = blocks.vector_sink_f() + + with tempfile.NamedTemporaryFile() as temp: + src = blocks.vector_source_f(src_data) + snk = blocks.file_sink(gr.sizeof_float, temp.name) + snk.set_unbuffered(True) + + src2 = blocks.file_source(gr.sizeof_float, temp.name) + src2.set_begin_tag(pmt.string_to_symbol("file_begin")) + + self.tb.connect(src, snk) + self.tb.run() + + self.tb.disconnect(src, snk) + self.tb.connect(src2, snk2) + self.tb.run() + + result_data = snk2.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk2.tags()), 1) + + def test_begin_tag_repeat(self): + src_data = range(1000) + expected_result = range(1000) + expected_result.extend(range(1000)) + + snk2 = blocks.vector_sink_f() + + with tempfile.NamedTemporaryFile() as temp: + src = blocks.vector_source_f(src_data) + snk = blocks.file_sink(gr.sizeof_float, temp.name) + snk.set_unbuffered(True) + + src2 = blocks.file_source(gr.sizeof_float, temp.name, True) + src2.set_begin_tag(pmt.string_to_symbol("file_begin")) + hd = blocks.head(gr.sizeof_float, 2000) + + self.tb.connect(src, snk) + self.tb.run() + + self.tb.disconnect(src, snk) + self.tb.connect(src2, hd, snk2) + self.tb.run() + + result_data = snk2.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + tags = snk2.tags() + self.assertEqual(len(tags), 2) + self.assertEqual(str(tags[0].key), "file_begin") + self.assertEqual(str(tags[0].value), "0") + self.assertEqual(tags[0].offset, 0) + self.assertEqual(str(tags[1].key), "file_begin") + self.assertEqual(str(tags[1].value), "1") + self.assertEqual(tags[1].offset, 1000) + if __name__ == '__main__': gr_unittest.run(test_file_source_sink, "test_file_source_sink.xml") |