Adjust for launch_ros modifications, add unit tests

Distro A; OPSEC #4584

Signed-off-by: Roger Strain <[email protected]>

launch/launch/actions/execute_local.py

@@ -90,7 +90,7 @@ def __init__(
             'sigkill_timeout', default=5),
         emulate_tty: bool = False,
         output: Text = 'log',
-        output_format: Text = '[{this.name}] {line}',
+        output_format: Text = '[{this.process_description.name}] {line}',
         log_cmd: bool = False,
         on_exit: Optional[Union[
             SomeActionsType,
@@ -607,7 +607,7 @@ def execute(self, context: LaunchContext) -> Optional[List[LaunchDescriptionEnti
 
         if self.__executed:
             raise RuntimeError(
-                f"ExecuteProcess action '{name}': executed more than once: {self.describe()}"
+                f"ExecuteLocal action '{name}': executed more than once: {self.describe()}"
             )
         self.__executed = True
 

launch/launch/actions/execute_process.py

@@ -144,9 +144,9 @@ def __init__(
             Defaults to 'False'.
         :param: respawn_delay a delay time to relaunch the died process if respawn is 'True'.
         """
-        self.__executable = Executable(cmd=cmd, prefix=prefix, name=name, cwd=cwd, env=env,
-                                       additional_env=additional_env)
-        super().__init__(process_description=self.__executable, **kwargs)
+        executable = Executable(cmd=cmd, prefix=prefix, name=name, cwd=cwd, env=env,
+                                additional_env=additional_env)
+        super().__init__(process_description=executable, **kwargs)
 
     @classmethod
     def _parse_cmdline(
@@ -287,32 +287,32 @@ def parse(
     @property
     def name(self):
         """Getter for name."""
-        if self.__executable.final_name is not None:
-            return self.__executable.final_name
-        return self.__executable.name
+        if self.process_description.final_name is not None:
+            return self.process_description.final_name
+        return self.process_description.name
 
     @property
     def cmd(self):
         """Getter for cmd."""
-        if self.__executable.final_cmd is not None:
-            return self.__executable.final_cmd
-        return self.__executable.cmd
+        if self.process_description.final_cmd is not None:
+            return self.process_description.final_cmd
+        return self.process_description.cmd
 
     @property
     def cwd(self):
         """Getter for cwd."""
-        if self.__executable.final_cwd is not None:
-            return self.__executable.final_cwd
-        return self.__executable.cwd
+        if self.process_description.final_cwd is not None:
+            return self.process_description.final_cwd
+        return self.process_description.cwd
 
     @property
     def env(self):
         """Getter for env."""
-        if self.__executable.final_env is not None:
-            return self.__executable.final_env
-        return self.__executable.env
+        if self.process_description.final_env is not None:
+            return self.process_description.final_env
+        return self.process_description.env
 
     @property
     def additional_env(self):
         """Getter for additional_env."""
-        return self.__executable.additional_env
+        return self.process_description.additional_env

launch/launch/descriptions/executable.py

@@ -31,12 +31,13 @@
 from typing import Optional
 from typing import Tuple
 
-from launch.some_substitutions_type import SomeSubstitutionsType
-from launch.substitution import Substitution
-from launch.substitutions import LaunchConfiguration
-from launch.launch_context import LaunchContext
-from launch.utilities import normalize_to_list_of_substitutions
-from launch.utilities import perform_substitutions
+from ..action import Action
+from ..launch_context import LaunchContext
+from ..some_substitutions_type import SomeSubstitutionsType
+from ..substitution import Substitution
+from ..substitutions import LaunchConfiguration
+from ..utilities import normalize_to_list_of_substitutions
+from ..utilities import perform_substitutions
 
 _executable_process_counter_lock = threading.Lock()
 _executable_process_counter = 0  # in Python3, this number is unbounded (no rollover)
@@ -166,7 +167,7 @@ def __expand_substitutions(self, context):
         with _executable_process_counter_lock:
             global _executable_process_counter
             _executable_process_counter += 1
-            self.__final_name = f"{name}-{_executable_process_counter}"
+            self.__final_name = f'{name}-{_executable_process_counter}'
         cwd = None
         if self.__cwd is not None:
             cwd = ''.join([context.perform_substitution(x) for x in self.__cwd])

launch/launch/event_handlers/on_process_exit.py

@@ -31,7 +31,7 @@
 from ..some_actions_type import SomeActionsType
 
 if TYPE_CHECKING:
-    from ..actions import ExecuteProcess  # noqa: F401
+    from ..actions import ExecuteLocal  # noqa: F401
 
 
 class OnProcessExit(BaseEventHandler):
@@ -45,15 +45,15 @@ class OnProcessExit(BaseEventHandler):
     def __init__(
         self,
         *,
-        target_action: 'ExecuteProcess' = None,
+        target_action: 'ExecuteLocal' = None,
         on_exit: Union[SomeActionsType,
                        Callable[[ProcessExited, LaunchContext], Optional[SomeActionsType]]],
         **kwargs
     ) -> None:
         """Create an OnProcessExit event handler."""
-        from ..actions import ExecuteProcess  # noqa
-        if not isinstance(target_action, (ExecuteProcess, type(None))):
-            raise TypeError("OnProcessExit requires an 'ExecuteProcess' action as the target")
+        from ..actions import ExecuteLocal  # noqa
+        if not isinstance(target_action, (ExecuteLocal, type(None))):
+            raise TypeError("OnProcessExit requires an 'ExecuteLocal' action as the target")
         super().__init__(
             matcher=(
                 lambda event: (
@@ -109,6 +109,6 @@ def matcher_description(self) -> Text:
         """Return the string description of the matcher."""
         if self.__target_action is None:
             return 'event == ProcessExited'
-        return 'event == ProcessExited and event.action == ExecuteProcess({})'.format(
+        return 'event == ProcessExited and event.action == ExecuteLocal({})'.format(
             hex(id(self.__target_action))
         )

launch/launch/event_handlers/on_process_io.py

@@ -38,16 +38,16 @@ class OnProcessIO(BaseEventHandler):
     def __init__(
         self,
         *,
-        target_action: Optional['ExecuteProcess'] = None,
+        target_action: Optional['ExecuteLocal'] = None,
         on_stdin: Callable[[ProcessIO], Optional[SomeActionsType]] = None,
         on_stdout: Callable[[ProcessIO], Optional[SomeActionsType]] = None,
         on_stderr: Callable[[ProcessIO], Optional[SomeActionsType]] = None,
         **kwargs
     ) -> None:
         """Create an OnProcessIO event handler."""
-        from ..actions import ExecuteProcess  # noqa
-        if not isinstance(target_action, (ExecuteProcess, type(None))):
-            raise TypeError("OnProcessIO requires an 'ExecuteProcess' action as the target")
+        from ..actions import ExecuteLocal  # noqa
+        if not isinstance(target_action, (ExecuteLocal, type(None))):
+            raise TypeError("OnProcessIO requires an 'ExecuteLocal' action as the target")
         super().__init__(matcher=self._matcher, **kwargs)
         self.__target_action = target_action
         self.__on_stdin = on_stdin
@@ -95,6 +95,6 @@ def matcher_description(self) -> Text:
         """Return the string description of the matcher."""
         if self.__target_action is None:
             return 'event issubclass of ProcessIO'
-        return 'event issubclass of ProcessIO and event.action == ExecuteProcess({})'.format(
+        return 'event issubclass of ProcessIO and event.action == ExecuteLocal({})'.format(
             hex(id(self.__target_action))
         )

launch/test/launch/test_executable.py

@@ -20,28 +20,59 @@
 #
 # This notice must appear in all copies of this file and its derivatives.
 
+import os
+
 from launch.descriptions.executable import Executable
 from launch.launch_context import LaunchContext
+from launch.substitutions import EnvironmentVariable
 
 
 def test_executable():
-    exe = Executable(cmd="test")
+    exe = Executable(cmd='test')
     assert exe is not None
 
 
 def test_cmd_string_in_list():
     exe = Executable(cmd=['ls "my/subdir/with spaces/"'])
     exe.apply_context(LaunchContext())
-    assert all([a == b for a, b in zip(exe.final_cmd, ['ls "my/subdir/with spaces/"'])])
+    assert all(a == b for a, b in zip(exe.final_cmd, ['ls "my/subdir/with spaces/"']))
 
 
 def test_cmd_strings_in_list():
     exe = Executable(cmd=['ls', '"my/subdir/with spaces/"'])
     exe.apply_context(LaunchContext())
-    assert all([a == b for a, b in zip(exe.final_cmd, ['ls', '"my/subdir/with spaces/"'])])
+    assert all(a == b for a, b in zip(exe.final_cmd, ['ls', '"my/subdir/with spaces/"']))
 
 
 def test_cmd_multiple_arguments_in_string():
     exe = Executable(cmd=['ls', '-opt1', '-opt2', '-opt3'])
     exe.apply_context(LaunchContext())
-    assert all([a == b for a, b in zip(exe.final_cmd, ['ls', '-opt1', '-opt2', '-opt3'])])
+    assert all(a == b for a, b in zip(exe.final_cmd, ['ls', '-opt1', '-opt2', '-opt3']))
+
+def test_passthrough_properties():
+    name = 'name'
+    cwd = 'cwd'
+    env = {'a': '1'}
+    exe = Executable(cmd=['test'], name=name, cwd=cwd, env=env)
+    exe.apply_context(LaunchContext())
+    assert exe.final_name.startswith(name)
+    assert exe.final_cwd == cwd
+    assert exe.final_env == env
+
+def test_substituted_properties():
+    os.environ['EXECUTABLE_TEST_NAME'] = 'name'
+    os.environ['EXECUTABLE_TEST_CWD'] = 'cwd'
+    os.environ['EXECUTABLE_TEST_ENVVAR'] = 'var'
+    os.environ['EXECUTABLE_TEST_ENVVAL'] = 'value'
+    name = EnvironmentVariable('EXECUTABLE_TEST_NAME')
+    cwd = EnvironmentVariable('EXECUTABLE_TEST_CWD')
+    env = {EnvironmentVariable('EXECUTABLE_TEST_ENVVAR'): EnvironmentVariable('EXECUTABLE_TEST_ENVVAL')}
+    exe = Executable(cmd=['test'], name=name, cwd=cwd, env=env)
+    exe.apply_context(LaunchContext())
+    assert exe.final_name.startswith('name')
+    assert exe.final_cwd == 'cwd'
+    assert exe.final_env == {'var': 'value'}
+    del os.environ['EXECUTABLE_TEST_NAME']
+    del os.environ['EXECUTABLE_TEST_CWD']
+    del os.environ['EXECUTABLE_TEST_ENVVAR']
+    del os.environ['EXECUTABLE_TEST_ENVVAL']

launch/test/launch/test_execute_local.py

@@ -0,0 +1,130 @@
+# Copyright 2021 Southwest Research Institute, All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# DISTRIBUTION A. Approved for public release; distribution unlimited.
+# OPSEC #4584.
+#
+# Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS
+# Part 252.227-7013 or 7014 (Feb 2014).
+#
+# This notice must appear in all copies of this file and its derivatives.
+
+"""Tests for the ExecuteLocal Action."""
+
+import os
+import sys
+
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import ExecuteLocal
+from launch.actions import OpaqueFunction
+from launch.actions import Shutdown
+from launch.actions import TimerAction
+from launch.descriptions import Executable
+
+import pytest
+
+@pytest.mark.parametrize('test_input,expected', [
+    (None, [True, False]),
+    ({'TEST_NEW_ENV': '2'}, [False, True])
+])
+def test_execute_process_with_env(test_input, expected):
+    """Test launching a process with an environment variable."""
+    os.environ['TEST_CHANGE_CURRENT_ENV'] = '1'
+    additional_env = {'TEST_PROCESS_WITH_ENV': 'Hello World'}
+    executable = ExecuteLocal(
+        process_description=Executable(
+            cmd=[sys.executable, 'TEST_PROCESS_WITH_ENV'],
+            env=test_input,
+            additional_env=additional_env
+        ),
+        output='screen'
+    )
+    ld = LaunchDescription([executable])
+    ls = LaunchService()
+    ls.include_launch_description(ld)
+    assert 0 == ls.run()
+    env = executable.process_details['env']
+    assert env['TEST_PROCESS_WITH_ENV'] == 'Hello World'
+    assert ('TEST_CHANGE_CURRENT_ENV' in env) is expected[0]
+    if expected[0]:
+        assert env['TEST_CHANGE_CURRENT_ENV'] == '1'
+    assert ('TEST_NEW_ENV' in env) is expected[1]
+    if expected[1]:
+        assert env['TEST_NEW_ENV'] == '2'
+
+
+def test_execute_process_with_on_exit_behavior():
+    """Test a process' on_exit callback and actions are processed."""
+    def on_exit_callback(event, context):
+        on_exit_callback.called = True
+    on_exit_callback.called = False
+
+    executable_with_on_exit_callback = ExecuteLocal(
+        process_description=Executable(cmd=[sys.executable, '-c', "print('callback')"]),
+        output='screen', on_exit=on_exit_callback
+    )
+    assert len(executable_with_on_exit_callback.get_sub_entities()) == 0
+
+    def on_exit_function(context):
+        on_exit_function.called = True
+    on_exit_function.called = False
+    on_exit_action = OpaqueFunction(function=on_exit_function)
+    executable_with_on_exit_action = ExecuteLocal(
+        process_description=Executable(cmd=[sys.executable, '-c', "print('callback')"]),
+        output='screen', on_exit=[on_exit_action]
+    )
+    assert executable_with_on_exit_action.get_sub_entities() == [on_exit_action]
+
+    ld = LaunchDescription([
+        executable_with_on_exit_callback,
+        executable_with_on_exit_action
+    ])
+    ls = LaunchService()
+    ls.include_launch_description(ld)
+    assert 0 == ls.run()
+    assert on_exit_callback.called
+    assert on_exit_function.called
+
+
+def test_execute_process_with_respawn():
+    """Test launching a process with a respawn and respawn_delay attribute."""
+    def on_exit_callback(event, context):
+        on_exit_callback.called_count = on_exit_callback.called_count + 1
+    on_exit_callback.called_count = 0
+
+    respawn_delay = 2.0
+    shutdown_time = 3.0   # to shutdown the launch service, so that the process only respawn once
+    expected_called_count = 2   # normal exit and respawn exit
+
+    def generate_launch_description():
+        return LaunchDescription([
+
+            ExecuteLocal(
+                process_description=Executable(cmd=[sys.executable, '-c', "print('action')"]),
+                respawn=True, respawn_delay=respawn_delay, on_exit=on_exit_callback
+            ),
+
+            TimerAction(
+                period=shutdown_time,
+                actions=[
+                    Shutdown(reason='Timer expired')
+                ]
+            )
+        ])
+
+    ls = LaunchService()
+    ls.include_launch_description(generate_launch_description())
+    assert 0 == ls.run()
+    assert expected_called_count == on_exit_callback.called_count