diff options
| author | Martin Braun <martin.braun@ettus.com> | 2021-03-03 16:51:04 +0100 |
|---|---|---|
| committer | mormj <34754695+mormj@users.noreply.github.com> | 2021-03-03 13:39:45 -0500 |
| commit | 6f1cac5d60a19f7f127c35d83eb369f0a99189cd (patch) | |
| tree | 89321d30338e8e15ef2fb20cd5fd877b58a3d8c8 /gnuradio-runtime/python/gnuradio/gr_unittest.py | |
| parent | 78dc0d8b5b289b108deebce3567b69902d65d202 (diff) | |
runtime: gr_unittest: Add waitFor() API call
This is a helper for adding conditional waits into test cases. Useful
for tests with a non-deterministic execution time.
Signed-off-by: Martin Braun <martin.braun@ettus.com>
Diffstat (limited to 'gnuradio-runtime/python/gnuradio/gr_unittest.py')
| -rw-r--r-- | gnuradio-runtime/python/gnuradio/gr_unittest.py | 41 |
1 files changed, 41 insertions, 0 deletions
diff --git a/gnuradio-runtime/python/gnuradio/gr_unittest.py b/gnuradio-runtime/python/gnuradio/gr_unittest.py index f82355338e..6177f0f32f 100644 --- a/gnuradio-runtime/python/gnuradio/gr_unittest.py +++ b/gnuradio-runtime/python/gnuradio/gr_unittest.py @@ -11,8 +11,12 @@ GNU radio specific extension of unittest. """ +import time import unittest +# We allow snakeCase here for consistency with unittest +# pylint: disable=invalid-name + class TestCase(unittest.TestCase): """A subclass of unittest.TestCase that adds additional assertions @@ -107,6 +111,43 @@ class TestCase(unittest.TestCase): for (x, y) in zip(a, b) ]) + def waitFor( + self, + condition, + timeout=5.0, + poll_interval=0.2, + fail_on_timeout=True, + fail_msg=None): + """ + Helper function: Wait for a callable to return True within a given + timeout. + + This is useful for running tests where an exact wait time is not known. + + Arguments: + - condition: A callable. Must return True when a 'good' condition is met. + - timeout: Timeout in seconds. `condition` must return True within this + timeout. + - poll_interval: Time between calls to condition() in seconds + - fail_on_timeout: If True, the test case will fail when the timeout + occurs. If False, this function will return False in + that case. + - fail_msg: The message that is printed when a timeout occurs and + fail_on_timeout is true. + """ + if not callable(condition): + self.fail("Invalid condition provided to waitFor()!") + stop_time = time.monotonic() + timeout + while time.monotonic() <= stop_time: + if condition(): + return True + time.sleep(poll_interval) + if fail_on_timeout: + fail_msg = fail_msg or "Timeout exceeded during call to waitFor()!" + self.fail(fail_msg) + return False + + TestResult = unittest.TestResult TestSuite = unittest.TestSuite FunctionTestCase = unittest.FunctionTestCase |