Prevent the default value from being used when values were already passed

Author: kinow

cwltool/argparser.py

@@ -817,6 +817,42 @@ class DirectoryAppendAction(FSAppendAction):
     objclass = "Directory"
 
 
+class AppendAction(argparse.Action):
+    """An argparse action that clears the default values if any value is provided.
+
+    Attributes:
+        _called (bool): Initially set to ``False``, changed if any value is appended.
+    """
+
+    def __init__(
+        self,
+        option_strings: List[str],
+        dest: str,
+        nargs: Any = None,
+        **kwargs: Any,
+    ) -> None:
+        """Intialize."""
+        super().__init__(option_strings, dest, **kwargs)
+        self._called = False
+
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: Union[str, Sequence[Any], None],
+        option_string: Optional[str] = None,
+    ) -> None:
+        g = getattr(namespace, self.dest, None)
+        if g is None:
+            g = []
+        if values is not None and not self._called:
+            # If any value was specified, we then clear the list of options before appending.
+            # We cannot always clear the ``default`` attribute since it collects the ``values`` appended.
+            self.default.clear()
+            self._called = True
+        g.append(values)
+
+
 def add_argument(
     toolparser: argparse.ArgumentParser,
     name: str,
@@ -864,7 +900,7 @@ def add_argument(
         elif inptype["items"] == "Directory":
             action = DirectoryAppendAction
         else:
-            action = "append"
+            action = AppendAction
     elif isinstance(inptype, MutableMapping) and inptype["type"] == "enum":
         atype = str
     elif isinstance(inptype, MutableMapping) and inptype["type"] == "record":

cwltool/executors.py

@@ -278,7 +278,7 @@ def __init__(self) -> None:
         self.pending_jobs = []  # type: List[JobsType]
         self.pending_jobs_lock = threading.Lock()
 
-        self.max_ram = int(psutil.virtual_memory().available / 2**20)  # type: ignore[no-untyped-call]
+        self.max_ram = int(psutil.virtual_memory().available / 2 ** 20)  # type: ignore[no-untyped-call]
         self.max_cores = float(psutil.cpu_count())
         self.allocated_ram = float(0)
         self.allocated_cores = float(0)

cwltool/job.py

@@ -555,7 +555,7 @@ def get_tree_mem_usage(memory_usage: MutableSequence[Optional[int]]) -> None:
             _logger.info(
                 "[job %s] Max memory used: %iMiB",
                 self.name,
-                round(memory_usage[0] / (2**20)),
+                round(memory_usage[0] / (2 ** 20)),
             )
         else:
             _logger.debug(
@@ -943,7 +943,7 @@ def docker_monitor(
         _logger.info(
             "[job %s] Max memory used: %iMiB",
             self.name,
-            int((max_mem_percent / 100 * max_mem) / (2**20)),
+            int((max_mem_percent / 100 * max_mem) / (2 ** 20)),
         )
         if cleanup_cidfile:
             os.remove(cidfile)

tests/default_values_list.cwl

@@ -0,0 +1,20 @@
+#!/usr/bin/env cwl-runner
+# From https://github.com/common-workflow-language/cwltool/issues/1632
+
+cwlVersion: v1.2
+class: CommandLineTool
+
+baseCommand: [cat]
+
+stdout: "cat_file"
+
+inputs:
+  file_paths:
+    type: string[]?
+    inputBinding:
+        position: 1
+    default: ["/home/bart/cwl_test/test1"]
+
+outputs:
+  output:
+    type: stdout

tests/test_toolargparse.py

@@ -1,7 +1,7 @@
 import argparse
 from io import StringIO
 from pathlib import Path
-from typing import Callable
+from typing import Any, Callable, Dict, List
 
 import pytest
 
@@ -195,3 +195,35 @@ def test_argparser_without_doc() -> None:
     p = argparse.ArgumentParser()
     parser = generate_parser(p, tool, {}, [], False)
     assert parser.description is None
+
+
+@pytest.mark.parametrize(
+    "job_order,expected_values",
+    [
+        # no arguments, so we expect the default value
+        ([], ["/home/bart/cwl_test/test1"]),
+        # arguments, provided, one or many, meaning that the default value is not expected
+        (["--file_paths", "/home/bart/cwl_test/test2"], ["/home/bart/cwl_test/test2"]),
+        (
+            [
+                "--file_paths",
+                "/home/bart/cwl_test/test2",
+                "--file_paths",
+                "/home/bart/cwl_test/test3",
+            ],
+            ["/home/bart/cwl_test/test2", "/home/bart/cwl_test/test3"],
+        ),
+    ],
+)
+def test_argparse_append_with_default(
+    job_order: List[str], expected_values: List[str]
+) -> None:
+    """The appended arguments must not include the default. But if no appended argument, then the default is used."""
+    loadingContext = LoadingContext()
+    tool = load_tool(get_data("tests/default_values_list.cwl"), loadingContext)
+    toolparser = generate_parser(
+        argparse.ArgumentParser(prog="test"), tool, {}, [], False
+    )
+    cmd_line = vars(toolparser.parse_args(job_order))  # type: ignore[call-overload]
+    file_paths = list(cmd_line["file_paths"])
+    assert expected_values == file_paths