Support passing keyword arguments to rosidl CLI extensions

rosidl_cli/rosidl_cli/extensions.py

@@ -13,9 +13,12 @@
 # limitations under the License.
 
 import logging
+import re
 
 from rosidl_cli.entry_points import load_entry_points
 
+import yaml
+
 
 logger = logging.getLogger(__name__)
 
@@ -31,26 +34,65 @@ def name(self):
         return self.__name
 
 
-def load_extensions(group_name, *, strict=False, **kwargs):
+SPECS_PATTERN = re.compile(r'^(\w+)(?:\[(.+)\])?$')
+
+
+def parse_extension_specification(spec):
     """
-    Load extensions for a specific group.
+    Parse extension specification.
 
-    See :py:function:`load_entry_points` for further reference on
-    additional keyword arguments.
+    :param str spec: specification string in
+      'name[key0: value0, ...]' or 'name' format.
+      Key-value pairs are parsed as YAML dictionaries.
+    :returns: a tuple of specification name and
+      keyword arguments, if any, as a dict.
+    """
+    match = SPECS_PATTERN.match(spec)
+    if not match:
+        raise ValueError(f'{spec} is not a valid spec')
+    name = match.group(1)
+    kwargs = match.group(2)
+    if kwargs is not None:
+        try:
+            kwargs = yaml.safe_load('{' + kwargs + '}')
+        except Exception as e:
+            raise ValueError(
+                f'{spec} is not a valid spec'
+            ) from e
+    else:
+        kwargs = {}
+    return name, kwargs
+
+
+def load_extensions(group_name, *, specs=None, strict=False):
+    """
+    Load extensions for a specific group.
 
     :param str group_name: the name of the extension group
+    :param list specs: an optional collection of extension specs
+      (see :py:function:`parse_extension_specification` for spec format)
     :param bool strict: whether to raise or warn on error
     :returns: a list of :py:class:`Extension` instances
     :rtype: list
     """
     extensions = []
+
+    if specs is not None:
+        kwargs = dict(map(
+            parse_extension_specification, specs))
+        specs = list(kwargs.keys())
+    else:
+        kwargs = {}
+
     for name, factory in load_entry_points(
-        group_name, strict=strict, **kwargs
+        group_name, specs=specs, strict=strict
     ).items():
         try:
-            extensions.append(factory(name))
+            extensions.append(factory(name, **kwargs.get(name, {})))
         except Exception as e:
-            msg = f"Failed to instantiate extension '{name}': {e}"
+            msg = f"Failed to instantiate extension '{name}' "
+            which = kwargs.get(name, 'default')
+            msg += f"with '{which}' arguments: {e}"
             if strict:
                 raise RuntimeError(msg)
             logger.warning(msg)

rosidl_cli/test/rosidl_cli/test_extensions.py

@@ -0,0 +1,36 @@
+# Copyright 2021 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 pytest
+
+from rosidl_cli.extensions import parse_extension_specification
+
+
+def test_extension_specification_parsing():
+    with pytest.raises(ValueError):
+        parse_extension_specification('bad[')
+
+    with pytest.raises(ValueError):
+        parse_extension_specification('bad[]')
+
+    with pytest.raises(ValueError):
+        parse_extension_specification('bad[:]')
+
+    name, kwargs = parse_extension_specification('no_args')
+    assert name == 'no_args'
+    assert kwargs == {}
+
+    name, kwargs = parse_extension_specification('with_args[key: value]')
+    assert name == 'with_args'
+    assert kwargs == {'key': 'value'}