Applies misc fixes after Windows triaging.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

launch/launch/actions/execute_process.py

@@ -68,14 +68,6 @@
 _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."""
 
@@ -254,12 +246,9 @@ 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():
-        # 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()):
+        if self._subprocess_protocol.complete.done():
             # the process is done or is cleaning up, no need to signal
-            _logger.debug("signal '{}' not set to '{}' because it is already closing".format(
+            _logger.debug("signal '{}' not sent to '{}' because it is already closing".format(
                 typed_event.signal_name, self.process_details['name']
             ))
             return None
@@ -274,11 +263,16 @@ def __on_signal_process_event(
         _logger.info("sending signal '{}' to process[{}]".format(
             typed_event.signal_name, self.process_details['name']
         ))
-        if typed_event.signal_name == 'SIGKILL':
-            self._subprocess_transport.kill()  # works on both Windows and POSIX
+        try:
+            if typed_event.signal_name == 'SIGKILL':
+                self._subprocess_transport.kill()  # works on both Windows and POSIX
+                return None
+            self._subprocess_transport.send_signal(typed_event.signal)
             return None
-        self._subprocess_transport.send_signal(typed_event.signal)
-        return None
+        except ProcessLookupError:
+            _logger.debug("signal '{}' not sent to '{}' because it has closed already".format(
+                typed_event.signal_name, self.process_details['name']
+            ))
 
     def __on_process_stdin_event(
         self,

launch/launch/event_handlers/on_execution_complete.py

@@ -110,7 +110,7 @@ def handler_description(self) -> Text:
         #                    via the 'entities' property.
         if self.__actions_on_completion:
             return '<actions>'
-        return '{}'.format(self.__on_exit)
+        return '{}'.format(self.__on_completion)
 
     @property
     def matcher_description(self) -> Text:

launch/launch/launch_introspector.py

@@ -110,7 +110,7 @@ def format_action(action: Action) -> List[Text]:
             ),
             typed_action.env if typed_action.env is None else '{' + ', '.join(
                 ['{}: {}'.format(format_substitutions(k), format_substitutions(v))
-                 for k, v in typed_action.env.items()]) + '}',
+                 for k, v in typed_action.env]) + '}',
             typed_action.shell,
         )
         return [msg]

launch_testing/launch_testing/__init__.py

@@ -48,6 +48,14 @@ def _finish(
         if all(status != 'armed' for status in self.__tests.values()):
             return [EmitEvent(event=Shutdown(reason='all tests finished'))]
 
+    def _drop(
+        self,
+        test_name
+    ):
+        """Mark test as dropped."""
+        assert test_name in self.__tests
+        self.__tests[test_name] = 'dropped'
+
     def _fail(
         self,
         test_name,
@@ -122,7 +130,7 @@ def add_test_action(
             def on_test_process_exit(event, context):
                 if event.returncode != 0:
                     process_name = event.action.process_details['name']
-                    self._processes_rc[process_name] = event.returncode
+                    self.__processes_rc[process_name] = event.returncode
                     return self._fail(
                         test_name, reason='{} test failed!'.format(
                             process_name
@@ -136,11 +144,19 @@ def on_test_process_exit(event, context):
                 ))
             )
         else:
+            def on_test_completion(event, context):
+                future = event.action.get_asyncio_future()
+                if future is not None:
+                    if future.cancelled():
+                        return self._drop(test_name)
+                    exc = future.exception()
+                    if exc is not None:
+                         return self._fail(test_name, str(exc))
+                return self._succeed(test_name)
+
             launch_description.add_action(
                 RegisterEventHandler(OnExecutionComplete(
-                    target_action=action, on_completion=(
-                        lambda *args: self._succeed(test_name)
-                    )
+                    target_action=action, on_completion=on_test_completion
                 ))
             )
         launch_description.add_action(action)

launch_testing/launch_testing/output.py

@@ -58,6 +58,7 @@ def create_output_lines_filter(
             filtered_rmw_implementation, 'patterns'
         ))
     filtered_patterns = map(re.compile, filtered_patterns)
+    encoded_line_sep = os.linesep.encode('ascii')
 
     def _filter(output):
         filtered_output = []
@@ -69,23 +70,22 @@ def _filter(output):
             if any(pattern.match(line) for pattern in filtered_patterns):
                 continue
             filtered_output.append(line)
-        if output.endswith(b'\n'):
-            filtered_output.append(b'\n')
-        return b'\n'.join(filtered_output)
+        if output.endswith(encoded_line_sep):
+            filtered_output.append(encoded_line_sep)
+        return encoded_line_sep.join(filtered_output)
     return _filter
 
 
 def create_output_lines_test(expected_lines):
     """Create output test given a list of expected lines."""
     def _collate(output, addendum):
-        output.extend(addendum.splitlines())
+        output.write(addendum)
         return output
 
     def _match(output, pattern):
-        print(output, pattern, pattern in output)
-        return any(pattern in line for line in output)
+        return any(pattern in line for line in output.getvalue().splitlines())
 
-    return [], _collate, _match, expected_lines
+    return io.BytesIO(), _collate, _match, expected_lines
 
 
 def create_output_regex_test(expected_patterns):