diff --git a/launch_pytest/launch_pytest/tools/__init__.py b/launch_pytest/launch_pytest/tools/__init__.py index 14279b69b..6a7215950 100644 --- a/launch_pytest/launch_pytest/tools/__init__.py +++ b/launch_pytest/launch_pytest/tools/__init__.py @@ -13,6 +13,10 @@ # limitations under the License. from . import process +from .process import assert_output +from .process import assert_output_sync +from .process import assert_stderr +from .process import assert_stderr_sync from .process import wait_for_exit from .process import wait_for_exit_sync from .process import wait_for_output @@ -23,6 +27,10 @@ from .process import wait_for_stderr_sync __all__ = [ + 'assert_output', + 'assert_output_sync', + 'assert_stderr', + 'assert_stderr_sync', 'process', 'wait_for_exit', 'wait_for_exit_sync', diff --git a/launch_pytest/launch_pytest/tools/process.py b/launch_pytest/launch_pytest/tools/process.py index 4479984db..274044bf0 100644 --- a/launch_pytest/launch_pytest/tools/process.py +++ b/launch_pytest/launch_pytest/tools/process.py @@ -47,11 +47,7 @@ async def _wait_for_event_with_condition( ): pyevent = asyncio.Event() event_handler = get_launch_event_handler(execute_process_action, pyevent) - cond_value = False - try: - cond_value = condition() - except AssertionError: - pass + cond_value = condition() with register_event_handler(launch_context, event_handler): start = time.time() now = start @@ -61,15 +57,9 @@ async def _wait_for_event_with_condition( except asyncio.TimeoutError: break pyevent.clear() - try: - cond_value = condition() - except AssertionError: - pass + cond_value = condition() now = time.time() - # Call condition() again, if before it returned False. - # If assertions were being used and the condition is still not satisfied it should raise here, - # pytest renders assertion errors nicely. - return condition() if not cond_value else cond_value + return cond_value def _wait_for_event_sync( @@ -86,26 +76,16 @@ def _wait_for_event_with_condition_sync( ): pyevent = threading.Event() event_handler = get_launch_event_handler(execute_process_action, pyevent) - cond_value = False - try: - cond_value = condition() - except AssertionError: - pass # Allow asserts in the condition closures + cond_value = condition() with register_event_handler(launch_context, event_handler): start = time.time() now = start while not cond_value and (timeout is None or now < start + timeout): pyevent.wait(start - now + timeout) pyevent.clear() - try: - cond_value = condition() - except AssertionError: - pass + cond_value = condition() now = time.time() - # Call condition() again, if before it returned False. - # If assertions were being used and the condition is still not satisfied it should raise here, - # pytest renders assertion errors nicely. - return condition() if not cond_value else cond_value + return cond_value def _get_stdout_event_handler(action, pyevent): @@ -124,6 +104,25 @@ async def wait_for_output( timeout) +async def assert_output( + launch_context, execute_process_action, validate_output, timeout=None +): + def condition(): + try: + validate_output(execute_process_action.get_stdout()) + except AssertionError: + return False + return True + cond_value = await _wait_for_event_with_condition( + launch_context, + execute_process_action, + _get_stdout_event_handler, + condition, + timeout) + if not cond_value: + validate_output(execute_process_action.get_stdout()) + + def wait_for_output_sync( launch_context, execute_process_action, validate_output, timeout=None ): @@ -135,6 +134,25 @@ def wait_for_output_sync( timeout) +def assert_output_sync( + launch_context, execute_process_action, validate_output, timeout=None +): + def condition(): + try: + validate_output(execute_process_action.get_stdout()) + except AssertionError: + return False + return True + cond_value = _wait_for_event_with_condition_sync( + launch_context, + execute_process_action, + _get_stdout_event_handler, + condition, + timeout) + if not cond_value: + validate_output(execute_process_action.get_stdout()) + + def _get_stderr_event_handler(action, pyevent): return event_handlers.OnProcessIO( target_action=action, on_stderr=lambda _1: pyevent.set()) @@ -151,6 +169,25 @@ async def wait_for_stderr( timeout) +async def assert_stderr( + launch_context, execute_process_action, validate_output, timeout=None +): + def condition(): + try: + validate_output(execute_process_action.get_stderr()) + except AssertionError: + return False + return True + cond_value = await _wait_for_event_with_condition( + launch_context, + execute_process_action, + _get_stderr_event_handler, + condition, + timeout) + if not cond_value: + validate_output(execute_process_action.get_stderr()) + + def wait_for_stderr_sync( launch_context, execute_process_action, validate_output, timeout=None ): @@ -162,6 +199,25 @@ def wait_for_stderr_sync( timeout) +def assert_stderr_sync( + launch_context, execute_process_action, validate_output, timeout=None +): + def condition(): + try: + validate_output(execute_process_action.get_stderr()) + except AssertionError: + return False + return True + cond_value = _wait_for_event_with_condition_sync( + launch_context, + execute_process_action, + _get_stderr_event_handler, + condition, + timeout) + if not cond_value: + validate_output(execute_process_action.get_stderr()) + + def _get_on_process_start_event_handler(execute_process_action, pyevent): return event_handlers.OnProcessStart( target_action=execute_process_action, on_start=lambda _1, _2: pyevent.set()) diff --git a/launch_pytest/test/launch_pytest/examples/pytest_hello_world.py b/launch_pytest/test/launch_pytest/examples/pytest_hello_world.py index ced833c42..805a6516f 100644 --- a/launch_pytest/test/launch_pytest/examples/pytest_hello_world.py +++ b/launch_pytest/test/launch_pytest/examples/pytest_hello_world.py @@ -50,8 +50,7 @@ def validate_output(output): # this function can use assertions to validate the output or return a boolean. # pytest generates easier to understand failures when assertions are used. assert output.splitlines() == ['hello_world'], 'process never printed hello_world' - return True - assert process_tools.wait_for_output_sync( + process_tools.assert_output_sync( launch_context, hello_world_proc, validate_output, timeout=5) def validate_output(output): diff --git a/launch_pytest/test/launch_pytest/tools/test_process.py b/launch_pytest/test/launch_pytest/tools/test_process.py index 7d1de121e..8ba152801 100644 --- a/launch_pytest/test/launch_pytest/tools/test_process.py +++ b/launch_pytest/test/launch_pytest/tools/test_process.py @@ -52,14 +52,12 @@ async def test_async_process_tools(dut, launch_context): def check_output(output): assert output.splitlines() == ['hello'] - return True - assert await tools.wait_for_output( + await tools.assert_output( launch_context, dut, check_output, timeout=10) def check_stderr(err): assert err.splitlines() == ['world'] - return True - assert await tools.wait_for_stderr( + await tools.assert_stderr( launch_context, dut, check_stderr, timeout=10) assert await tools.wait_for_exit(launch_context, dut, timeout=10) @@ -70,13 +68,11 @@ def test_sync_process_tools(dut, launch_context): def check_output(output): assert output.splitlines() == ['hello'] - return True - assert tools.wait_for_output_sync( + tools.assert_output_sync( launch_context, dut, check_output, timeout=10) def check_stderr(err): assert err.splitlines() == ['world'] - return True - assert tools.wait_for_stderr_sync( + tools.assert_stderr_sync( launch_context, dut, check_stderr, timeout=10) assert tools.wait_for_exit_sync(launch_context, dut, timeout=10)