diff --git a/tests/tools/test_singularity_preflight.py b/tests/tools/test_singularity_preflight.py new file mode 100644 index 000000000..c9ece8751 --- /dev/null +++ b/tests/tools/test_singularity_preflight.py @@ -0,0 +1,77 @@ +"""Tests for Singularity/Apptainer preflight availability check. + +Verifies that a clear error is raised when neither apptainer nor +singularity is installed, instead of a cryptic FileNotFoundError. + +See: https://github.com/NousResearch/hermes-agent/issues/1511 +""" + +import subprocess +from unittest.mock import patch, MagicMock + +import pytest + +from tools.environments.singularity import ( + _find_singularity_executable, + _ensure_singularity_available, +) + + +class TestFindSingularityExecutable: + """_find_singularity_executable resolution tests.""" + + def test_prefers_apptainer(self): + """When both are available, apptainer should be preferred.""" + def which_both(name): + return f"/usr/bin/{name}" if name in ("apptainer", "singularity") else None + + with patch("shutil.which", side_effect=which_both): + assert _find_singularity_executable() == "apptainer" + + def test_falls_back_to_singularity(self): + """When only singularity is available, use it.""" + def which_singularity_only(name): + return "/usr/bin/singularity" if name == "singularity" else None + + with patch("shutil.which", side_effect=which_singularity_only): + assert _find_singularity_executable() == "singularity" + + def test_raises_when_neither_found(self): + """Must raise RuntimeError with install instructions.""" + with patch("shutil.which", return_value=None): + with pytest.raises(RuntimeError, match="neither.*apptainer.*nor.*singularity"): + _find_singularity_executable() + + +class TestEnsureSingularityAvailable: + """_ensure_singularity_available preflight tests.""" + + def test_returns_executable_on_success(self): + """Returns the executable name when version check passes.""" + fake_result = MagicMock(returncode=0, stderr="") + + with patch("shutil.which", side_effect=lambda n: "/usr/bin/apptainer" if n == "apptainer" else None), \ + patch("subprocess.run", return_value=fake_result): + assert _ensure_singularity_available() == "apptainer" + + def test_raises_on_version_failure(self): + """Raises RuntimeError when version command fails.""" + fake_result = MagicMock(returncode=1, stderr="unknown flag") + + with patch("shutil.which", side_effect=lambda n: "/usr/bin/apptainer" if n == "apptainer" else None), \ + patch("subprocess.run", return_value=fake_result): + with pytest.raises(RuntimeError, match="version.*failed"): + _ensure_singularity_available() + + def test_raises_on_timeout(self): + """Raises RuntimeError when version command times out.""" + with patch("shutil.which", side_effect=lambda n: "/usr/bin/apptainer" if n == "apptainer" else None), \ + patch("subprocess.run", side_effect=subprocess.TimeoutExpired("apptainer", 10)): + with pytest.raises(RuntimeError, match="timed out"): + _ensure_singularity_available() + + def test_raises_when_not_installed(self): + """Raises RuntimeError when neither executable exists.""" + with patch("shutil.which", return_value=None): + with pytest.raises(RuntimeError, match="neither.*apptainer.*nor.*singularity"): + _ensure_singularity_available() diff --git a/tools/environments/singularity.py b/tools/environments/singularity.py index f462cd055..72afbac5f 100644 --- a/tools/environments/singularity.py +++ b/tools/environments/singularity.py @@ -25,6 +25,57 @@ logger = logging.getLogger(__name__) _SNAPSHOT_STORE = get_hermes_home() / "singularity_snapshots.json" +def _find_singularity_executable() -> str: + """Locate the apptainer or singularity CLI binary. + + Returns the executable name (``"apptainer"`` or ``"singularity"``). + Raises ``RuntimeError`` with install instructions if neither is found. + """ + if shutil.which("apptainer"): + return "apptainer" + if shutil.which("singularity"): + return "singularity" + raise RuntimeError( + "Neither 'apptainer' nor 'singularity' was found in PATH. " + "Install Apptainer (https://apptainer.org/docs/admin/main/installation.html) " + "or Singularity and ensure the CLI is available." + ) + + +def _ensure_singularity_available() -> str: + """Preflight check: resolve the executable and verify it responds. + + Returns the executable name on success. + Raises ``RuntimeError`` with an actionable message on failure. + """ + exe = _find_singularity_executable() + + try: + result = subprocess.run( + [exe, "version"], + capture_output=True, + text=True, + timeout=10, + ) + except FileNotFoundError: + raise RuntimeError( + f"Singularity backend selected but the resolved executable '{exe}' " + "could not be executed. Check your installation." + ) + except subprocess.TimeoutExpired: + raise RuntimeError( + f"'{exe} version' timed out. The runtime may be misconfigured." + ) + + if result.returncode != 0: + stderr = result.stderr.strip()[:200] + raise RuntimeError( + f"'{exe} version' failed (exit code {result.returncode}): {stderr}" + ) + + return exe + + def _load_snapshots() -> Dict[str, str]: if _SNAPSHOT_STORE.exists(): try: @@ -169,7 +220,7 @@ class SingularityEnvironment(BaseEnvironment): task_id: str = "default", ): super().__init__(cwd=cwd, timeout=timeout) - self.executable = "apptainer" if shutil.which("apptainer") else "singularity" + self.executable = _ensure_singularity_available() self.image = _get_or_build_sif(image, self.executable) self.instance_id = f"hermes_{uuid.uuid4().hex[:12]}" self._instance_started = False