Add support for eval args in roslaunch XML reader (fixes #267) (#372)

* added _resolve_eval method

* added separate _resolve methods

* updated method names

* implemented _resolve_eval

* fixed bad syntax

* fixed various issues detected by mypy and flake8

* updated _resolve_find to take optional second argument

* added husky config for testing

* fixed incorrect image name

* bug fix: missing import

* bug fix: use empty string instead

* added regression test

* updated CHANGELOG

Author: ChrisTimperley

CHANGELOG.md

@@ -18,6 +18,7 @@
 * Updated handling of `command` attributes in `param` tags during parsing
   of XML launch files: Only the output of `stdout` is recorded, and `stderr`
   is now ignored.
+* Added handling of `$(eval ...)` tags in XML launch files.
 
 
 # 1.3.0 (2020-06-02)

src/roswire/proxy/roslaunch/substitution.py

@@ -31,15 +31,13 @@ class ArgumentResolver:
     files: dockerblade.FileSystem
     context: Dict[str, Any] = attr.ib(default=None)
 
-    def _resolve_arg(self, s: str) -> str:
+    def _resolve_substitution_arg(self, s: str) -> str:
         """
         Raises
         ------
         EnvNotFoundError
             if a given environment variable is not found.
         """
-        shell = self.shell
-        context = self.context
         logger.debug(f"resolving substitution argument: {s}")
         s = s[2:-1]
         logger.debug(f"stripped delimiters: {s}")
@@ -49,31 +47,49 @@ def _resolve_arg(self, s: str) -> str:
         # we deal with find in a later stage
         if kind == 'find':
             return f'$({s})'
-        if kind == 'env':
-            return shell.environ(params[0])
-        if kind == 'optenv':
-            try:
-                return shell.environ(params[0])
-            except dockerblade.exceptions.EnvNotFoundError:
-                return ' '.join(params[1:])
-        if kind == 'dirname':
-            try:
-                dirname = os.path.dirname(context['filename'])
-            except KeyError:
-                m = 'filename is not provided by the launch context'
-                raise SubstitutionError(m)
-            dirname = os.path.normpath(dirname)
-            return dirname
-        if kind == 'arg':
+        elif kind == 'env':
+            var = params[0]
+            return self._resolve_env(var)
+        elif kind == 'optenv':
+            var = params[0]
+            default = ' '.join(params[1:])
+            return self._resolve_optenv(var, default)
+        elif kind == 'dirname':
+            return self._resolve_dirname()
+        elif kind == 'arg':
             arg_name = params[0]
-            if 'arg' not in context or arg_name not in context['arg']:
-                m = f'arg not supplied to launch context [{arg_name}]'
-                raise SubstitutionError(m)
-            return context['arg'][arg_name]
-
-        # TODO $(anon name)
+            return self._resolve_arg(arg_name)
+        elif kind == 'anon':
+            return self._resolve_anon(params[0])
         return s
 
+    def _resolve_dirname(self) -> str:
+        try:
+            dirname = os.path.dirname(self.context['filename'])
+        except KeyError:
+            m = 'filename is not provided by the launch context'
+            raise SubstitutionError(m)
+        return os.path.normpath(dirname)
+
+    def _resolve_anon(self, name: str) -> str:
+        raise NotImplementedError
+
+    def _resolve_env(self, var: str) -> str:
+        return self.shell.environ(var)
+
+    def _resolve_optenv(self, var: str, default: str) -> str:
+        try:
+            return self.shell.environ(var)
+        except dockerblade.exceptions.EnvNotFoundError:
+            return default
+
+    def _resolve_arg(self, arg_name: str) -> str:
+        context = self.context
+        if 'arg' not in context or arg_name not in context['arg']:
+            m = f'arg not supplied to launch context [{arg_name}]'
+            raise SubstitutionError(m)
+        return context['arg'][arg_name]
+
     def _find_package_path(self, package: str) -> str:
         cmd = f'rospack find {shlex.quote(package)}'
         try:
@@ -125,7 +141,7 @@ def _find_resource(self, package: str, path: str) -> str:
             raise SubstitutionError(m)
         return path_in_package
 
-    def _resolve_find(self, package: str, path: str) -> str:
+    def _resolve_find(self, package: str, path: str = '') -> str:
         logger.debug(f'resolving find: {package}')
         path_original = path
 
@@ -153,12 +169,42 @@ def _resolve_find(self, package: str, path: str) -> str:
         resolved_path = self._find_package_path(package) + path_original
         return resolved_path
 
+    def _resolve_eval(self, attribute_string: str) -> str:
+        logger.debug(f'resolving eval: {attribute_string}')
+        assert attribute_string.startswith('$(eval ')
+        assert attribute_string[-1] == ')'
+        eval_string = attribute_string[7:-1]
+
+        if '__' in attribute_string:
+            m = ("$(eval ...): refusing to evaluate potentially dangerous "
+                 "expression -- must not contain double underscores")
+            raise SubstitutionError(m)
+
+        _builtins = {x: __builtins__[x]  # type: ignore
+                     for x in ('dict', 'float', 'int', 'list', 'map')}
+        _locals = {
+            'true': True,
+            'True': True,
+            'false': False,
+            'False': False,
+            '__builtins__': _builtins,
+            'arg': self._resolve_arg,
+            'anon': self._resolve_anon,
+            'dirname': self._resolve_dirname,
+            'env': self._resolve_env,
+            'find': self._resolve_find,
+            'optenv': self._resolve_optenv
+        }
+
+        result = str(eval(eval_string, {}, _locals))
+        logger.debug(f'resolved eval [{attribute_string}]: {result}')
+        return result
+
     def resolve(self, s: str) -> str:
         """Resolves a given argument string."""
-        # TODO $(eval ...)
         if s.startswith('$(eval ') and s[-1] == ')':
-            raise NotImplementedError
-        s = R_ARG.sub(lambda m: self._resolve_arg(m.group(0)), s)
+            return self._resolve_eval(s)
+        s = R_ARG.sub(lambda m: self._resolve_substitution_arg(m.group(0)), s)
 
         def process_find_arg(match: Match[str]) -> str:
             # split tag and optional trailing path

test/systems/husky.yml

@@ -0,0 +1,4 @@
+image: therobotcooperative/husky
+sources:
+- /opt/ros/melodic/setup.bash
+- /ros_ws/devel/setup.bash

test/test_roslaunch.py

@@ -26,6 +26,20 @@ def test_read(sut):
         assert actual_node_names == expected_node_names
 
 
+@pytest.mark.parametrize('sut', ['husky'], indirect=True)
+def test_eval_args_in_launch_file(sut):
+    with sut.roscore() as ros:
+        config = ros.roslaunch.read('spawn_husky.launch', package='husky_gazebo')
+        actual_node_names = {node.name for node in config.nodes}
+        expected_node_names = {'base_controller_spawner',
+                               'ekf_localization',
+                               'robot_state_publisher',
+                               'spawn_husky_model',
+                               'twist_marker_server',
+                               'twist_mux'}
+        assert actual_node_names == expected_node_names
+
+
 @pytest.mark.parametrize('sut', ['fetch'], indirect=True)
 def test_remappings(sut):
     with sut.roscore() as ros: