Unit test for exception raised in custom GNU radio python block

Question

I have created a custom python sync block for use in a gnuradio flowgraph. The block tests for invalid input and, if found, raises a ValueError exception. I would like to create a unit test to verify that the exception is raised when the block indeed receives invalid input data.

As part of the python-based qa test for this block, I created a flowgraph such that the block receives invalid data. When I run the test, the block does appear to raise the exception but then hangs.

What is the appropriate way to test for this? Here is a minimal working example:

#!/usr/bin/env python

import numpy as np

from gnuradio import gr, gr_unittest, blocks

class validate_input(gr.sync_block):

    def __init__(self):
        gr.sync_block.__init__(self,
            name="validate_input",
            in_sig=[np.float32],
            out_sig=[np.float32])

        self.max_input = 100

    def work(self, input_items, output_items):
        in0 = input_items[0]

        if (np.max(in0) > self.max_input):
           raise ValueError('input exceeds max.')

        validated_in = output_items[0]
        validated_in[:] = in0

        return len(output_items[0])

class qa_validate_input (gr_unittest.TestCase):

    def setUp (self):
        self.tb = gr.top_block ()

    def tearDown (self):
        self.tb = None

    def test_check_valid_data(self):

        src_data = (0, 201, 92)

        src = blocks.vector_source_f(src_data)
        validate = validate_input()
        snk = blocks.vector_sink_f()

        self.tb.connect (src, validate)
        self.tb.connect (validate, snk)

        self.assertRaises(ValueError, self.tb.run)

if __name__ == '__main__':
    gr_unittest.run(qa_validate_input, "qa_validate_input.xml")

which produces:

DEPRECATED: Using filename with gr_unittest does no longer have any effect.
handler caught exception: input exceeds max.
Traceback (most recent call last):
  File "/home/xxx/devel/gnuradio3_8/lib/python3.6/dist-packages/gnuradio/gr/gateway.py", line 60, in eval
    try: self._callback()
  File "/home/xxx/devel/gnuradio3_8/lib/python3.6/dist-packages/gnuradio/gr/gateway.py", line 230, in __gr_block_handle
    ) for i in range(noutputs)],
  File "qa_validate_input.py", line 21, in work
    raise ValueError('input exceeds max.')
ValueError: input exceeds max.
thread[thread-per-block[1]: <block validate_input(2)>]: SWIG director method error. Error detected when calling 'feval_ll.eval'
^CF
======================================================================
FAIL: test_check_valid_data (__main__.qa_validate_input)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "qa_validate_input.py", line 47, in test_check_valid_data
    self.assertRaises(ValueError, self.tb.run)
AssertionError: ValueError not raised by run

----------------------------------------------------------------------
Ran 1 test in 1.634s

FAILED (failures=1)

Answer 1

The top_block's run() function does not call the block's work() function directly but starts the internal task scheduler and its threads and waits them to finish.

One way to unit test the error handling in your block is to call the work() function directly

    def test_check_valid_data(self):
        src_data = [[0, 201, 92]]
        output_items = [[]]

        validate = validate_input()
        self.assertRaises(ValueError, lambda: validate.work(src_data, output_items))