test: improve test and linting

Author: natthan-pigoux

cwltool/singularity.py

@@ -6,10 +6,9 @@
 import os.path
 import re
 import shutil
-import subprocess
 import sys
 from collections.abc import Callable, MutableMapping
-from subprocess import DEVNULL, check_call, check_output, run  # nosec
+from subprocess import check_call, check_output, run  # nosec
 from typing import cast
 
 from schema_salad.sourceline import SourceLine
@@ -146,6 +145,7 @@ def _normalize_sif_id(string: str) -> str:
         string += "_latest"
     return string.replace("/", "_") + ".sif"
 
+
 def _inspect_singularity_image(path: str) -> bool:
     """Inspect singularity image to be sure it is not an empty directory."""
     cmd = [
@@ -158,16 +158,17 @@ def _inspect_singularity_image(path: str) -> bool:
         result = run(cmd, capture_output=True, text=True)
     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', {}):
+        if output.get("data", {}).get("attributes", {}):
             return True
     return False
 
+
 class SingularityCommandLineJob(ContainerCommandLineJob):
     def __init__(
         self,
@@ -253,7 +254,9 @@ def get_image(
                 found = True
         elif "dockerImageId" not in dockerRequirement and "dockerPull" in dockerRequirement:
             # looking for local singularity sandbox image and handle it as a local image
-            if os.path.isdir(dockerRequirement["dockerPull"]) and _inspect_singularity_image(dockerRequirement["dockerPull"]):
+            if os.path.isdir(dockerRequirement["dockerPull"]) and _inspect_singularity_image(
+                dockerRequirement["dockerPull"]
+            ):
                 dockerRequirement["dockerImageId"] = dockerRequirement["dockerPull"]
                 _logger.info(
                     "Using local Singularity sandbox image found in %s",
@@ -280,7 +283,9 @@ def get_image(
                 if is_version_3_or_newer():
                     candidates.append(_normalize_sif_id(dockerRequirement["dockerImageId"]))
             # handling local singularity sandbox image
-            elif os.path.isdir(dockerRequirement["dockerImageId"]) and _inspect_singularity_image(dockerRequirement["dockerImageId"]):
+            elif os.path.isdir(dockerRequirement["dockerImageId"]) and _inspect_singularity_image(
+                dockerRequirement["dockerImageId"]
+            ):
                 _logger.info(
                     "Using local Singularity sandbox image found in %s",
                     dockerRequirement["dockerImageId"],

tests/test_singularity.py

@@ -1,11 +1,13 @@
 """Tests to find local Singularity image."""
 
 import shutil
+import subprocess
 from pathlib import Path
 
 import pytest
 
 from cwltool.main import main
+from cwltool.singularity import _inspect_singularity_image
 
 from .util import (
     get_data,
@@ -160,22 +162,23 @@ def test_singularity3_docker_image_id_in_tool(tmp_path: Path) -> None:
         )
         assert result_code1 == 0
 
+
 @needs_singularity
 def test_singularity_local_sandbox_image(tmp_path: Path):
-    import subprocess
     workdir = tmp_path / "working_dir"
     workdir.mkdir()
     with working_directory(workdir):
         # build a sandbox image
-        container_path = Path(f"{workdir}/container_repo/")
+        container_path = workdir / "container_repo"
         container_path.mkdir()
         cmd = [
-            "apptainer",
+            "singularity",
             "build",
             "--sandbox",
-            container_path / "alpine",
-            "docker://alpine:latest"
+            str(container_path / "alpine"),
+            "docker://alpine:latest",
         ]
+
         build = subprocess.run(cmd, capture_output=True, text=True)
         if build.returncode == 0:
             result_code, stdout, stderr = get_main_output(
@@ -189,4 +192,39 @@ def test_singularity_local_sandbox_image(tmp_path: Path):
             )
             assert result_code == 0
         else:
-            pytest.skip(f"Failed to build the Singularity image: {build.stderr}")
+            pytest.skip(f"Failed to build the singularity image: {build.stderr}")
+
+
+@needs_singularity
+def test_singularity_inspect_image(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    workdir = tmp_path / "working_dir"
+    workdir.mkdir()
+    repo_path = workdir / "container_repo"
+    image_path = repo_path / "alpine"
+
+    # test path does not exists
+    res_inspect = _inspect_singularity_image(str(image_path))
+    assert res_inspect is False
+
+    # test image exists
+    repo_path.mkdir()
+    cmd = [
+        "singularity",
+        "build",
+        "--sandbox",
+        str(image_path),
+        "docker://alpine:latest",
+    ]
+    build = subprocess.run(cmd, capture_output=True, text=True)
+    if build.returncode == 0:
+        # Verify the path is a container image
+        res_inspect = _inspect_singularity_image(image_path)
+        assert res_inspect is True
+
+    # test wrong subprocess call
+    def mock_failed_subprocess(*args, **kwargs):
+        raise subprocess.CalledProcessError(returncode=1, cmd=args[0])
+    
+    monkeypatch.setattr("cwltool.singularity.run", mock_failed_subprocess)
+    res_inspect = _inspect_singularity_image(str(image_path))
+    assert res_inspect is False

tests/test_tmpdir.py

@@ -285,6 +285,93 @@ def test_dockerfile_singularity_build(monkeypatch: pytest.MonkeyPatch, tmp_path:
     shutil.rmtree(subdir)
 
 
+@needs_singularity
+def test_singularity_get_image_from_sandbox(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
+    """Test that SingularityCommandLineJob.get_image correctly handle sandbox image."""
+    
+    (tmp_path / "out").mkdir(exist_ok=True)
+    tmp_outdir_prefix = tmp_path / "out"
+    (tmp_path / "3").mkdir(exist_ok=True)
+    tmpdir_prefix = str(tmp_path / "tmp")
+    runtime_context = RuntimeContext(
+        {"tmpdir_prefix": tmpdir_prefix, "user_space_docker_cmd": None}
+    )
+    builder = Builder(
+        {},
+        [],
+        [],
+        {},
+        schema.Names(),
+        [],
+        [],
+        {},
+        None,
+        None,
+        StdFsAccess,
+        StdFsAccess(""),
+        None,
+        0.1,
+        True,
+        False,
+        False,
+        "no_listing",
+        runtime_context.get_outdir(),
+        runtime_context.get_tmpdir(),
+        runtime_context.get_stagedir(),
+        INTERNAL_VERSION,
+        "singularity",
+    )
+
+    workdir = tmp_path / "working_dir"
+    workdir.mkdir()
+    repo_path = workdir / "container_repo"
+    repo_path.mkdir()
+    image_path = repo_path / "alpine"
+    image_path.mkdir()
+
+    # directory exists but is not an image
+    monkeypatch.setattr("cwltool.singularity._inspect_singularity_image", lambda *args, **kwargs: False)
+    req = {"class": "DockerRequirement", "dockerPull": f"{image_path}"}
+    res = SingularityCommandLineJob(
+        builder, {}, CommandLineTool.make_path_mapper, [], [], ""
+    ).get_image(
+        req,
+        pull_image=False,
+        tmp_outdir_prefix=str(tmp_outdir_prefix),
+        force_pull=False,
+    )
+    assert req["dockerPull"].startswith("docker://")
+    assert res is False
+
+    # directory exists and is an image:
+    monkeypatch.setattr("cwltool.singularity._inspect_singularity_image", lambda *args, **kwargs: True)
+    req = {"class": "DockerRequirement", "dockerPull": f"{image_path}"}
+    res =  SingularityCommandLineJob(
+        builder, {}, CommandLineTool.make_path_mapper, [], [], ""
+    ).get_image(
+        req,
+        pull_image=False,
+        tmp_outdir_prefix=str(tmp_outdir_prefix),
+        force_pull=False,
+    )
+    assert req["dockerPull"] == str(image_path)
+    assert req["dockerImageId"] == str(image_path)
+    assert res
+
+    # test that dockerImageId is set and image exists:
+    req = {"class": "DockerRequirement", "dockerImageId": f"{image_path}"}
+    res = SingularityCommandLineJob(
+        builder, {}, CommandLineTool.make_path_mapper, [], [], ""
+    ).get_image(
+        req,
+        pull_image=False,
+        tmp_outdir_prefix=str(tmp_outdir_prefix),
+        force_pull=False,
+    )
+    assert req["dockerImageId"] == str(image_path)
+    assert res
+
+
 def test_docker_tmpdir_prefix(tmp_path: Path) -> None:
     """Test that DockerCommandLineJob respects temp directory directives."""
     (tmp_path / "3").mkdir()