diff --git a/launch/launch/actions/execute_process.py b/launch/launch/actions/execute_process.py index 7ede77086..c1d23b6e1 100644 --- a/launch/launch/actions/execute_process.py +++ b/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( diff --git a/launch_testing/example_processes/terminating_proc b/launch_testing/example_processes/terminating_proc index 9dc03cdf4..f258ad143 100755 --- a/launch_testing/example_processes/terminating_proc +++ b/launch_testing/example_processes/terminating_proc @@ -41,4 +41,4 @@ if __name__ == "__main__": print("Shutting Down") - sys.exit(0) + sys.exit(0) diff --git a/launch_testing/examples/README.md b/launch_testing/examples/README.md index f2d8875be..7332a1e0e 100644 --- a/launch_testing/examples/README.md +++ b/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: diff --git a/launch_testing/examples/example_test_context.test.py b/launch_testing/examples/example_test_context.test.py index ec840f8c7..df4df3558 100644 --- a/launch_testing/examples/example_test_context.test.py +++ b/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, diff --git a/launch_testing/examples/terminating_proc.test.py b/launch_testing/examples/terminating_proc.test.py new file mode 100644 index 000000000..a6d78331c --- /dev/null +++ b/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] + ) diff --git a/launch_testing/launch_testing/test_runner.py b/launch_testing/launch_testing/test_runner.py index cc68e5425..ccf2311f9 100644 --- a/launch_testing/launch_testing/test_runner.py +++ b/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 diff --git a/launch_testing/launch_testing/tools/__init__.py b/launch_testing/launch_testing/tools/__init__.py index 6bc09d0e7..93d820ab1 100644 --- a/launch_testing/launch_testing/tools/__init__.py +++ b/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', ] diff --git a/launch_testing/launch_testing/tools/launchers.py b/launch_testing/launch_testing/tools/launchers.py new file mode 100644 index 000000000..ec77e7eaf --- /dev/null +++ b/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) + ) + )