[Backport] [Foxy] Support non-interactive launch.LaunchService runs (#475)

launch/launch/actions/execute_process.py

@@ -562,9 +562,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()