Enhanced detection of singularity version including a distribution detection

cwltool/singularity.py

@@ -20,40 +20,96 @@
 from .singularity_utils import singularity_supports_userns
 from .utils import CWLObjectType, create_tmp_dir, ensure_non_writable, ensure_writable
 
-_SINGULARITY_VERSION = ""
-
-
-def get_version() -> str:
+# Cached version number of singularity
+# This is a list containing major and minor versions as integer.
+# (The number of minor version digits can vary among different distributions,
+#  therefore we need a list here.)
+_SINGULARITY_VERSION: Optional[List[int]] = None
+# Cached flavor / distribution of singularity
+# Can be singularity, singularity-ce or apptainer
+_SINGULARITY_FLAVOR: str = ""
+
+
+def get_version() -> Tuple[List[int], str]:
+    """
+    Parse the output of 'singularity --version' to determine the singularity flavor /
+    distribution (singularity, singularity-ce or apptainer) and the singularity version.
+    Both pieces of information will be cached.
+
+    Returns
+    -------
+    A tuple containing:
+    - A tuple with major and minor version numbers as integer.
+    - A string with the name of the singularity flavor.
+    """
     global _SINGULARITY_VERSION  # pylint: disable=global-statement
-    if _SINGULARITY_VERSION == "":
-        _SINGULARITY_VERSION = check_output(  # nosec
+    global _SINGULARITY_FLAVOR  # pylint: disable=global-statement
+    if _SINGULARITY_VERSION is None:
+        version_output = check_output(  # nosec
             ["singularity", "--version"], universal_newlines=True
         ).strip()
-        if _SINGULARITY_VERSION.startswith("singularity version "):
-            _SINGULARITY_VERSION = _SINGULARITY_VERSION[20:]
-        if _SINGULARITY_VERSION.startswith("singularity-ce version "):
-            _SINGULARITY_VERSION = _SINGULARITY_VERSION[23:]
-        _logger.debug(f"Singularity version: {_SINGULARITY_VERSION}.")
-    return _SINGULARITY_VERSION
+
+        version_match = re.match(r"(.+) version ([0-9\.]+)", version_output)
+        if version_match is None:
+            raise RuntimeError("Output of 'singularity --version' not recognized.")
+
+        version_string = version_match.group(2)
+        _SINGULARITY_VERSION = [int(i) for i in version_string.split(".")]
+        _SINGULARITY_FLAVOR = version_match.group(1)
+
+        _logger.debug(
+            f"Singularity version: {version_string}" " ({_SINGULARITY_FLAVOR}."
+        )
+    return (_SINGULARITY_VERSION, _SINGULARITY_FLAVOR)
+
+
+def is_apptainer_1_or_newer() -> bool:
+    """
+    Check if apptainer singularity distribution is version 1.0 or higher.
+
+    Apptainer v1.0.0 is compatible with SingularityCE 3.9.5.
+    See: https://github.com/apptainer/apptainer/releases
+    """
+    v = get_version()
+    if v[1] != "apptainer":
+        return False
+    return v[0][0] >= 1
 
 
 def is_version_2_6() -> bool:
-    return get_version().startswith("2.6")
+    """
+    Check if this singularity version is exactly version 2.6.
+
+    Also returns False if the flavor is not singularity or singularity-ce.
+    """
+    v = get_version()
+    if v[1] != "singularity" and v[1] != "singularity-ce":
+        return False
+    return v[0][0] == 2 and v[0][1] == 6
 
 
 def is_version_3_or_newer() -> bool:
-    return int(get_version()[0]) >= 3
+    """Check if this version is singularity version 3 or newer or equivalent."""
+    if is_apptainer_1_or_newer():
+        return True  # this is equivalent to singularity-ce > 3.9.5
+    v = get_version()
+    return v[0][0] >= 3
 
 
 def is_version_3_1_or_newer() -> bool:
-    version = get_version().split(".")
-    return int(version[0]) >= 4 or (int(version[0]) == 3 and int(version[1]) >= 1)
+    """Check if this version is singularity version 3.1 or newer or equivalent."""
+    if is_apptainer_1_or_newer():
+        return True  # this is equivalent to singularity-ce > 3.9.5
+    v = get_version()
+    return v[0][0] >= 4 or (v[0][0] == 3 and v[0][1] >= 1)
 
 
 def is_version_3_4_or_newer() -> bool:
     """Detect if Singularity v3.4+ is available."""
-    version = get_version().split(".")
-    return int(version[0]) >= 4 or (int(version[0]) == 3 and int(version[1]) >= 4)
+    if is_apptainer_1_or_newer():
+        return True  # this is equivalent to singularity-ce > 3.9.5
+    v = get_version()
+    return v[0][0] >= 4 or (v[0][0] == 3 and v[0][1] >= 4)
 
 
 def _normalize_image_id(string: str) -> str:

tests/test_environment.py

@@ -131,10 +131,10 @@ def PWD(v: str, env: Env) -> bool:
         }
 
         # Singularity variables appear to be in flux somewhat.
