[launch_pytest] Modify how wait_for_output()/wait_for_stderr() work, add assert_*() alternatives (#553)

Signed-off-by: Ivan Santiago Paunovic <[email protected]>

Author: ivanpauno

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',

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())

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):

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)