network: Address review comments on VPNaaS addition

- Fold single-caller methods to their call site
- Replace str2dict_type with existing utility

Change-Id: Icab0efce27d3676a8350b911ab738344e5e60735
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>

Author: stephenfin

openstackclient/network/v2/vpnaas/ikepolicy.py

@@ -18,6 +18,7 @@
 from typing import Any
 
 from osc_lib.cli import identity as identity_utils
+from osc_lib.cli import parseractions
 from osc_lib import exceptions
 from osc_lib import utils
 from osc_lib.utils import columns as column_util
@@ -163,9 +164,14 @@ def _get_common_parser(
     )
     parser.add_argument(
         '--lifetime',
-        metavar="units=UNITS,value=VALUE",
-        type=vpn_utils.str2dict_type(optional_keys=['units', 'value']),
-        help=vpn_utils.lifetime_help("IKE"),
+        metavar='units=<units>,value=<value>',
+        action=parseractions.MultiKeyValueAction,
+        optional_keys=['units', 'value'],
+        help=_(
+            "IKE lifetime attributes. "
+            "'units'-seconds, default:seconds. "
+            "'value'-non negative integer, default:3600."
+        ),
     )
     return parser
 

openstackclient/network/v2/vpnaas/ipsec_site_connection.py

@@ -20,6 +20,7 @@
 from openstack.network import v2 as network_v2
 from osc_lib.cli import format_columns
 from osc_lib.cli import identity as identity_utils
+from osc_lib.cli import parseractions
 from osc_lib import exceptions
 from osc_lib import utils
 from osc_lib.utils import columns as column_util
@@ -98,11 +99,17 @@ def _get_common_parser(
     )
     parser.add_argument(
         '--dpd',
-        metavar="action=ACTION,interval=INTERVAL,timeout=TIMEOUT",
-        type=vpn_utils.str2dict_type(
-            optional_keys=['action', 'interval', 'timeout']
+        metavar="action=<action>,interval=<interval>,timeout=<timeout>",
+        action=parseractions.MultiKeyValueAction,
+        optional_keys=['action', 'interval', 'timeout'],
+        help=_(
+            "IPSec Connection Dead Peer Detection attributes. "
+            "'action'-hold,clear,disabled,restart,restart-by-peer. "
+            "'interval' and 'timeout' are non negative integers.  "
+            "'interval' should be less than 'timeout' value. "
+            "'action', default:hold 'interval', default:30, "
+            "'timeout', default:120."
         ),
-        help=vpn_utils.dpd_help("IPsec connection"),
     )
     parser.add_argument('--mtu', help=_('MTU size for the connection'))
     parser.add_argument(

openstackclient/network/v2/vpnaas/ipsecpolicy.py

@@ -18,6 +18,7 @@
 from typing import Any
 
 from osc_lib.cli import identity as identity_utils
+from osc_lib.cli import parseractions
 from osc_lib import exceptions
 from osc_lib import utils
 from osc_lib.utils import columns as column_util
@@ -145,9 +146,14 @@ def _get_common_parser(parser: argparse.ArgumentParser) -> None:
     )
     parser.add_argument(
         '--lifetime',
-        metavar="units=UNITS,value=VALUE",
-        type=vpn_utils.str2dict_type(optional_keys=['units', 'value']),
-        help=vpn_utils.lifetime_help("IPsec"),
+        metavar='units=<units>,value=<value>',
+        action=parseractions.MultiKeyValueAction,
+        optional_keys=['units', 'value'],
+        help=_(
+            "IPsec lifetime attributes. "
+            "'units'-seconds, default:seconds. "
+            "'value'-non negative integer, default:3600."
+        ),
     )
     parser.add_argument(
         '--pfs',

openstackclient/network/v2/vpnaas/utils.py

@@ -15,9 +15,6 @@
 
 """VPN Utilities and helper functions."""
 
-import argparse
-from collections.abc import Callable
-import functools
 from typing import Any
 
 from osc_lib import exceptions
@@ -100,101 +97,3 @@ def validate_lifetime_dict(lifetime_dict: dict[str, Any]) -> None:
             else:
                 lifetime_dict['value'] = int(value)
     return
-
-
-def lifetime_help(policy: str) -> str:
-    lifetime = (
-        _(
-            "%s lifetime attributes. "
-            "'units'-seconds, default:seconds. "
-            "'value'-non negative integer, default:3600."
-        )
-        % policy
-    )
-    return lifetime
-
-
-def dpd_help(policy: str) -> str:
-    dpd = (
-        _(
-            " %s Dead Peer Detection attributes."
-            " 'action'-hold,clear,disabled,restart,restart-by-peer."
-            " 'interval' and 'timeout' are non negative integers. "
-            " 'interval' should be less than 'timeout' value. "
-            " 'action', default:hold 'interval', default:30, "
-            " 'timeout', default:120."
-        )
-        % policy.capitalize()
-    )
-    return dpd
-
-
-def str2dict(
-    strdict: str,
-    required_keys: list[str] | None = None,
-    optional_keys: list[str] | None = None,
-) -> dict[str, str]:
-    """Convert key1=value1,key2=value2,... string into dictionary.
-
-    :param strdict: string in the form of key1=value1,key2=value2
-    :param required_keys: list of required keys. All keys in this list must be
-                       specified. Otherwise ArgumentTypeError will be raised.
-                       If this parameter is unspecified, no required key check
-                       will be done.
-    :param optional_keys: list of optional keys.
-                       This parameter is used for valid key check.
-                       When at least one of required_keys and optional_keys,
-                       a key must be a member of either of required_keys or
-                       optional_keys. Otherwise, ArgumentTypeError will be
-                       raised. When both required_keys and optional_keys are
-                       unspecified, no valid key check will be done.
-    """
-    result = {}
-    if strdict:
-        i = 0
-        kvlist = []
-        for kv in strdict.split(','):
-            if '=' in kv:
-                kvlist.append(kv)
-                i += 1
-            elif i == 0:
-                msg = _("missing value for key '%s'")
-                raise argparse.ArgumentTypeError(msg % kv)
-            else:
-                kvlist[i - 1] = f"{kvlist[i - 1]},{kv}"
-        for kv in kvlist:
-            key, sep, value = kv.partition('=')
-            if not sep:
-                msg = _("invalid key-value '%s', expected format: key=value")
-                raise argparse.ArgumentTypeError(msg % kv)
-            result[key] = value
-    valid_keys = set(required_keys or []) | set(optional_keys or [])
-    if valid_keys:
-        invalid_keys = [k for k in result if k not in valid_keys]
-        if invalid_keys:
-            msg = _(
-                "Invalid key(s) '%(invalid_keys)s' specified. "
-                "Valid key(s): '%(valid_keys)s'."
-            )
-            raise argparse.ArgumentTypeError(
-                msg
-                % {
-                    'invalid_keys': ', '.join(sorted(invalid_keys)),
-                    'valid_keys': ', '.join(sorted(valid_keys)),
-                }
-            )
-    if required_keys:
-        not_found_keys = [k for k in required_keys if k not in result]
-        if not_found_keys:
-            msg = _("Required key(s) '%s' not specified.")
-            raise argparse.ArgumentTypeError(msg % ', '.join(not_found_keys))
-    return result
-
-
-def str2dict_type(
-    optional_keys: list[str] | None = None,
-    required_keys: list[str] | None = None,
-) -> Callable[[str], dict[str, str]]:
-    return functools.partial(
-        str2dict, optional_keys=optional_keys, required_keys=required_keys
-    )