Merge remote-tracking branch 'origin/main' into try-typed-dicts-from-cwl-utils

Author: mr-c

cwltool/argparser.py

@@ -543,6 +543,16 @@ def arg_parser() -> argparse.ArgumentParser:
         dest="pull_image",
     )
 
+    container_group.add_argument(
+        "--singularity-sandbox-path",
+        default=None,
+        type=str,
+        help="Singularity/Apptainer sandbox image base path. "
+        "Will use a pre-existing sandbox image. "
+        "Will be prepended to the dockerPull path. "
+        "Equivalent to use CWL_SINGULARITY_IMAGES variable. ",
+        dest="image_base_path",
+    )
     container_group.add_argument(
         "--force-docker-pull",
         action="store_true",

cwltool/checker.py

@@ -25,6 +25,7 @@ def check_types(
     srctype: SinkType | None,
     sinktype: SinkType | None,
     linkMerge: str | None,
+    pickValue: str | None,
     valueFrom: str | None,
 ) -> Literal["pass"] | Literal["warning"] | Literal["exception"]:
     """
@@ -34,23 +35,49 @@ def check_types(
     """
     if valueFrom is not None:
         return "pass"
+    if pickValue is not None:
+        if (
+            isinstance((_srctype := _get_type(srctype)), MutableMapping)
+            and _srctype["type"] == "array"
+        ):
+            match pickValue:
+                case "all_non_null":
+                    _srctype = {"type": "array", "items": _srctype["items"]}
+                    if (
+                        isinstance(_srctype["items"], MutableSequence)
+                        and "null" in _srctype["items"]
+                    ):
+                        _srctype["items"].remove("null")
+                case "first_non_null" | "the_only_non_null":
+                    if (
+                        isinstance(_srctype["items"], MutableSequence)
+                        and "null" in _srctype["items"]
+                    ):
+                        _srctype = [elem for elem in _srctype["items"] if elem != "null"]
+                case _:
+                    raise WorkflowException(f"Unrecognized pickValue enum {pickValue!r}")
+        _sinktype = _get_type(sinktype)
+    else:
+        _srctype = srctype
+        _sinktype = sinktype
     match linkMerge:
         case None:
-            if can_assign_src_to_sink(srctype, sinktype, strict=True):
+            if can_assign_src_to_sink(_srctype, _sinktype, strict=True):
                 return "pass"
-            if can_assign_src_to_sink(srctype, sinktype, strict=False):
+            if can_assign_src_to_sink(_srctype, _sinktype, strict=False):
                 return "warning"
             return "exception"
         case "merge_nested":
             return check_types(
-                {"items": _get_type(srctype), "type": "array"},
+                {"items": _get_type(_srctype), "type": "array"},
                 _get_type(sinktype),
                 None,
                 None,
+                None,
             )
         case "merge_flattened":
             return check_types(
-                merge_flatten_type(_get_type(srctype)), _get_type(sinktype), None, None
+                merge_flatten_type(_get_type(_srctype)), _get_type(sinktype), None, None, None
             )
         case _:
             raise WorkflowException(f"Unrecognized linkMerge enum {linkMerge!r}")
@@ -373,7 +400,15 @@ def _check_all_types(
                 srcs_of_sink: list[CWLObjectType] = []
                 for parm_id in cast(MutableSequence[str], sink[sourceField]):
                     srcs_of_sink += [src_dict[parm_id]]
-                    if is_conditional_step(param_to_step, parm_id) and pickValue is None:
+                    sink_type = cast(
+                        Union[str, list[str], list[CWLObjectType], CWLObjectType], sink["type"]
+                    )
+                    if (
+                        is_conditional_step(param_to_step, parm_id)
+                        and "null" != sink_type
+                        and "null" not in sink_type
+                        and pickValue is None
+                    ):
                         validation["warning"].append(
                             _SrcSink(
                                 src_dict[parm_id],
@@ -435,7 +470,7 @@ def _check_all_types(
                     }
 
             for src in srcs_of_sink:
-                check_result = check_types(src, sink, linkMerge, valueFrom)
+                check_result = check_types(src, sink, linkMerge, pickValue, valueFrom)
                 if check_result == "warning":
                     validation["warning"].append(
                         _SrcSink(src, sink, linkMerge, message=extra_message)

cwltool/context.py

@@ -152,6 +152,7 @@ def __init__(self, kwargs: dict[str, Any] | None = None) -> None:
         self.streaming_allowed: bool = False
 
         self.singularity: bool = False
+        self.image_base_path: str | None = None
         self.podman: bool = False
         self.debug: bool = False
         self.compute_checksum: bool = True

cwltool/docker.py

@@ -204,6 +204,7 @@ def get_from_requirements(
         pull_image: bool,
         force_pull: bool,
         tmp_outdir_prefix: str,
+        image_base_path: str | None = None,
     ) -> str | None:
         if not shutil.which(self.docker_exec):
             raise WorkflowException(f"{self.docker_exec} executable is not available")

cwltool/extensions-v1.1.yml

@@ -30,6 +30,11 @@ $graph:
   type: record
   inVocab: true
   extends: cwl:Process
+  specialize:
+    - specializeFrom: cwl:InputParameter
+      specializeTo: cwl:WorkflowInputParameter
+    - specializeFrom: cwl:OutputParameter
+      specializeTo: cwl:ExpressionToolOutputParameter
   documentRoot: true
   fields:
     - name: class

cwltool/extensions-v1.2.yml

@@ -30,6 +30,11 @@ $graph:
   type: record
   inVocab: true
   extends: cwl:Process
+  specialize:
+    - specializeFrom: cwl:InputParameter
+      specializeTo: cwl:WorkflowInputParameter
+    - specializeFrom: cwl:OutputParameter
+      specializeTo: cwl:ExpressionToolOutputParameter
   documentRoot: true
   fields:
     - name: class

cwltool/extensions-v1.3.yml

@@ -29,6 +29,11 @@ $graph:
   type: record
   inVocab: true
   extends: cwl:Process
+  specialize:
+    - specializeFrom: cwl:InputParameter
+      specializeTo: cwl:WorkflowInputParameter
+    - specializeFrom: cwl:OutputParameter
+      specializeTo: cwl:ExpressionToolOutputParameter
   documentRoot: true
   fields:
     - name: class

cwltool/extensions.yml

@@ -141,6 +141,9 @@ $graph:
   type: record
   inVocab: true
   extends: cwl:Process
+  specialize:
+    - specializeFrom: cwl:OutputParameter
+      specializeTo: cwl:ExpressionToolOutputParameter
   documentRoot: true
   fields:
     - name: class

cwltool/job.py

@@ -629,6 +629,7 @@ def get_from_requirements(
         pull_image: bool,
         force_pull: bool,
         tmp_outdir_prefix: str,
+        image_base_path: str | None = None,
     ) -> str | None:
         pass
 
@@ -790,6 +791,7 @@ def run(
                             runtimeContext.pull_image,
                             runtimeContext.force_docker_pull,
                             runtimeContext.tmp_outdir_prefix,
+                            runtimeContext.image_base_path,
                         )
                     )
                 if img_id is None:

cwltool/singularity.py

@@ -2,6 +2,7 @@
 
 import copy
 import hashlib
+import json
 import logging
 import os
 import os.path
@@ -10,10 +11,11 @@
 import sys
 import threading
 from collections.abc import Callable, MutableMapping, MutableSequence
-from subprocess import check_call, check_output  # nosec
+from subprocess import check_call, check_output, run  # nosec
 from typing import cast
 
 from cwl_utils.types import CWLDirectoryType, CWLFileType, CWLObjectType
+from mypy_extensions import mypyc_attr
 from packaging.version import Version
 from schema_salad.sourceline import SourceLine
 from schema_salad.utils import json_dumps
@@ -165,6 +167,30 @@ def _normalize_sif_id(string: str) -> str:
     return string.replace("/", "_") + ".sif"
 
 
+@mypyc_attr(allow_interpreted_subclasses=True)
+def _inspect_singularity_sandbox_image(path: str) -> bool:
+    """Inspect singularity sandbox image to be sure it is not an empty directory."""
+    cmd = [
+        "singularity",
+        "inspect",
+        "--json",
+        path,
+    ]
+    try:
+        result = run(cmd, capture_output=True, text=True)  # nosec
+    except Exception:
+        return False
+
+    if result.returncode == 0:
+        try:
+            output = json.loads(result.stdout)
+        except json.JSONDecodeError:
+            return False
+        if output.get("data", {}).get("attributes", {}):
+            return True
+    return False
+
+
 class SingularityCommandLineJob(ContainerCommandLineJob):
     def __init__(
         self,
@@ -186,6 +212,7 @@ def get_image(
         pull_image: bool,
         tmp_outdir_prefix: str,
         force_pull: bool = False,
+        sandbox_base_path: str | None = None,
     ) -> bool:
         """
         Acquire the software container image in the specified dockerRequirement.
@@ -204,17 +231,34 @@ def get_image(
 
         with _IMAGES_LOCK:
             if "dockerImageId" in dockerRequirement:
-                if (d_image_id := dockerRequirement["dockerImageId"]) in _IMAGES:
+                d_image_id = dockerRequirement["dockerImageId"]
+                if d_image_id in _IMAGES:
                     if (resolved_image_id := _IMAGES[d_image_id]) != d_image_id:
                         dockerRequirement["dockerImage_id"] = resolved_image_id
                     return True
+                if d_image_id.startswith("/"):
+                    _logger.info(
+                        SourceLine(dockerRequirement, "dockerImageId").makeError(
+                            f"Non-portable: using an absolute file path in a 'dockerImageId': {d_image_id}"
+                        )
+                    )
 
         docker_req = copy.deepcopy(dockerRequirement)  # thread safety
         if "CWL_SINGULARITY_CACHE" in os.environ:
             cache_folder = os.environ["CWL_SINGULARITY_CACHE"]
         elif is_version_2_6() and "SINGULARITY_PULLFOLDER" in os.environ:
             cache_folder = os.environ["SINGULARITY_PULLFOLDER"]
 
+        if os.environ.get("CWL_SINGULARITY_IMAGES", None):
+            image_base_path = os.environ["CWL_SINGULARITY_IMAGES"]
+        else:
+            image_base_path = cache_folder if cache_folder else ""
+
+        if not sandbox_base_path:
+            sandbox_base_path = os.path.abspath(image_base_path)
+        else:
+            sandbox_base_path = os.path.abspath(sandbox_base_path)
+
         if "dockerFile" in docker_req:
             if cache_folder is None:  # if environment variables were not set
                 cache_folder = create_tmp_dir(tmp_outdir_prefix)
@@ -264,21 +308,44 @@ def get_image(
                 )
                 found = True
         elif "dockerImageId" not in docker_req and "dockerPull" in docker_req:
-            match = re.search(pattern=r"([a-z]*://)", string=docker_req["dockerPull"])
-            img_name = _normalize_image_id(docker_req["dockerPull"])
-            candidates.append(img_name)
-            if is_version_3_or_newer():
-                sif_name = _normalize_sif_id(docker_req["dockerPull"])
-                candidates.append(sif_name)
-                docker_req["dockerImageId"] = sif_name
+            # looking for local singularity sandbox image and handle it as a local image
+            sandbox_image_path = os.path.join(sandbox_base_path, dockerRequirement["dockerPull"])
+            if os.path.isdir(sandbox_image_path) and _inspect_singularity_sandbox_image(
+                sandbox_image_path
+            ):
+                docker_req["dockerImageId"] = sandbox_image_path
+                _logger.info(
+                    "Using local Singularity sandbox image found in %s",
+                    sandbox_image_path,
+                )
+                found = True
             else:
-                docker_req["dockerImageId"] = img_name
-            if not match:
-                docker_req["dockerPull"] = "docker://" + docker_req["dockerPull"]
+                match = re.search(pattern=r"([a-z]*://)", string=docker_req["dockerPull"])
+                img_name = _normalize_image_id(docker_req["dockerPull"])
+                candidates.append(img_name)
+                if is_version_3_or_newer():
+                    sif_name = _normalize_sif_id(docker_req["dockerPull"])
+                    candidates.append(sif_name)
+                    docker_req["dockerImageId"] = sif_name
+                else:
+                    docker_req["dockerImageId"] = img_name
+                if not match:
+                    docker_req["dockerPull"] = "docker://" + docker_req["dockerPull"]
         elif "dockerImageId" in docker_req:
-            if os.path.isfile(docker_req["dockerImageId"]):
+            sandbox_image_path = os.path.join(sandbox_base_path, dockerRequirement["dockerImageId"])
+            # handling local singularity sandbox image
+            if os.path.isdir(sandbox_image_path) and _inspect_singularity_sandbox_image(
+                sandbox_image_path
+            ):
+                _logger.info(
+                    "Using local Singularity sandbox image found in %s",
+                    sandbox_image_path,
+                )
+                docker_req["dockerImageId"] = sandbox_image_path
                 found = True
             else:
+                if os.path.isfile(docker_req["dockerImageId"]):
+                    found = True
                 candidates.append(docker_req["dockerImageId"])
                 candidates.append(_normalize_image_id(docker_req["dockerImageId"]))
                 if is_version_3_or_newer():
@@ -297,18 +364,19 @@ def get_image(
                             path = os.path.join(dirpath, entry)
                             if os.path.isfile(path):
                                 _logger.info(
-                                    "Using local copy of Singularity image found in %s",
+                                    "Using local copy of Singularity image %s found in %s",
+                                    entry,
                                     dirpath,
                                 )
                                 docker_req["dockerImageId"] = path
                                 found = True
         if (force_pull or not found) and pull_image:
             cmd: list[str] = []
             if "dockerPull" in docker_req:
-                if cache_folder:
+                if image_base_path:
                     env = os.environ.copy()
                     if is_version_2_6():
-                        env["SINGULARITY_PULLFOLDER"] = cache_folder
+                        env["SINGULARITY_PULLFOLDER"] = image_base_path
                         cmd = [
                             "singularity",
                             "pull",
@@ -323,14 +391,14 @@ def get_image(
                             "pull",
                             "--force",
                             "--name",
-                            "{}/{}".format(cache_folder, docker_req["dockerImageId"]),
+                            "{}/{}".format(image_base_path, docker_req["dockerImageId"]),
                             str(docker_req["dockerPull"]),
                         ]
 
                     _logger.info(str(cmd))
                     check_call(cmd, env=env, stdout=sys.stderr)  # nosec
                     docker_req["dockerImageId"] = "{}/{}".format(
-                        cache_folder, docker_req["dockerImageId"]
+                        image_base_path, docker_req["dockerImageId"]
                     )
                     found = True
                 else:
@@ -388,6 +456,7 @@ def get_from_requirements(
         pull_image: bool,
         force_pull: bool,
         tmp_outdir_prefix: str,
+        image_base_path: str | None = None,
     ) -> str | None:
         """
         Return the filename of the Singularity image.
@@ -397,8 +466,14 @@ def get_from_requirements(
         if not bool(shutil.which("singularity")):
             raise WorkflowException("singularity executable is not available")
 
-        if not self.get_image(cast(dict[str, str], r), pull_image, tmp_outdir_prefix, force_pull):
-            raise WorkflowException("Container image {} not found".format(r["dockerImageId"]))
+        if not self.get_image(
+            cast(dict[str, str], r),
+            pull_image,
+            tmp_outdir_prefix,
+            force_pull,
+            sandbox_base_path=image_base_path,
+        ):
+            raise WorkflowException(f"Container image not found for {r}")
 
         return os.path.abspath(cast(str, r["dockerImageId"]))