Refactors launch_testing API a bit.

To cope with more test cases.

Signed-off-by: Michel Hidalgo <[email protected]>

launch/launch/actions/execute_process.py

@@ -68,6 +68,14 @@
 _global_process_counter = 0  # in Python3, this number is unbounded (no rollover)
 
 
+def _is_process_running(pid):
+    try:
+        os.kill(pid, 0)
+        return True
+    except OSError:
+        return False
+
+
 class ExecuteProcess(Action):
     """Action that begins executing a process and sets up event handlers for the process."""
 
@@ -246,7 +254,10 @@ def __on_signal_process_event(
             raise RuntimeError('Signal event received before execution.')
         if self._subprocess_transport is None:
             raise RuntimeError('Signal event received before subprocess transport available.')
-        if self._subprocess_protocol.complete.done():
+        # if self._subprocess_protocol.complete.done():
+        # disable above's check as this handler may get called *after* the process has
+        # terminated but *before* the asyncio future has been resolved.
+        if not _is_process_running(self._subprocess_transport.get_pid()):
             # the process is done or is cleaning up, no need to signal
             _logger.debug("signal '{}' not set to '{}' because it is already closing".format(
                 typed_event.signal_name, self.process_details['name']

launch_testing/launch_testing/__init__.py

@@ -17,13 +17,12 @@
 from launch.actions import EmitEvent
 from launch.actions import ExecuteProcess
 from launch.actions import RegisterEventHandler
+from launch.actions import UnregisterEventHandler
 from launch.event_handlers import OnExecutionComplete
 from launch.event_handlers import OnProcessExit
 from launch.event_handlers import OnProcessIO
 from launch.events import Shutdown
 
-from .output import create_output_check
-
 
 class LaunchTestService():
 
@@ -49,22 +48,27 @@ def _fail(
         for test_name in self.__tests:
             if self.__tests[test_name] == 'armed':
                 self.__tests[test_name] = 'dropped'
-        return EmitEvent(event=Shutdown(reason=reason))
+        return [EmitEvent(event=Shutdown(reason=reason))]
 
     def _succeed(
         self,
-        test_name
+        test_name,
+        side_effect=None
     ):
         """Mark test as a success and shutdown if all other tests have succeeded too."""
         self.__tests[test_name] = 'succeeded'
         if all(status == 'succeeded' for status in self.__tests.values()):
-            return EmitEvent(event=Shutdown(reason='all tests finished'))
+            return [EmitEvent(event=Shutdown(reason='all tests finished'))]
+        if side_effect == 'shutdown':
+            return [EmitEvent(event=Shutdown(reason='shutdown after test'))]
+        return []
 
     def add_fixture_action(
         self,
         launch_description,
         action,
-        required=False
+        exit_allowed=[0],
+        ignore_returncode=False
     ):
         """
         Add action used as testing fixture.
@@ -75,9 +79,13 @@ def add_fixture_action(
         if isinstance(action, ExecuteProcess):
             def on_fixture_process_exit(event, context):
                 process_name = event.action.process_details['name']
-                if required or event.returncode != 0:
-                    rc = event.returncode if event.returncode else 1
-                    self.__processes_rc[process_name] = rc
+                allowed_to_exit = exit_allowed
+                if isinstance(exit_allowed, list):
+                    allowed_to_exit = event.returncode in exit_allowed
+                if not allowed_to_exit:
+                    if not ignore_returncode:
+                        rc = event.returncode if event.returncode else 1
+                        self.__processes_rc[process_name] = rc
                     return EmitEvent(event=Shutdown(
                         reason='{} fixture process died!'.format(process_name)
                     ))
@@ -133,60 +141,69 @@ def add_output_test(
         self,
         launch_description,
         action,
-        output_file,
-        filtered_prefixes=None,
-        filtered_patterns=None,
-        filtered_rmw_implementation=None
+        output_test,
+        test_suffix='output',
+        output_filter=None,
+        side_effect=None,
     ):
         """
-        Test an action process' output against text or regular expressions.
+        Test an action process' output against a given test.
 
         :param launch_description: test launch description that owns the given action.
         :param action: launch action to test whose output is to be tested.
-        :param output_file: basename (i.e. w/o extension) of either a .txt file containing the
-        lines to be matched or a .regex file containing patterns to be searched for.
-        :param filtered_prefixes: A list of byte strings representing prefixes that will cause
-        output lines to be ignored if they start with one of the prefixes. By default lines
-        starting with the process ID (`b'pid'`) and return code (`b'rc'`) will be ignored.
-        :param filtered_patterns: A list of byte strings representing regexes that will cause
-        output lines to be ignored if they match one of the regexes.
-        :param filtered_rmw_implementation: RMW implementation for which the output will be
-        ignored in addition to the `filtered_prefixes`/`filtered_patterns`.
+        :param output_test: test tuple as returned by launch_testing.output.create_* functions.
+        :param test_suffix: an optional test suffix to disambiguate multiple test instances, defaults
+        to 'output'.
+        :param output_filter: an optional function to filter out i.e. ignore output lines for the test.
+        :param side_effect: an optional side effect of a passing test, currently only 'shutdown'
+        is supported.
         """
         assert isinstance(action, ExecuteProcess)
-        test_name = 'test_{}_output'.format(id(action))
-        output, collate_output, match_output, match_patterns = create_output_check(
-            output_file, filtered_prefixes, filtered_patterns, filtered_rmw_implementation
-        )
+        test_name = 'test_{}_{}'.format(id(action), test_suffix)
+        output, collate_output, match_output, match_patterns = output_test
+        if not output_filter:
+            output_filter = (lambda x: x)
         assert any(match_patterns)
 
         def on_process_exit(event, context):
             nonlocal match_patterns
             if any(match_patterns):
                 process_name = event.action.process_details['name']
                 reason = 'not all {} output matched!'.format(process_name)
-                return self._fail(test_name, reason)
+                return [
+                    UnregisterEventHandler(on_output),
+                    UnregisterEventHandler(on_exit),
+                    *self._fail(test_name, reason)
+                ]
+
+        on_exit = OnProcessExit(
+            target_action=action, on_exit=on_process_exit
+        )
+
+        launch_description.add_action(
+            RegisterEventHandler(on_exit)
+        )
 
         def on_process_stdout(event):
             nonlocal output
             nonlocal match_patterns
-            output = collate_output(output, event.text)
+            output = collate_output(output, output_filter(event.text))
             match_patterns = [
                 pattern for pattern in match_patterns
                 if not match_output(output, pattern)
             ]
             if not any(match_patterns):
-                return self._succeed(test_name)
-
-        launch_description.add_action(
-            RegisterEventHandler(OnProcessExit(
-                target_action=action, on_exit=on_process_exit
-            ))
+                return [
+                    UnregisterEventHandler(on_output),
+                    UnregisterEventHandler(on_exit),
+                    *self._succeed(test_name, side_effect)
+                ]
+
+        on_output = OnProcessIO(
+            target_action=action, on_stdout=on_process_stdout
         )
         launch_description.add_action(
-            RegisterEventHandler(OnProcessIO(
-                target_action=action, on_stdout=on_process_stdout
-            ))
+            RegisterEventHandler(on_output)
         )
         self._arm(test_name)
 

launch_testing/launch_testing/output.py

@@ -32,19 +32,22 @@ def get_rmw_output_filter(rmw_implementation, filter_type):
     return [str.encode(l) for l in rmw_output_filter.splitlines()]
 
 
-def get_expected_output(output_file):
-    literal_file = output_file + '.txt'
-    if os.path.isfile(literal_file):
-        with open(literal_file, 'rb') as f:
-            return f.read().splitlines()
-    regex_file = output_file + '.regex'
-    if os.path.isfile(regex_file):
-        with open(regex_file, 'rb') as f:
-            return f.read().splitlines()
-
-
-def create_output_lines_filter(filtered_prefixes, filtered_patterns,
-                               filtered_rmw_implementation):
+def create_output_lines_filter(
+    filtered_prefixes=None,
+    filtered_patterns=None,
+    filtered_rmw_implementation=None
+):
+    """
+    Create a line filtering function to help output testing.
+
+    :param filtered_prefixes: A list of byte strings representing prefixes that will cause
+    output lines to be ignored if they start with one of the prefixes. By default lines
+    starting with the process ID (`b'pid'`) and return code (`b'rc'`) will be ignored.
+    :param filtered_patterns: A list of byte strings representing regexes that will cause
+    output lines to be ignored if they match one of the regexes.
+    :param filtered_rmw_implementation: RMW implementation for which the output will be
+    ignored in addition to the `filtered_prefixes`/`filtered_patterns`.
+    """
     filtered_prefixes = filtered_prefixes or get_default_filtered_prefixes()
     filtered_patterns = filtered_patterns or get_default_filtered_patterns()
     if filtered_rmw_implementation:
@@ -57,49 +60,68 @@ def create_output_lines_filter(filtered_prefixes, filtered_patterns,
     filtered_patterns = map(re.compile, filtered_patterns)
 
     def _filter(output):
+        filtered_output = []
         for line in output.splitlines():
             # Filter out stdout that comes from underlying DDS implementation
             # Note: we do not currently support matching filters across multiple stdout lines.
             if any(line.startswith(prefix) for prefix in filtered_prefixes):
                 continue
             if any(pattern.match(line) for pattern in filtered_patterns):
                 continue
-            yield line
+            filtered_output.append(line)
+        if output.endswith(b'\n'):
+            filtered_output.append(b'\n')
+        return b'\n'.join(filtered_output)
     return _filter
 
 
-def create_output_check(output_file, filtered_prefixes, filtered_patterns,
-                        filtered_rmw_implementation):
-    filter_output_lines = create_output_lines_filter(
-        filtered_prefixes, filtered_patterns, filtered_rmw_implementation
-    )
+def create_output_lines_test(expected_lines):
+    """
+    Create output test given a list of expected lines.
+    """
+    def _collate(output, addendum):
+        output.extend(addendum.splitlines())
+        return output
 
-    literal_file = output_file + '.txt'
-    if os.path.isfile(literal_file):
-        def _collate(output, addendum):
-            output.extend(filter_output_lines(addendum))
-            return output
+    def _match(output, pattern):
+        print(output, pattern, pattern in output)
+        return any(pattern in line for line in output)
 
-        def _match(output, pattern):
-            return pattern in output
+    return [], _collate, _match, expected_lines
 
+
+def create_output_regex_test(expected_patterns):
+    """
+    Create output test given a list of expected matching regular
+    expressions.
+    """
+    def _collate(output, addendum):
+        output.write(addendum)
+        return output
+
+    def _match(output, pattern):
+        return pattern.search(output.getvalue()) is not None
+
+    return io.BytesIO(), _collate, _match, expected_patterns
+
+
+def create_output_test_from_file(output_file):
+    """
+    Create output test using the given file content.
+
+    :param output_file: basename (i.e. w/o extension) of either a .txt file containing the
+    lines to be matched or a .regex file containing patterns to be searched for.
+    """
+    literal_file = output_file + '.txt'
+    if os.path.isfile(literal_file):
         with open(literal_file, 'rb') as f:
             expected_output = f.read().splitlines()
-        return [], _collate, _match, expected_output
+        return create_output_lines_test(expected_output)
 
     regex_file = output_file + '.regex'
     if os.path.isfile(regex_file):
-        def _collate(output, addendum):
-            output.write(b'\n'.join(
-                filter_output_lines(addendum)
-            ))
-            return output
-
-        def _match(output, pattern):
-            return pattern.search(output.getvalue()) is not None
-
         with open(regex_file, 'rb') as f:
             patterns = [re.compile(regex) for regex in f.read().splitlines()]
-        return io.BytesIO(), _collate, _match, patterns
+        return create_output_regex_test(patterns)
 
     raise RuntimeError('could not find output check file: {}'.format(output_file))

launch_testing/test/test_matching.py

@@ -20,7 +20,7 @@
 from launch import LaunchService
 from launch.actions import ExecuteProcess
 from launch_testing import LaunchTestService
-
+from launch_testing.output import create_output_test_from_file
 
 def test_matching():
     # This temporary directory and files contained in it
@@ -42,7 +42,9 @@ def test_matching():
     action = launch_test.add_fixture_action(
         ld, ExecuteProcess(cmd=executable_command, output='screen')
     )
-    launch_test.add_output_test(ld, action, output_file)
+    launch_test.add_output_test(
+        ld, action, create_output_test_from_file(output_file)
+    )
 
     launch_service = LaunchService()
     launch_service.include_launch_description(ld)