Support non-interactive launch.LaunchService runs (#475)

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

launch/launch/actions/execute_process.py

@@ -577,9 +577,10 @@ def __flush_buffers(self, event, context):
             self.__stderr_buffer.truncate(0)
 
     def __on_shutdown(self, event: Event, context: LaunchContext) -> Optional[SomeActionsType]:
+        due_to_sigint = cast(Shutdown, event).due_to_sigint
         return self._shutdown_process(
             context,
-            send_sigint=(not cast(Shutdown, event).due_to_sigint),
+            send_sigint=not due_to_sigint or context.noninteractive,
         )
 
     def __get_shutdown_timer_actions(self) -> List[Action]:

launch/launch/launch_context.py

@@ -33,13 +33,21 @@
 class LaunchContext:
     """Runtime context used by various launch entities when being visited or executed."""
 
-    def __init__(self, *, argv: Optional[Iterable[Text]] = None) -> None:
+    def __init__(
+        self,
+        *,
+        argv: Optional[Iterable[Text]] = None,
+        noninteractive: bool = False
+    ) -> None:
         """
         Create a LaunchContext.
 
         :param: argv stored in the context for access by the entities, None results in []
+        :param: noninteractive if True (not default), this service will assume it has
+            no terminal associated e.g. it is being executed from a non interactive script
         """
         self.__argv = argv if argv is not None else []
+        self.__noninteractive = noninteractive
 
         self._event_queue = asyncio.Queue()  # type: asyncio.Queue
         self._event_handlers = collections.deque()  # type: collections.deque
@@ -63,6 +71,11 @@ def argv(self):
         """Getter for argv."""
         return self.__argv
 
+    @property
+    def noninteractive(self):
+        """Getter for noninteractive."""
+        return self.__noninteractive
+
     def _set_is_shutdown(self, state: bool) -> None:
         self.__is_shutdown = state
 

launch/launch/launch_service.py

@@ -56,6 +56,7 @@ def __init__(
         self,
         *,
         argv: Optional[Iterable[Text]] = None,
+        noninteractive: bool = False,
         debug: bool = False
     ) -> None:
         """
@@ -67,6 +68,8 @@ def __init__(
         outside of the main-thread.
 
         :param: argv stored in the context for access by the entities, None results in []
+        :param: noninteractive if True (not default), this service will assume it has
+            no terminal associated e.g. it is being executed from a non interactive script
         :param: debug if True (not default), asyncio the logger are seutp for debug
         """
         # Setup logging and debugging.
@@ -82,7 +85,7 @@ def __init__(
         install_signal_handlers()
 
         # Setup context and register a built-in event handler for bootstrapping.
-        self.__context = LaunchContext(argv=self.__argv)
+        self.__context = LaunchContext(argv=self.__argv, noninteractive=noninteractive)
         self.__context.register_event_handler(OnIncludeLaunchDescription())
         self.__context.register_event_handler(OnShutdown(on_shutdown=self.__on_shutdown))
 

launch/test/launch/test_execute_process.py

@@ -15,14 +15,20 @@
 """Tests for the ExecuteProcess Action."""
 
 import os
+import platform
+import signal
 import sys
 
 from launch import LaunchDescription
 from launch import LaunchService
+from launch.actions.emit_event import EmitEvent
 from launch.actions.execute_process import ExecuteProcess
 from launch.actions.opaque_function import OpaqueFunction
+from launch.actions.register_event_handler import RegisterEventHandler
 from launch.actions.shutdown_action import Shutdown
 from launch.actions.timer_action import TimerAction
+from launch.event_handlers.on_process_start import OnProcessStart
+from launch.events.shutdown import Shutdown as ShutdownEvent
 
 import pytest
 
@@ -88,6 +94,50 @@ def on_exit_function(context):
     assert on_exit_function.called
 
 
+def test_execute_process_shutdown():
+    """Test shutting down a process in (non)interactive settings."""
+    def on_exit(event, ctx):
+        on_exit.returncode = event.returncode
+
+    def generate_launch_description():
+        process_action = ExecuteProcess(
+            cmd=[sys.executable, '-c', 'import signal; signal.pause()'],
+            sigterm_timeout='1',  # shorten timeouts
+            on_exit=on_exit
+        )
+        # Launch process and emit shutdown event as if
+        # launch had received a SIGINT
+        return LaunchDescription([
+            process_action,
+            RegisterEventHandler(event_handler=OnProcessStart(
+                target_action=process_action,
+                on_start=[
+                    EmitEvent(event=ShutdownEvent(
+                        reason='none',
+                        due_to_sigint=True
+                    ))
+                ]
+            ))
+        ])
+
+    ls = LaunchService(noninteractive=True)
+    ls.include_launch_description(generate_launch_description())
+    assert 0 == ls.run()
+    if platform.system() != 'Windows':
+        assert on_exit.returncode == -signal.SIGINT  # Got SIGINT
+    else:
+        assert on_exit.returncode != 0  # Process terminated
+
+    ls = LaunchService()  # interactive
+    ls.include_launch_description(generate_launch_description())
+    assert 0 == ls.run()
+    if platform.system() != 'Windows':
+        # Assume interactive Ctrl+C (i.e. SIGINT to process group)
+        assert on_exit.returncode == -signal.SIGTERM  # Got SIGTERM
+    else:
+        assert on_exit.returncode != 0  # Process terminated
+
+
 def test_execute_process_with_respawn():
     """Test launching a process with a respawn and respawn_delay attribute."""
     def on_exit_callback(event, context):

launch/test/launch/test_launch_context.py

@@ -38,6 +38,15 @@ def test_launch_context_get_argv():
     assert lc.argv == []
 
 
+def test_launch_context_get_noninteractive():
+    """Test the getting of noninteractive flag in the LaunchContext class."""
+    lc = LaunchContext(noninteractive=True)
+    assert lc.noninteractive
+
+    lc = LaunchContext()
+    assert not lc.noninteractive
+
+
 def test_launch_context_get_set_asyncio_loop():
     """Test the getting and settings for asyncio_loop in the LaunchContext class."""
     lc = LaunchContext()