Add MultiKeyValueAction to custom parser action

Class MultiKeyValueAction will be used to parse arguments like this:

--route destination=xxx,gateway=xxx --route destination=yyy,gateway=yyy

The result is a list like this:

[{destination:xxx, gateway:xxx}, {destination:yyy, gateway:yyy}]

This action also contain validation of the parameters.

Change-Id: Ie3aa8635c6a13fc2e429fe6922acd681dc7244cf

Author: Tang Chen

openstackclient/common/parseractions.py

@@ -17,6 +17,8 @@
 
 import argparse
 
+from openstackclient.i18n import _
+
 
 class KeyValueAction(argparse.Action):
     """A custom action to parse arguments as key=value pairs
@@ -36,6 +38,85 @@ def __call__(self, parser, namespace, values, option_string=None):
             getattr(namespace, self.dest, {}).pop(values, None)
 
 
+class MultiKeyValueAction(argparse.Action):
+    """A custom action to parse arguments as key1=value1,key2=value2 pairs
+
+    Ensure that ``dest`` is a list. The list will finally contain multiple
+    dicts, with key=value pairs in them.
+
+    NOTE: The arguments string should be a comma separated key-value pairs.
+    And comma(',') and equal('=') may not be used in the key or value.
+    """
+
+    def __init__(self, option_strings, dest, nargs=None,
+                 required_keys=None, optional_keys=None, **kwargs):
+        """Initialize the action object, and parse customized options
+
+        Required keys and optional keys can be specified when initializing
+        the action to enable the key validation. If none of them specified,
+        the key validation will be skipped.
+
+        :param required_keys: a list of required keys
+        :param optional_keys: a list of optional keys
+        """
+        if nargs:
+            raise ValueError("Parameter 'nargs' is not allowed, but got %s"
+                             % nargs)
+
+        super(MultiKeyValueAction, self).__init__(option_strings,
+                                                  dest, **kwargs)
+
+        # required_keys: A list of keys that is required. None by default.
+        if required_keys and not isinstance(required_keys, list):
+            raise TypeError("'required_keys' must be a list")
+        self.required_keys = set(required_keys or [])
+
+        # optional_keys: A list of keys that is optional. None by default.
+        if optional_keys and not isinstance(optional_keys, list):
+            raise TypeError("'optional_keys' must be a list")
+        self.optional_keys = set(optional_keys or [])
+
+    def __call__(self, parser, namespace, values, metavar=None):
+        # Make sure we have an empty list rather than None
+        if getattr(namespace, self.dest, None) is None:
+            setattr(namespace, self.dest, [])
+
+        params = {}
+        for kv in values.split(','):
+            # Add value if an assignment else raise ArgumentTypeError
+            if '=' in kv:
+                params.update([kv.split('=', 1)])
+            else:
+                msg = ("Expected key=value pairs separated by comma, "
+                       "but got: %s" % (str(kv)))
+                raise argparse.ArgumentTypeError(self, msg)
+
+        # Check key validation
+        valid_keys = self.required_keys | self.optional_keys
+        if valid_keys:
+            invalid_keys = [k for k in params if k not in valid_keys]
+            if invalid_keys:
+                msg = _("Invalid keys %(invalid_keys)s specified.\n"
+                        "Valid keys are: %(valid_keys)s.")
+                raise argparse.ArgumentTypeError(
+                    msg % {'invalid_keys': ', '.join(invalid_keys),
+                           'valid_keys': ', '.join(valid_keys)}
+                )
+
+        if self.required_keys:
+            missing_keys = [k for k in self.required_keys if k not in params]
+            if missing_keys:
+                msg = _("Missing required keys %(missing_keys)s.\n"
+                        "Required keys are: %(required_keys)s.")
+                raise argparse.ArgumentTypeError(
+                    msg % {'missing_keys': ', '.join(missing_keys),
+                           'required_keys': ', '.join(self.required_keys)}
+                )
+
+        # Update the dest dict
+        getattr(namespace, self.dest, []).append(params)
+
+
 class RangeAction(argparse.Action):
     """A custom action to parse a single value or a range of values
 

