Support LaunchService injection into pre-shutdown tests.

launch/launch/actions/execute_process.py

@@ -332,6 +332,10 @@ def __on_shutdown_process_event(
         self,
         context: LaunchContext
     ) -> Optional[LaunchDescription]:
+        typed_event = cast(ShutdownProcess, context.locals.event)
+        if not typed_event.process_matcher(self):
+            # this event whas not intended for this process
+            return None
         return self._shutdown_process(context, send_sigint=True)
 
     def __on_signal_process_event(

launch_testing/example_processes/terminating_proc

@@ -41,4 +41,4 @@ if __name__ == "__main__":
 
     print("Shutting Down")
 
-    sys.exit(0) 
+    sys.exit(0)

launch_testing/examples/README.md

@@ -18,6 +18,15 @@ After shutdown, we run a similar test that checks more output, and also checks t
 order of the output.  `test_out_of_order` demonstrates that the `assertSequentialStdout`
 context manager is able to detect out of order stdout.
 
+## `terminating_proc.test.py`
+
+Usage:
+```sh
+launch_test examples/terminating_proc.test.py
+```
+
+This test checks proper functionality of the _terminating\_proc_ example (source found in the [example\_processes\ folder](../example\_processes)).
+
 ## `args.test.py`
 
 Usage to view the arguments:

launch_testing/examples/example_test_context.test.py

@@ -25,26 +25,25 @@
 from launch_testing.asserts import assertSequentialStdout
 
 
-TEST_PROC_PATH = os.path.join(
-    ament_index_python.get_package_prefix('launch_testing'),
-    'lib/launch_testing',
-    'good_proc'
-)
+def get_test_process_action():
+    TEST_PROC_PATH = os.path.join(
+        ament_index_python.get_package_prefix('launch_testing'),
+        'lib/launch_testing',
+        'good_proc'
+    )
+    return launch.actions.ExecuteProcess(
+        cmd=[sys.executable, TEST_PROC_PATH],
+        name='good_proc',
+        # This is necessary to get unbuffered output from the process under test
+        additional_env={'PYTHONUNBUFFERED': '1'},
+    )
 
 
 # This launch description shows the prefered way to let the tests access launch actions.  By
 # adding them to the test context, it's not necessary to scope them at the module level like in
 # the good_proc.test.py example
 def generate_test_description(ready_fn):
-    # This is necessary to get unbuffered output from the process under test
-    proc_env = os.environ.copy()
-    proc_env['PYTHONUNBUFFERED'] = '1'
-
-    dut_process = launch.actions.ExecuteProcess(
-        cmd=[sys.executable, TEST_PROC_PATH],
-        name='good_proc',
-        env=proc_env,
-    )
+    dut_process = get_test_process_action()
 
     ld = launch.LaunchDescription([
         dut_process,

launch_testing/examples/terminating_proc.test.py

@@ -0,0 +1,91 @@
+# Copyright 2019 Open Source Robotics Foundation, Inc.
+#
+# 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.
+
+import os
+import sys
+import unittest
+
+import ament_index_python
+
+import launch
+import launch.actions
+
+import launch_testing
+import launch_testing.asserts
+import launch_testing.tools
+
+
+def get_test_process_action(*, args=[]):
+    test_proc_path = os.path.join(
+        ament_index_python.get_package_prefix('launch_testing'),
+        'lib/launch_testing',
+        'terminating_proc'
+    )
+    return launch.actions.ExecuteProcess(
+        cmd=[sys.executable, test_proc_path, *args],
+        name='terminating_proc',
+        # This is necessary to get unbuffered output from the process under test
+        additional_env={'PYTHONUNBUFFERED': '1'},
+    )
+
+
+def generate_test_description(ready_fn):
+    return launch.LaunchDescription([
+        launch_testing.util.KeepAliveProc(),
+        launch.actions.OpaqueFunction(function=lambda context: ready_fn()),
+    ])
+
+
+class TestTerminatingProc(unittest.TestCase):
+
+    def test_no_args(self, launch_service, proc_output, proc_info):
+        """Test terminating_proc without command line arguments."""
+        proc_action = get_test_process_action()
+        with launch_testing.tools.launch_process(launch_service, proc_action) as dut:
+            proc_info.assertWaitForStartup(process=dut, timeout=2)
+            proc_output.assertWaitFor('Starting Up', process=dut, timeout=2)
+            proc_output.assertWaitFor('Emulating Work', process=dut, timeout=2)
+            proc_output.assertWaitFor('Done', process=dut, timeout=2)
+            proc_output.assertWaitFor('Shutting Down', process=dut, timeout=2)
+            proc_info.assertWaitForShutdown(process=dut, timeout=2)
+        launch_testing.asserts.assertExitCodes(proc_info, process=dut)
+
+    def test_with_args(self, launch_service, proc_output, proc_info):
+        """Test terminating_proc with some command line arguments."""
+        proc_action = get_test_process_action(args=['--foo', 'bar'])
+        with launch_testing.tools.launch_process(launch_service, proc_action) as dut:
+            proc_info.assertWaitForStartup(process=dut, timeout=2)
+            proc_output.assertWaitFor('Starting Up', process=dut, timeout=2)
+            proc_output.assertWaitFor(
+                "Called with arguments ['--foo', 'bar']", process=dut, timeout=2
+            )
+            proc_output.assertWaitFor('Emulating Work', process=dut, timeout=2)
+            proc_output.assertWaitFor('Done', process=dut, timeout=2)
+            proc_output.assertWaitFor('Shutting Down', process=dut, timeout=2)
+            proc_info.assertWaitForShutdown(process=dut, timeout=2)
+        launch_testing.asserts.assertExitCodes(proc_info, process=dut)
+
+    def test_unhandled_exception(self, launch_service, proc_output, proc_info):
+        """Test terminating_proc forcing an unhandled exception."""
+        proc_action = get_test_process_action(args=['--exception'])
+        with launch_testing.tools.launch_process(launch_service, proc_action) as dut:
+            proc_info.assertWaitForStartup(process=dut, timeout=2)
+            proc_output.assertWaitFor('Starting Up', process=dut, timeout=2)
+            proc_output.assertWaitFor(
+                "Called with arguments ['--exception']", process=dut, timeout=2
+            )
+            proc_info.assertWaitForShutdown(process=dut, timeout=2)
+        launch_testing.asserts.assertExitCodes(
+            proc_info, process=dut, allowable_exit_codes=[1]
+        )

launch_testing/launch_testing/test_runner.py

@@ -87,6 +87,7 @@ def run(self):
         self._test_run.bind(
             self._test_run.pre_shutdown_tests,
             injected_attributes={
+                'launch_service': self._launch_service,
                 'proc_info': proc_info,
                 'proc_output': proc_output,
                 'test_args': test_args,
@@ -95,6 +96,7 @@ def run(self):
                 full_context,
                 # Add a few more things to the args dictionary:
                 **{
+                    'launch_service': self._launch_service,
                     'proc_info': proc_info,
                     'proc_output': proc_output,
                     'test_args': test_args

launch_testing/launch_testing/tools/__init__.py

@@ -12,10 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from .launchers import launch_process
 from .output import basic_output_filter
 from .output import expected_output_from_file
 
+
 __all__ = [
     'basic_output_filter',
     'expected_output_from_file',
+    'launch_process',
 ]

launch_testing/launch_testing/tools/launchers.py

@@ -0,0 +1,43 @@
+# Copyright 2019 Open Source Robotics Foundation, Inc.
+#
+# 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.
+
+import contextlib
+
+import launch
+import launch.events
+
+
+@contextlib.contextmanager
+def launch_process(launch_service, process_action):
+    """
+    Launch a process.
+
+    Start execution of a ``process_action`` using the given
+    ``launch_service`` upon context entering and shut it down
+    upon context exiting.
+    """
+    assert isinstance(process_action, launch.actions.ExecuteProcess)
+    launch_service.emit_event(
+        event=launch.events.IncludeLaunchDescription(
+            launch_description=launch.LaunchDescription([process_action])
+        )
+    )
+    try:
+        yield process_action
+    finally:
+        launch_service.emit_event(
+            event=launch.events.process.ShutdownProcess(
+                process_matcher=launch.events.matches_action(process_action)
+            )
+        )