From c413a4ec1e4343c9eec9dfb5032a4d383664fcfc Mon Sep 17 00:00:00 2001 From: "matthew.lanting" Date: Thu, 20 Jan 2022 15:18:36 -0500 Subject: [PATCH 1/2] Added case for instances of ExecuteLocal Distro A, OPSEC #4584. You may have additional rights; please see https://rosmilitary.org/faq/?category=ros-2-license Signed-off-by: matthew.lanting --- launch_testing/launch_testing/util/proc_lookup.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/launch_testing/launch_testing/util/proc_lookup.py b/launch_testing/launch_testing/util/proc_lookup.py index 64ac87108..54377dac3 100644 --- a/launch_testing/launch_testing/util/proc_lookup.py +++ b/launch_testing/launch_testing/util/proc_lookup.py @@ -104,7 +104,8 @@ def resolveProcesses(info_obj, *, process=None, cmd_args=None, strict_proc_match raise NoMatchingProcessException('No data recorded for any process') return all_procs - if isinstance(process, launch.actions.ExecuteProcess): + if (isinstance(process, launch.actions.ExecuteProcess) or + isinstance(process, launch.actions.ExecuteLocal)): # We want to search a specific process if process in info_obj.processes(): return [process] From c9ad1f85b940a3e3247d5b9fe9b250d32d310dfc Mon Sep 17 00:00:00 2001 From: "matthew.lanting" Date: Thu, 17 Feb 2022 14:04:32 -0500 Subject: [PATCH 2/2] Added unit tests for ExecuteLocal in test_resolve_process Added case for handling ExecuteLocal actions in _proc_to_name_and_args in proc_lookup.py. Distro A, OPSEC #4584. You may have additional rights; please see https://rosmilitary.org/faq/?category=ros-2-license Signed-off-by: matthew.lanting --- .../launch_testing/util/proc_lookup.py | 6 ++- .../launch_testing/test_resolve_process.py | 47 +++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/launch_testing/launch_testing/util/proc_lookup.py b/launch_testing/launch_testing/util/proc_lookup.py index 54377dac3..ea9660d38 100644 --- a/launch_testing/launch_testing/util/proc_lookup.py +++ b/launch_testing/launch_testing/util/proc_lookup.py @@ -51,7 +51,11 @@ def _proc_to_name_and_args(proc): def perform_subs(subs): return ''.join([sub.perform(_fake_context()) for sub in subs]) try: - cmd = [perform_subs(sub) for sub in proc.cmd] + cmd = [] + if (isinstance(proc, launch.actions.ExecuteProcess)): + cmd = [perform_subs(sub) for sub in proc.cmd] + elif (isinstance(proc, launch.actions.ExecuteLocal)): + cmd = [perform_subs(sub) for sub in proc.process_description.cmd] return ' '.join(cmd) except _FakeContextException: return 'Unknown - Process not launched yet' diff --git a/launch_testing/test/launch_testing/test_resolve_process.py b/launch_testing/test/launch_testing/test_resolve_process.py index 84a08d973..3ebfc3b4d 100644 --- a/launch_testing/test/launch_testing/test_resolve_process.py +++ b/launch_testing/test/launch_testing/test_resolve_process.py @@ -32,6 +32,29 @@ class TestResolveProcess(unittest.TestCase): def test_unlaunched_process_lookup(self): info_obj = launch_testing.ProcInfoHandler() + lookup_obj = launch.actions.ExecuteLocal( + process_description=launch.descriptions.Executable( + cmd=[ + 'python', + '-c', + '', + ] + ) + ) + + with self.assertRaises(launch_testing.util.NoMatchingProcessException) as cm: + launch_testing.util.resolveProcesses( + info_obj, + process=lookup_obj + ) + + # We'll get a good error mesasge here because there were no substitutions in + # the execute process cmd - it's all text + self.assertIn('python -c', str(cm.exception)) + + def test_backward_compatible_unlaunched_process_lookup(self): + info_obj = launch_testing.ProcInfoHandler() + lookup_obj = launch.actions.ExecuteProcess( cmd=[ 'python', @@ -53,6 +76,30 @@ def test_unlaunched_process_lookup(self): def test_unlaunched_process_lookup_with_substitutions(self): info_obj = launch_testing.ProcInfoHandler() + lookup_obj = launch.actions.ExecuteLocal( + process_description=launch.descriptions.Executable( + cmd=[ + launch.substitutions.LocalSubstitution('foo'), + 'python', + '-c', + '', + ] + ) + ) + + with self.assertRaises(launch_testing.util.NoMatchingProcessException) as cm: + launch_testing.util.resolveProcesses( + info_obj, + process=lookup_obj + ) + + # Since the process wasn't launched yet, and it has substitutions that need to be + # resolved by the launch system, we won't be able to take a guess at the command + self.assertIn('Unknown', str(cm.exception)) + + def test_backward_compatible_unlaunched_process_lookup_with_substitutions(self): + info_obj = launch_testing.ProcInfoHandler() + lookup_obj = launch.actions.ExecuteProcess( cmd=[ launch.substitutions.LocalSubstitution('foo'),