1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
#!/usr/bin/env python
#
# Copyright 2018 Free Software Foundation, Inc.
#
# This file is part of GNU Radio
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
#
from gnuradio import gr, gr_unittest, blocks
from multiprocessing import Process
import numpy
# This block just exists to pass data through and throw an exception
class except_block(gr.sync_block):
def __init__(self, throw_except, except_count=10000):
gr.sync_block.__init__(
self,
name="except_block",
in_sig=[numpy.complex64],
out_sig=[numpy.complex64],
)
self.throw_except = throw_except
self.except_count = except_count
self.count = 0
def work(self, input_items, output_items):
output_items[0][:] = input_items[0]
self.count += len(output_items[0])
if self.count >= self.except_count:
raise RuntimeError("Error in except_block")
return len(output_items[0])
def process_func(catch_exceptions):
tb = gr.top_block(catch_exceptions=catch_exceptions)
# some test data
src_data = [complex(x, x + 1) for x in range(65536)]
src = blocks.vector_source_c(src_data)
src.set_repeat(True)
e_block_1 = except_block(False)
e_block_2 = except_block(True)
sink_1 = blocks.null_sink(gr.sizeof_gr_complex)
sink_2 = blocks.null_sink(gr.sizeof_gr_complex)
tb.connect(src, e_block_1)
tb.connect(src, e_block_2)
tb.connect(e_block_1, sink_1)
tb.connect(e_block_2, sink_2)
tb.run()
class test_uncaught_exception(gr_unittest.TestCase):
def test_exception_throw_uncaught(self):
# Test to ensure that throwing an exception causes the
# process running top_block to exit
p = Process(target=process_func, args=(False,))
p.daemon = True
p.start()
p.join(2.5)
exit_code = p.exitcode
self.assertIsNotNone(
exit_code, "exception did not cause flowgraph exit")
def test_exception_throw_caught(self):
# Test to ensure that throwing an exception does not cause the process
# running top_block to exit (in catch_exceptions mode)
p = Process(target=process_func, args=(True,))
p.daemon = True
p.start()
p.join(2.5)
exit_code = p.exitcode
self.assertIsNone(
exit_code, "exception caused flowgraph exit")
if __name__ == '__main__':
gr_unittest.run(test_uncaught_exception)
|