-        version = get_version().split(".")
-        vmajor = int(version[0])
+        version = get_version()[0]
+        vmajor = version[0]
         assert vmajor == 3, "Tests only work for Singularity 3"
-        vminor = int(version[1])
+        vminor = version[1]
         sing_vars: EnvChecks = {
             "SINGULARITY_CONTAINER": None,
             "SINGULARITY_NAME": None,

tests/test_singularity_versions.py

@@ -0,0 +1,139 @@
+"""Test singularity{,-ce} & apptainer versions."""
+import cwltool.singularity
+from cwltool.singularity import (
+    get_version,
+    is_apptainer_1_or_newer,
+    is_version_2_6,
+    is_version_3_or_newer,
+    is_version_3_1_or_newer,
+    is_version_3_4_or_newer,
+)
+
+from subprocess import check_output  # nosec
+
+
+def reset_singularity_version_cache() -> None:
+    """Reset the cache for testing."""
+    cwltool.singularity._SINGULARITY_VERSION = None
+    cwltool.singularity._SINGULARITY_FLAVOR = ""
+
+
+def set_dummy_check_output(name: str, version: str) -> None:
+    """Mock out subprocess.check_output."""
+    cwltool.singularity.check_output = (  # type: ignore[attr-defined]
+        lambda c, universal_newlines: name + " version " + version
+    )
+
+
+def restore_check_output() -> None:
+    """Undo the mock of subprocess.check_output."""
+    cwltool.singularity.check_output = check_output  # type: ignore[attr-defined]
+
+
+def test_get_version() -> None:
+    """Confirm expected types of singularity.get_version()."""
+    set_dummy_check_output("apptainer", "1.0.1")
+    reset_singularity_version_cache()
+    v = get_version()
+    assert isinstance(v, tuple)
+    assert isinstance(v[0], list)
+    assert isinstance(v[1], str)
+    assert (
+        cwltool.singularity._SINGULARITY_VERSION is not None
+    )  # pylint: disable=protected-access
+    assert (
+        len(cwltool.singularity._SINGULARITY_FLAVOR) > 0
+    )  # pylint: disable=protected-access
+    v_cached = get_version()
+    assert v == v_cached
+
+    assert v[0][0] == 1
+    assert v[0][1] == 0
+    assert v[0][2] == 1
+    assert v[1] == "apptainer"
+
+    set_dummy_check_output("singularity", "3.8.5")
+    reset_singularity_version_cache()
+    v = get_version()
+
+    assert v[0][0] == 3
+    assert v[0][1] == 8
+    assert v[0][2] == 5
+    assert v[1] == "singularity"
+    restore_check_output()
+
+
+def test_version_checks() -> None:
+    """Confirm logic in the various singularity version checks."""
+    set_dummy_check_output("apptainer", "1.0.1")
+    reset_singularity_version_cache()
+    assert is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert is_version_3_or_newer()
+    assert is_version_3_1_or_newer()
+    assert is_version_3_4_or_newer()
+
+    set_dummy_check_output("apptainer", "0.0.1")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert not is_version_3_or_newer()
+    assert not is_version_3_1_or_newer()
+    assert not is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "0.0.1")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert not is_version_3_or_newer()
+    assert not is_version_3_1_or_newer()
+    assert not is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "0.1")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert not is_version_3_or_newer()
+    assert not is_version_3_1_or_newer()
+    assert not is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "2.6")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert is_version_2_6()
+    assert not is_version_3_or_newer()
+    assert not is_version_3_1_or_newer()
+    assert not is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "3.0")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert is_version_3_or_newer()
+    assert not is_version_3_1_or_newer()
+    assert not is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "3.1")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert is_version_3_or_newer()
+    assert is_version_3_1_or_newer()
+    assert not is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "3.4")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert is_version_3_or_newer()
+    assert is_version_3_1_or_newer()
+    assert is_version_3_4_or_newer()
+
+    set_dummy_check_output("singularity", "3.6.3")
+    reset_singularity_version_cache()
+    assert not is_apptainer_1_or_newer()
+    assert not is_version_2_6()
+    assert is_version_3_or_newer()
+    assert is_version_3_1_or_newer()
+    assert is_version_3_4_or_newer()
+    restore_check_output()