Let argparse convert the args

We'll pass type=comma_separated_kv_to_dict when defining
user script args, letting it do the parsing for us.

Author: petrutlucian94

coriolisclient/cli/deployments.py

@@ -186,6 +186,7 @@ def get_parser(self, prog_name):
             action='append',
             required=False,
             dest="global_scripts",
+            type=cli_utils.comma_separated_kv_to_dict,
             help='A script that will run for a particular os_type. This '
                  'option can be used multiple times. '
                  'Use: linux=/path/to/script.sh or '
@@ -197,6 +198,7 @@ def get_parser(self, prog_name):
             action='append',
             required=False,
             dest="instance_scripts",
+            type=cli_utils.comma_separated_kv_to_dict,
             help='A script that will run for a particular '
                  'instance specified by the --instance option. '
                  'This option can be used multiple times. '

coriolisclient/cli/transfers.py

@@ -198,13 +198,15 @@ def get_parser(self, prog_name):
         parser.add_argument('--user-script-global', action='append',
                             required=False,
                             dest="global_scripts",
+                            type=cli_utils.comma_separated_kv_to_dict,
                             help='A script that will run for a particular '
                             'os_type. This option can be used multiple '
                             'times. Use: linux=/path/to/script.sh or '
                             'windows=/path/to/script.ps1')
         parser.add_argument('--user-script-instance', action='append',
                             required=False,
                             dest="instance_scripts",
+                            type=cli_utils.comma_separated_kv_to_dict,
                             help='A script that will run for a particular '
                             'instance specified by the --instance option. '
                             'This option can be used multiple times. '
@@ -361,13 +363,15 @@ def get_parser(self, prog_name):
         parser.add_argument('--user-script-global', action='append',
                             required=False,
                             dest="global_scripts",
+                            type=cli_utils.comma_separated_kv_to_dict,
                             help='A script that will run for a particular '
                             'os_type. This option can be used multiple '
                             'times. Use: linux=/path/to/script.sh or '
                             'windows=/path/to/script.ps1')
         parser.add_argument('--user-script-instance', action='append',
                             required=False,
                             dest="instance_scripts",
+                            type=cli_utils.comma_separated_kv_to_dict,
                             help='A script that will run for a particular '
                             'instance specified by the --instance option. '
                             'This option can be used multiple times. '

coriolisclient/cli/utils.py

@@ -202,29 +202,39 @@ def comma_separated_kv_to_dict(input_string: str) -> dict:
     for kv_pair in kv_pairs:
         try:
             key, value = kv_pair.split("=")
-        except ValueError:
-            raise ValueError("Not a <key>=<value> pair: %s" % kv_pair)
+        except ValueError as err:
+            raise ValueError(
+                "Not a <key>=<value> pair: %s. " % kv_pair) from err
         out[key] = value
     return out
 
 
-def compose_user_scripts(global_scripts, instance_scripts):
+def compose_user_scripts(
+    global_scripts: list[dict],
+    instance_scripts: list[dict],
+) -> dict:
+    """Process user script arguments.
+
+    :param global_scripts: a list of dicts describing user scripts, the target
+        OS and the phase in which the scripts should be invoked.
+        The dicts are expected to contain the following keys:
+            * <os>: one of the operating systems supported by Coriolis,
+              e.g. "windows" or "linux". The value will be a local script path.
+            * "phase": optional phase, defaults to "osmorphing_post_os_mount".
+    :param instance_scripts: a list of dicts describing user scripts, each
+        having a corresponding instance.
+        The dicts are expected to contain the following keys:
+            * <instance>: The name of the instance where the script must run.
+            * "phase": optional phase, defaults to "osmorphing_post_os_mount".
+    :returns: the processed list of scripts as expected by the Coriolis API.
+    """
     ret = {
         "global": {},
         "instances": {}
     }
     global_scripts = global_scripts or []
     instance_scripts = instance_scripts or []
-    for global_script_str_params in global_scripts:
-        try:
-            params = comma_separated_kv_to_dict(global_script_str_params)
-        except ValueError:
-            raise ValueError(
-                "Invalid global user script parameter: %s. Expecting "
-                "<os_type>=<script_path>. Can optionally include a comma "
-                "separated phase parameter, "
-                "e.g. <os_type>=<script_path>,phase=<phase>" %
-                global_script_str_params)
+    for params in global_scripts:
         phase = params.pop("phase", constants.PHASE_OSMORPHING_POST_OS_MOUNT)
         if phase not in constants.USER_SCRIPT_PHASES:
             raise ValueError(
@@ -264,17 +274,7 @@ def compose_user_scripts(global_scripts, instance_scripts):
         }
         ret["global"][os_type].append(script_entry)
 
-    for instance_scripts_str_params in instance_scripts:
-        try:
-            params = comma_separated_kv_to_dict(instance_scripts_str_params)
-        except ValueError:
-            raise ValueError(
-                "Invalid instance user script parameter: %s. Expecting "
-                "<instance>=<script_path>. Can optionally include a comma "
-                "separated phase parameter, "
-                "e.g. <instance>=<script_path>,phase=<phase>" %
-                instance_scripts_str_params)
-
+    for params in instance_scripts:
         phase = params.pop("phase", constants.PHASE_OSMORPHING_POST_OS_MOUNT)
         if phase not in constants.USER_SCRIPT_PHASES:
             raise ValueError(

coriolisclient/tests/cli/test_deployments.py

@@ -210,13 +210,16 @@ def test_get_parser(self):
         parser = self.cli.get_parser('coriolis')
         global_script = "linux=/linux/path"
         instance_script = "instance1=/instance1/path"
+
+        processed_global_scripts = [{"linux": "/linux/path"}]
+        processed_instance_scripts = [{"instance1": "/instance1/path"}]
         args = parser.parse_args([
             'transfer_id', '--force', '--dont-clone-disks',
             '--skip-os-morphing', '--user-script-global', global_script,
             '--user-script-instance', instance_script])
         self.assertEqual(
-            ('transfer_id', True, False, True, [global_script],
-             [instance_script]),
+            ('transfer_id', True, False, True,
+              processed_global_scripts, processed_instance_scripts),
             (args.transfer, args.force, args.clone_disks,
              args.skip_os_morphing, args.global_scripts,
              args.instance_scripts))

coriolisclient/tests/cli/test_utils.py

@@ -234,6 +234,17 @@ def test_get_option_value_from_args_json_value_error(self):
             "option-name"
         )
 
+    @ddt.data(
+        "linux script",
+        "linux=script,phase",
+    )
+    def test_comma_separated_kv_to_dict_invalid(self, params):
+        self.assertRaises(
+            ValueError,
+            utils.comma_separated_kv_to_dict,
+            params,
+        )
+
     @ddt.data(
         {
             "global_scripts": None,
@@ -253,11 +264,6 @@ def test_get_option_value_from_args_json_value_error(self):
                 "instances": {"instance_1": None}
             }
         },
-        {
-            "global_scripts": ["linux script"],
-            "instance_scripts": ["linux script"],
-            "expected_result": None
-        },
         {
             "global_scripts": ["invalid_os=scrips"],
             "instance_scripts": None,
@@ -289,8 +295,14 @@ def test_get_option_value_from_args_json_value_error(self):
         }
     )
     def test_compose_user_scripts(self, data):
-        global_scripts = data["global_scripts"]
-        instance_scripts = data["instance_scripts"]
+        global_scripts = [
+            utils.comma_separated_kv_to_dict(params)
+            for params in data["global_scripts"]
+        ] if data["global_scripts"] else None
+        instance_scripts = [
+            utils.comma_separated_kv_to_dict(params)
+            for params in data["instance_scripts"]
+        ] if data["instance_scripts"] else None
         expected_result = data["expected_result"]
 
         if expected_result:
@@ -323,6 +335,17 @@ def test_compose_user_scripts_from_file(self):
             f"instance1={script_path},phase=osmorphing_pre_os_mount",  # noqa
         ]
 
+        # We could've provided the dicts directly but we'll exercise
+        # comma_separated_kv_to_dict instead.
+        global_scripts = [
+            utils.comma_separated_kv_to_dict(params)
+            for params in global_scripts
+        ]
+        instance_scripts = [
+            utils.comma_separated_kv_to_dict(params)
+            for params in instance_scripts
+        ]
+
         result = utils.compose_user_scripts(global_scripts, instance_scripts)
 
         payload = '"mock_script1"\n"mock_script2"\n'