openstackclient/tests/common/test_parseractions.py

@@ -61,6 +61,135 @@ def test_error_values(self):
         self.assertDictEqual(expect, actual)
 
 
+class TestMultiKeyValueAction(utils.TestCase):
+
+    def setUp(self):
+        super(TestMultiKeyValueAction, self).setUp()
+
+        self.parser = argparse.ArgumentParser()
+
+        # Set up our typical usage
+        self.parser.add_argument(
+            '--test',
+            metavar='req1=xxx,req2=yyy',
+            action=parseractions.MultiKeyValueAction,
+            dest='test',
+            default=None,
+            required_keys=['req1', 'req2'],
+            optional_keys=['opt1', 'opt2'],
+            help='Test'
+        )
+
+    def test_good_values(self):
+        results = self.parser.parse_args([
+            '--test', 'req1=aaa,req2=bbb',
+            '--test', 'req1=,req2=',
+        ])
+
+        actual = getattr(results, 'test', [])
+        expect = [
+            {'req1': 'aaa', 'req2': 'bbb'},
+            {'req1': '', 'req2': ''},
+        ]
+        # Need to sort the lists before comparing them
+        key = lambda x: x['req1']
+        expect.sort(key=key)
+        actual.sort(key=key)
+        self.assertListEqual(expect, actual)
+
+    def test_empty_required_optional(self):
+        self.parser.add_argument(
+            '--test-empty',
+            metavar='req1=xxx,req2=yyy',
+            action=parseractions.MultiKeyValueAction,
+            dest='test_empty',
+            default=None,
+            required_keys=[],
+            optional_keys=[],
+            help='Test'
+        )
+
+        results = self.parser.parse_args([
+            '--test-empty', 'req1=aaa,req2=bbb',
+            '--test-empty', 'req1=,req2=',
+        ])
+
+        actual = getattr(results, 'test_empty', [])
+        expect = [
+            {'req1': 'aaa', 'req2': 'bbb'},
+            {'req1': '', 'req2': ''},
+        ]
+        # Need to sort the lists before comparing them
+        key = lambda x: x['req1']
+        expect.sort(key=key)
+        actual.sort(key=key)
+        self.assertListEqual(expect, actual)
+
+    def test_error_values_with_comma(self):
+        self.assertRaises(
+            argparse.ArgumentTypeError,
+            self.parser.parse_args,
+            [
+                '--test', 'mmm,nnn=zzz',
+            ]
+        )
+
+    def test_error_values_without_comma(self):
+        self.assertRaises(
+            argparse.ArgumentTypeError,
+            self.parser.parse_args,
+            [
+                '--test', 'mmmnnn',
+            ]
+        )
+
+    def test_missing_key(self):
+        self.assertRaises(
+            argparse.ArgumentTypeError,
+            self.parser.parse_args,
+            [
+                '--test', 'req2=ddd',
+            ]
+        )
+
+    def test_invalid_key(self):
+        self.assertRaises(
+            argparse.ArgumentTypeError,
+            self.parser.parse_args,
+            [
+                '--test', 'req1=aaa,req2=bbb,aaa=req1',
+            ]
+        )
+
+    def test_required_keys_not_list(self):
+        self.assertRaises(
+            TypeError,
+            self.parser.add_argument,
+            '--test-required-dict',
+            metavar='req1=xxx,req2=yyy',
+            action=parseractions.MultiKeyValueAction,
+            dest='test_required_dict',
+            default=None,
+            required_keys={'aaa': 'bbb'},
+            optional_keys=['opt1', 'opt2'],
+            help='Test'
+        )
+
+    def test_optional_keys_not_list(self):
+        self.assertRaises(
+            TypeError,
+            self.parser.add_argument,
+            '--test-optional-dict',
+            metavar='req1=xxx,req2=yyy',
+            action=parseractions.MultiKeyValueAction,
+            dest='test_optional_dict',
+            default=None,
+            required_keys=['req1', 'req2'],
+            optional_keys={'aaa': 'bbb'},
+            help='Test'
+        )
+
+
 class TestNonNegativeAction(utils.TestCase):
 
     def setUp(self):