diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/setup.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/setup.py index 5f06f6a4e4..1982ef60b0 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/setup.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/setup.py @@ -28,6 +28,7 @@ import yaml as _yaml from nemo_platform import NeMoPlatform from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.config import validate_docker_available from nemo_platform_plugin.secrets.client import SecretsClient from nemo_platform_plugin.secrets.types import PlatformSecretCreateRequest, PlatformSecretUpdateRequest from nmp.common.config import nmp_user_data_dir @@ -808,17 +809,23 @@ def _wait_for_platform( timeout: int = _SERVICE_STARTUP_TIMEOUT_SECONDS, poll_interval: float = _SERVICE_STARTUP_POLL_INTERVAL, log_path: Path | None = None, + proc: subprocess.Popen | None = None, ) -> bool: """Poll until the platform health endpoint responds. Returns True on success. When *log_path* is provided, the spinner shows the last service that finished loading so users see that progress is being made during a slow cold start. + + When *proc* is provided, return False immediately if the service process + exits before the platform becomes ready (avoid waiting the full timeout). """ start = time.monotonic() deadline = start + timeout with console.status("[bold cyan]Waiting for platform...") as status: while time.monotonic() < deadline: + if proc is not None and proc.poll() is not None: + return False elapsed = int(time.monotonic() - start) svc = _last_startup_service(log_path) hint = f" — loaded {svc}" if svc else "" @@ -829,6 +836,39 @@ def _wait_for_platform( return False +_DOCKER_FAILURE_LOG_MARKERS = ( + "Docker daemon is unavailable", + "Docker is unavailable", + "docker.from_env", + "Error while fetching server API version", +) + + +def _services_log_suggests_docker_failure(log_path: Path | None) -> bool: + """Return True when services.log contains known Docker soft-skip / daemon errors.""" + if log_path is None or not log_path.is_file(): + return False + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return any(marker in text for marker in _DOCKER_FAILURE_LOG_MARKERS) + + +def _should_hint_docker_unavailable(*, exit_code: int | None, log_path: Path | None) -> bool: + """Decide whether to print a Docker-missing hint after a failed startup wait. + + Prefer log evidence. Otherwise only hint when the process exited early and + a Docker ping confirms the daemon is unavailable — not on a pure readiness + timeout while the process is still alive. + """ + if _services_log_suggests_docker_failure(log_path): + return True + if exit_code is not None and not validate_docker_available(): + return True + return False + + def _kill_existing_services(base_url: str) -> None: """Find and kill any running ``nemo services run`` processes. @@ -938,7 +978,7 @@ def _maybe_start_services( log = log_path_for(compute_scope(port=_resolve_services_port(base_url))) - if not _wait_for_platform(base_url, timeout=timeout, log_path=log): + if not _wait_for_platform(base_url, timeout=timeout, log_path=log, proc=proc): exit_code = proc.poll() if exit_code is not None: console.print(f"{CROSS} Service process exited early (exit code {exit_code})") @@ -946,6 +986,11 @@ def _maybe_start_services( proc.terminate() console.print(f"{CROSS} Platform did not become ready within {timeout}s") console.print(f" Check {log} for details.") + if _should_hint_docker_unavailable(exit_code=exit_code, log_path=log): + console.print( + " Docker does not appear to be available. " + "Install and start Docker, or configure non-Docker executors, then retry." + ) raise typer.Exit(1) console.print(f"{CHECK} Platform running at {base_url} (pid {proc.pid})\n") diff --git a/packages/nemo_platform_ext/tests/cli/commands/test_setup.py b/packages/nemo_platform_ext/tests/cli/commands/test_setup.py index fa128545f1..6567594b85 100644 --- a/packages/nemo_platform_ext/tests/cli/commands/test_setup.py +++ b/packages/nemo_platform_ext/tests/cli/commands/test_setup.py @@ -760,6 +760,66 @@ def test_allows_start_when_port_free(self, maybe_start_preflight_mocks): _maybe_start_services("http://localhost:8080", auto=False, start_services=True) maybe_start_preflight_mocks.assert_called_once() + def test_early_exit_prints_docker_hint_when_daemon_unavailable(self, maybe_start_preflight_mocks, capsys): + dead = MagicMock(pid=999) + dead.poll.return_value = 3 + wait = MagicMock(return_value=False) + with ( + patch(f"{SETUP_MOD}.check_port_available_for_start", return_value=None), + patch(f"{SETUP_MOD}._wait_for_platform", wait), + patch(f"{SETUP_MOD}.validate_docker_available", return_value=False), + patch(f"{SETUP_MOD}.log_path_for", return_value=MagicMock(__str__=lambda self: "/tmp/services.log")), + patch(f"{SETUP_MOD}._pause"), + pytest.raises(ClickExit), + ): + maybe_start_preflight_mocks.return_value = dead + _maybe_start_services("http://localhost:8080", auto=False, start_services=True) + wait.assert_called_once() + assert wait.call_args.kwargs.get("proc") is dead + captured = capsys.readouterr() + assert "exited early (exit code 3)" in captured.err + assert "Check /tmp/services.log for details." in captured.err + assert "Docker does not appear to be available" in captured.err + + def test_readiness_timeout_does_not_hint_docker_without_evidence( + self, maybe_start_preflight_mocks, capsys, tmp_path + ): + alive = MagicMock(pid=999) + alive.poll.return_value = None + log = tmp_path / "services.log" + log.write_text("starting auth\n", encoding="utf-8") + with ( + patch(f"{SETUP_MOD}.check_port_available_for_start", return_value=None), + patch(f"{SETUP_MOD}._wait_for_platform", return_value=False), + patch(f"{SETUP_MOD}.validate_docker_available", return_value=False), + patch(f"{SETUP_MOD}.log_path_for", return_value=log), + patch(f"{SETUP_MOD}._pause"), + pytest.raises(ClickExit), + ): + maybe_start_preflight_mocks.return_value = alive + _maybe_start_services("http://localhost:8080", auto=False, start_services=True) + captured = capsys.readouterr() + assert "did not become ready" in captured.err + assert "Docker does not appear to be available" not in captured.err + + def test_docker_hint_from_services_log_markers(self, maybe_start_preflight_mocks, capsys, tmp_path): + alive = MagicMock(pid=999) + alive.poll.return_value = None + log = tmp_path / "services.log" + log.write_text("Skipping executor 'local-docker': Docker daemon is unavailable\n", encoding="utf-8") + with ( + patch(f"{SETUP_MOD}.check_port_available_for_start", return_value=None), + patch(f"{SETUP_MOD}._wait_for_platform", return_value=False), + patch(f"{SETUP_MOD}.validate_docker_available", return_value=True), + patch(f"{SETUP_MOD}.log_path_for", return_value=log), + patch(f"{SETUP_MOD}._pause"), + pytest.raises(ClickExit), + ): + maybe_start_preflight_mocks.return_value = alive + _maybe_start_services("http://localhost:8080", auto=False, start_services=True) + captured = capsys.readouterr() + assert "Docker does not appear to be available" in captured.err + class TestRemoteConnection: def test_prompts_again_until_platform_is_reachable(self, capsys): @@ -2019,6 +2079,21 @@ def test_spinner_works_without_log_path(self, spinner_console): update_texts = [c.args[0] for c in mock_status.update.call_args_list] assert all("loaded" not in t for t in update_texts) + def test_returns_false_immediately_when_process_exits(self, spinner_console): + """Do not wait the full timeout when the service process already exited.""" + dead = MagicMock() + dead.poll.return_value = 3 + with ( + patch(f"{self._MOD}._pause") as pause, + patch(f"{self._MOD}.time.monotonic", side_effect=[0, 0, 1, 2]), + patch(f"{self._MOD}._check_platform_reachable") as reachable, + ): + result = _wait_for_platform("http://localhost:8080", timeout=120, proc=dead) + + assert result is False + reachable.assert_not_called() + pause.assert_not_called() + # --------------------------------------------------------------------------- # _last_startup_service tests diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/config.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/config.py index 4703772145..f638c83afe 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/config.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/config.py @@ -82,6 +82,7 @@ def test_something(): from pydantic import BaseModel, Field, field_validator, model_validator from pydantic._internal._model_construction import ModelMetaclass from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict +from requests.exceptions import ConnectionError as RequestsConnectionError from requests.exceptions import Timeout as RequestsTimeout import docker @@ -388,7 +389,7 @@ def validate_docker_available() -> bool: client = docker.from_env(timeout=5) client.ping() return True - except (DockerException, RequestsTimeout): + except (DockerException, RequestsConnectionError, RequestsTimeout, OSError): return False finally: if client: diff --git a/packages/nemo_platform_plugin/tests/test_config.py b/packages/nemo_platform_plugin/tests/test_config.py index e1a7307446..995836aa16 100644 --- a/packages/nemo_platform_plugin/tests/test_config.py +++ b/packages/nemo_platform_plugin/tests/test_config.py @@ -268,3 +268,25 @@ class _RegisteredPlatformConfig(NemoPlatformConfig): def test_get_nemo_platform_config_alias() -> None: """get_nemo_platform_config is an alias for get_platform_config.""" assert get_nemo_platform_config is get_platform_config + + +def test_validate_docker_available_returns_false_on_connection_failures() -> None: + from unittest.mock import MagicMock, patch + + from docker.errors import DockerException + from nemo_platform_plugin.config import validate_docker_available + from requests.exceptions import ConnectionError as RequestsConnectionError + + for exc in ( + DockerException("boom"), + RequestsConnectionError("refused"), + OSError("no such file"), + ): + with patch("nemo_platform_plugin.config.docker.from_env", side_effect=exc): + assert validate_docker_available() is False + + client = MagicMock() + client.ping.side_effect = exc + with patch("nemo_platform_plugin.config.docker.from_env", return_value=client): + assert validate_docker_available() is False + client.close.assert_called() diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/base.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/base.py index 16c5d2c30c..f96e5fcb4d 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/base.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/base.py @@ -16,13 +16,13 @@ class MissingBackendDependencyError(RuntimeError): - """A backend's optional dependency (its packaging extra) is not installed. + """A backend cannot initialize because a required capability is unavailable. - Backends whose substrate SDK ships as an optional extra (e.g. ``openshell``) - raise this when that SDK is absent, so the executor registry can skip the - executor with a warning instead of failing the whole deployments service at - startup. Subclasses ``RuntimeError`` so existing ``except RuntimeError`` paths - keep working. + Raised when an optional packaging extra is missing (e.g. ``openshell``) or when + a runtime substrate is unreachable (e.g. Docker daemon/socket). The executor + registry catches this, skips that executor with a warning, and continues + starting the deployments service. Subclasses ``RuntimeError`` so existing + ``except RuntimeError`` paths keep working. """ diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py index 988cfaa0d5..df1418aecb 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py @@ -14,6 +14,7 @@ BackendStatusUpdate, DeploymentBackend, LogResult, + MissingBackendDependencyError, VolumeStatusUpdate, ) from nemo_deployments_plugin.backends.docker import volumes as volume_ops @@ -65,6 +66,7 @@ from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityNotFoundError from requests.exceptions import ConnectionError as RequestsConnectionError from requests.exceptions import ReadTimeout +from requests.exceptions import Timeout as RequestsTimeout from urllib3.exceptions import ReadTimeoutError as Urllib3ReadTimeoutError if TYPE_CHECKING: @@ -97,10 +99,12 @@ class DockerDeploymentBackend(DeploymentBackend): def init(self) -> None: try: + from docker.errors import DockerException + import docker from docker import errors as docker_errors except ImportError as exc: - raise RuntimeError( + raise MissingBackendDependencyError( "docker package is required for DockerDeploymentBackend. " "Install with: uv sync --package nemo-deployments-plugin --extra docker" ) from exc @@ -110,7 +114,12 @@ def init(self) -> None: self._executor_config = DockerExecutorConfig.model_validate(self._config) self._entities = NemoEntitiesClient(client_from_platform(self._sdk, AsyncEntitiesClient)) self._gpu_pool = get_shared_gpu_pool() - self._client = self._create_client() + try: + self._client = self._create_client() + except (DockerException, RequestsConnectionError, RequestsTimeout, OSError) as exc: + raise MissingBackendDependencyError( + f"Docker daemon is unavailable ({exc}). Docker-backed deployments will be disabled." + ) from exc def _create_client(self) -> docker.DockerClient: kwargs: dict[str, Any] = {"timeout": self._executor_config.docker_timeout} @@ -118,6 +127,7 @@ def _create_client(self) -> docker.DockerClient: kwargs["base_url"] = self._executor_config.docker_host client = self._docker.from_env(**kwargs) client.api.timeout = self._executor_config.docker_timeout + client.ping() return client def shutdown(self) -> None: diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/registry.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/registry.py index 9915586b78..7e09da6840 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/registry.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/registry.py @@ -66,11 +66,10 @@ def from_config( try: executors[spec.name] = classes[spec.backend](sdk, spec.config) except MissingBackendDependencyError as exc: - # An opt-in backend whose optional extra isn't installed (e.g. - # 'openshell') must not take down the whole deployments service: - # skip just that executor. Resolving it later raises - # ExecutorNotFoundError, and a `default_executor` that can't be - # built still fails fast via the check below. + # Capability missing (optional packaging extra, unreachable Docker + # daemon, etc.): skip just that executor so the deployments service + # can still boot. Resolving a skipped name later raises + # ExecutorNotFoundError; a skipped default is cleared below. logger.warning( "Skipping executor '%s': backend '%s' is unavailable (%s)", spec.name, @@ -78,7 +77,12 @@ def from_config( exc, ) if default_executor and default_executor not in executors: - raise ExecutorNotFoundError(f"default_executor '{default_executor}' is not registered.") + logger.warning( + "Clearing default_executor '%s' because it is not registered " + "(unavailable backend or missing from executor config).", + default_executor, + ) + default_executor = None except Exception: for backend in executors.values(): backend.shutdown() diff --git a/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py b/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py index b228293242..510cb6d248 100644 --- a/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py +++ b/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py @@ -14,8 +14,8 @@ published_port_config, sample_config, ) -from docker.errors import APIError, NotFound -from nemo_deployments_plugin.backends.base import BackendStatusUpdate +from docker.errors import APIError, DockerException, NotFound +from nemo_deployments_plugin.backends.base import BackendStatusUpdate, MissingBackendDependencyError from nemo_deployments_plugin.backends.docker import ports as ports_mod from nemo_deployments_plugin.backends.docker.backend import ( _PORT_CONFLICT_ATTEMPTS, @@ -43,6 +43,30 @@ from requests.exceptions import ReadTimeout +def test_init_raises_missing_dependency_when_docker_daemon_unavailable(mock_sdk: MagicMock) -> None: + with ( + patch("nemo_deployments_plugin.backends.docker.backend.client_from_platform"), + patch("nemo_deployments_plugin.backends.docker.backend.NemoEntitiesClient"), + patch("nemo_deployments_plugin.backends.docker.backend.get_shared_gpu_pool", return_value=None), + patch("docker.from_env", side_effect=DockerException("Error while fetching server API version")), + ): + with pytest.raises(MissingBackendDependencyError, match="Docker daemon is unavailable"): + DockerDeploymentBackend(mock_sdk, {"docker_timeout": 5, "pull_images": False}) + + +def test_init_raises_missing_dependency_when_docker_ping_fails(mock_sdk: MagicMock) -> None: + client = MagicMock() + client.ping.side_effect = RequestsConnectionError("Connection refused") + with ( + patch("nemo_deployments_plugin.backends.docker.backend.client_from_platform"), + patch("nemo_deployments_plugin.backends.docker.backend.NemoEntitiesClient"), + patch("nemo_deployments_plugin.backends.docker.backend.get_shared_gpu_pool", return_value=None), + patch("docker.from_env", return_value=client), + ): + with pytest.raises(MissingBackendDependencyError, match="Docker daemon is unavailable"): + DockerDeploymentBackend(mock_sdk, {"docker_timeout": 5, "pull_images": False}) + + @pytest.mark.asyncio async def test_create_deployment_starts_container( docker_backend: DockerDeploymentBackend, diff --git a/plugins/nemo-deployments/tests/unit/test_registry.py b/plugins/nemo-deployments/tests/unit/test_registry.py index a9b1ff8bb8..658ff0456a 100644 --- a/plugins/nemo-deployments/tests/unit/test_registry.py +++ b/plugins/nemo-deployments/tests/unit/test_registry.py @@ -181,20 +181,50 @@ def test_registry_skips_backend_with_missing_dependency( registry.resolve("sandbox-local") -def test_registry_missing_dependency_default_executor_still_fails( +def test_registry_missing_dependency_default_executor_cleared( backend_classes: dict[str, type[DeploymentBackend]], + caplog: pytest.LogCaptureFixture, ) -> None: - # A skipped backend is fine as a secondary executor, but if it is the - # configured default the registry must still fail fast. + # A skipped default must not take down the deployments service: clear it + # and boot with no default so callers must name an executor explicitly. sdk = AsyncNeMoPlatform(base_url="http://localhost:8080") classes = {**backend_classes, "sandbox": _MissingDepBackend} - with pytest.raises(ExecutorNotFoundError): - ExecutorRegistry.from_config( + with caplog.at_level("WARNING"): + registry = ExecutorRegistry.from_config( sdk, [ExecutorSpec(name="sandbox-local", backend="sandbox", config={})], default_executor="sandbox-local", backend_classes=classes, ) + assert registry.registered_names() == [] + assert "Clearing default_executor 'sandbox-local'" in caplog.text + with pytest.raises(ExecutorNotFoundError, match="no default_executor"): + registry.resolve() + + +def test_registry_skips_unavailable_docker_and_keeps_other_backends( + backend_classes: dict[str, type[DeploymentBackend]], +) -> None: + sdk = AsyncNeMoPlatform(base_url="http://localhost:8080") + + class _UnavailableDocker(_StubBackend): + def init(self) -> None: + raise MissingBackendDependencyError("Docker daemon is unavailable") + + classes = {**backend_classes, "docker": _UnavailableDocker} + registry = ExecutorRegistry.from_config( + sdk, + [ + ExecutorSpec(name="local-docker", backend="docker", config={}), + ExecutorSpec(name="cluster-a", backend="k8s", config={}), + ], + default_executor="local-docker", + backend_classes=classes, + ) + assert registry.registered_names() == ["cluster-a"] + assert registry.resolve("cluster-a") is not None + with pytest.raises(ExecutorNotFoundError, match="no default_executor"): + registry.resolve() def test_multiple_docker_executors_distinct_config() -> None: diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/commands/setup.py b/sdk/python/nemo-platform/src/nemo_platform/cli/commands/setup.py index d5d6b32ab9..b0df960b6d 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/cli/commands/setup.py +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/commands/setup.py @@ -28,6 +28,7 @@ import yaml as _yaml from nemo_platform import NeMoPlatform from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.config import validate_docker_available from nemo_platform_plugin.secrets.client import SecretsClient from nemo_platform_plugin.secrets.types import PlatformSecretCreateRequest, PlatformSecretUpdateRequest from nmp.common.config import nmp_user_data_dir @@ -808,17 +809,23 @@ def _wait_for_platform( timeout: int = _SERVICE_STARTUP_TIMEOUT_SECONDS, poll_interval: float = _SERVICE_STARTUP_POLL_INTERVAL, log_path: Path | None = None, + proc: subprocess.Popen | None = None, ) -> bool: """Poll until the platform health endpoint responds. Returns True on success. When *log_path* is provided, the spinner shows the last service that finished loading so users see that progress is being made during a slow cold start. + + When *proc* is provided, return False immediately if the service process + exits before the platform becomes ready (avoid waiting the full timeout). """ start = time.monotonic() deadline = start + timeout with console.status("[bold cyan]Waiting for platform...") as status: while time.monotonic() < deadline: + if proc is not None and proc.poll() is not None: + return False elapsed = int(time.monotonic() - start) svc = _last_startup_service(log_path) hint = f" — loaded {svc}" if svc else "" @@ -829,6 +836,39 @@ def _wait_for_platform( return False +_DOCKER_FAILURE_LOG_MARKERS = ( + "Docker daemon is unavailable", + "Docker is unavailable", + "docker.from_env", + "Error while fetching server API version", +) + + +def _services_log_suggests_docker_failure(log_path: Path | None) -> bool: + """Return True when services.log contains known Docker soft-skip / daemon errors.""" + if log_path is None or not log_path.is_file(): + return False + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return any(marker in text for marker in _DOCKER_FAILURE_LOG_MARKERS) + + +def _should_hint_docker_unavailable(*, exit_code: int | None, log_path: Path | None) -> bool: + """Decide whether to print a Docker-missing hint after a failed startup wait. + + Prefer log evidence. Otherwise only hint when the process exited early and + a Docker ping confirms the daemon is unavailable — not on a pure readiness + timeout while the process is still alive. + """ + if _services_log_suggests_docker_failure(log_path): + return True + if exit_code is not None and not validate_docker_available(): + return True + return False + + def _kill_existing_services(base_url: str) -> None: """Find and kill any running ``nemo services run`` processes. @@ -938,7 +978,7 @@ def _maybe_start_services( log = log_path_for(compute_scope(port=_resolve_services_port(base_url))) - if not _wait_for_platform(base_url, timeout=timeout, log_path=log): + if not _wait_for_platform(base_url, timeout=timeout, log_path=log, proc=proc): exit_code = proc.poll() if exit_code is not None: console.print(f"{CROSS} Service process exited early (exit code {exit_code})") @@ -946,6 +986,11 @@ def _maybe_start_services( proc.terminate() console.print(f"{CROSS} Platform did not become ready within {timeout}s") console.print(f" Check {log} for details.") + if _should_hint_docker_unavailable(exit_code=exit_code, log_path=log): + console.print( + " Docker does not appear to be available. " + "Install and start Docker, or configure non-Docker executors, then retry." + ) raise typer.Exit(1) console.print(f"{CHECK} Platform running at {base_url} (pid {proc.pid})\n") diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_setup.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_setup.py index 038bdba8fc..d75bc24dc4 100644 --- a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_setup.py +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_setup.py @@ -760,6 +760,66 @@ def test_allows_start_when_port_free(self, maybe_start_preflight_mocks): _maybe_start_services("http://localhost:8080", auto=False, start_services=True) maybe_start_preflight_mocks.assert_called_once() + def test_early_exit_prints_docker_hint_when_daemon_unavailable(self, maybe_start_preflight_mocks, capsys): + dead = MagicMock(pid=999) + dead.poll.return_value = 3 + wait = MagicMock(return_value=False) + with ( + patch(f"{SETUP_MOD}.check_port_available_for_start", return_value=None), + patch(f"{SETUP_MOD}._wait_for_platform", wait), + patch(f"{SETUP_MOD}.validate_docker_available", return_value=False), + patch(f"{SETUP_MOD}.log_path_for", return_value=MagicMock(__str__=lambda self: "/tmp/services.log")), + patch(f"{SETUP_MOD}._pause"), + pytest.raises(ClickExit), + ): + maybe_start_preflight_mocks.return_value = dead + _maybe_start_services("http://localhost:8080", auto=False, start_services=True) + wait.assert_called_once() + assert wait.call_args.kwargs.get("proc") is dead + captured = capsys.readouterr() + assert "exited early (exit code 3)" in captured.err + assert "Check /tmp/services.log for details." in captured.err + assert "Docker does not appear to be available" in captured.err + + def test_readiness_timeout_does_not_hint_docker_without_evidence( + self, maybe_start_preflight_mocks, capsys, tmp_path + ): + alive = MagicMock(pid=999) + alive.poll.return_value = None + log = tmp_path / "services.log" + log.write_text("starting auth\n", encoding="utf-8") + with ( + patch(f"{SETUP_MOD}.check_port_available_for_start", return_value=None), + patch(f"{SETUP_MOD}._wait_for_platform", return_value=False), + patch(f"{SETUP_MOD}.validate_docker_available", return_value=False), + patch(f"{SETUP_MOD}.log_path_for", return_value=log), + patch(f"{SETUP_MOD}._pause"), + pytest.raises(ClickExit), + ): + maybe_start_preflight_mocks.return_value = alive + _maybe_start_services("http://localhost:8080", auto=False, start_services=True) + captured = capsys.readouterr() + assert "did not become ready" in captured.err + assert "Docker does not appear to be available" not in captured.err + + def test_docker_hint_from_services_log_markers(self, maybe_start_preflight_mocks, capsys, tmp_path): + alive = MagicMock(pid=999) + alive.poll.return_value = None + log = tmp_path / "services.log" + log.write_text("Skipping executor 'local-docker': Docker daemon is unavailable\n", encoding="utf-8") + with ( + patch(f"{SETUP_MOD}.check_port_available_for_start", return_value=None), + patch(f"{SETUP_MOD}._wait_for_platform", return_value=False), + patch(f"{SETUP_MOD}.validate_docker_available", return_value=True), + patch(f"{SETUP_MOD}.log_path_for", return_value=log), + patch(f"{SETUP_MOD}._pause"), + pytest.raises(ClickExit), + ): + maybe_start_preflight_mocks.return_value = alive + _maybe_start_services("http://localhost:8080", auto=False, start_services=True) + captured = capsys.readouterr() + assert "Docker does not appear to be available" in captured.err + class TestRemoteConnection: def test_prompts_again_until_platform_is_reachable(self, capsys): @@ -2019,6 +2079,21 @@ def test_spinner_works_without_log_path(self, spinner_console): update_texts = [c.args[0] for c in mock_status.update.call_args_list] assert all("loaded" not in t for t in update_texts) + def test_returns_false_immediately_when_process_exits(self, spinner_console): + """Do not wait the full timeout when the service process already exited.""" + dead = MagicMock() + dead.poll.return_value = 3 + with ( + patch(f"{self._MOD}._pause") as pause, + patch(f"{self._MOD}.time.monotonic", side_effect=[0, 0, 1, 2]), + patch(f"{self._MOD}._check_platform_reachable") as reachable, + ): + result = _wait_for_platform("http://localhost:8080", timeout=120, proc=dead) + + assert result is False + reachable.assert_not_called() + pause.assert_not_called() + # --------------------------------------------------------------------------- # _last_startup_service tests diff --git a/services/core/jobs/src/nmp/core/jobs/config.py b/services/core/jobs/src/nmp/core/jobs/config.py index e0f3146c7c..3ab8698a0d 100644 --- a/services/core/jobs/src/nmp/core/jobs/config.py +++ b/services/core/jobs/src/nmp/core/jobs/config.py @@ -70,11 +70,13 @@ def validate_executors(self) -> Self: # Module-level singleton instances config = get_service_config(JobsServiceConfig) +platform_runtime = get_platform_config().runtime profiles = merge_executor_profiles( config.executors, get_default_executor_profiles_for_runtime( - runtime=get_platform_config().runtime, + runtime=platform_runtime, defaults=config.executor_defaults, enable_subprocess_executor=config.resolved_enable_subprocess_executor(), ), + runtime=platform_runtime, ) diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/config.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/config.py index 72a9010d00..998d22eedd 100644 --- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/config.py +++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/config.py @@ -3,6 +3,7 @@ import logging +from nemo_platform_plugin.config import validate_docker_available from nmp.common.config import Runtime from nmp.core.jobs.app.profiles import ExecutionProfileT from nmp.core.jobs.controllers.backends.docker import DockerJobExecutionProfile, DockerJobExecutionProfileConfig @@ -20,6 +21,8 @@ logger = logging.getLogger(__name__) +_RUNTIME_NONE_UNSUPPORTED_BACKENDS = frozenset({"kubernetes_job", "volcano_job"}) + class DefaultExecutionProfileConfig(BaseModel): """Holds default execution profile configurations for various backends.""" @@ -126,22 +129,59 @@ def get_default_executor_profiles_for_runtime( def merge_executor_profiles( custom_executors: list[ExecutionProfileT], default_executors: list[ExecutionProfileT], + *, + runtime: Runtime | None = None, ) -> list[ExecutionProfileT]: """ Merge custom executor profiles with default profiles, giving precedence to custom profiles. If a custom profile has the same provider and profile as a default profile, the custom profile will override the default. If the custom profile matching a default profile has custom config values, those should override the default config values. + When runtime is NONE, Kubernetes- and Volcano-backed custom profiles are skipped. Docker profiles (default or custom) + are skipped whenever Docker is unavailable, independent of Runtime, so explicit Docker executors can still run in + compose/test harnesses with a mounted socket. """ merged_executors: dict[tuple[str, str], ExecutionProfileT] = {} + docker_available: bool | None = None + + def _docker_is_available() -> bool: + nonlocal docker_available + if docker_available is None: + docker_available = validate_docker_available() + return docker_available # Add default profiles first for executor in default_executors: + if executor.backend == "docker" and not _docker_is_available(): + logger.warning( + "Skipping executor profile %s/%s using backend '%s' because Docker is unavailable.", + executor.provider, + executor.profile, + executor.backend, + ) + continue merged_executors[(executor.provider, executor.profile)] = executor # Override with custom profiles for custom_executor in custom_executors: + if runtime == Runtime.NONE and custom_executor.backend in _RUNTIME_NONE_UNSUPPORTED_BACKENDS: + logger.warning( + "Skipping executor profile %s/%s using backend '%s' because platform runtime is NONE.", + custom_executor.provider, + custom_executor.profile, + custom_executor.backend, + ) + continue + if custom_executor.backend == "docker" and not _docker_is_available(): + logger.warning( + "Skipping executor profile %s/%s using backend '%s' because Docker is unavailable.", + custom_executor.provider, + custom_executor.profile, + custom_executor.backend, + ) + continue + key = (custom_executor.provider, custom_executor.profile) # If the custom executor matches a default, update the default config with custom values diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/registry.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/registry.py index 7d572ba1d9..90d3a2247f 100644 --- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/registry.py +++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/registry.py @@ -1,10 +1,13 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import logging from dataclasses import dataclass from typing import Self, Sequence +from docker.errors import DockerException from nemo_platform import NeMoPlatform +from nemo_platform_plugin.config import validate_docker_available from nmp.core.jobs.app.profiles import ExecutionProfileT from nmp.core.jobs.app.schemas import BackendRef, ProfileRef, ProviderRef from nmp.core.jobs.controllers.backends.base import DEFAULT_PROFILE, DEFAULT_PROVIDER, JobBackend @@ -16,6 +19,14 @@ ) from nmp.core.jobs.controllers.backends.subprocess import SubprocessJobBackend from nmp.core.jobs.controllers.backends.test import TestE2ECPUJobBackend, TestE2EGPUJobBackend +from requests.exceptions import ConnectionError as RequestsConnectionError +from requests.exceptions import Timeout as RequestsTimeout + +logger = logging.getLogger(__name__) + +# Soft-skip only daemon/connection failures after validate_docker_available() was True. +# ValidationError and other programming/config errors must still fail startup. +_DOCKER_BACKEND_INIT_SOFT_SKIP_ERRORS = (DockerException, RequestsConnectionError, RequestsTimeout, OSError) @dataclass(frozen=True) @@ -107,20 +118,51 @@ def from_config( ValidationError: If a profile's configuration is invalid for its backend type """ registry: dict[RegistryKey, JobBackend] = {} + docker_available: bool | None = None for executor in profiles: # Execution profiles are unique with respect to the provider # and profile combination registry_key = RegistryKey(executor.provider, executor.profile) backend_key = BackendKey(executor.provider, executor.backend) + if backend_key not in backends: + raise KeyError( + f"No backend registered for provider '{executor.provider}' and backend '{executor.backend}'" + ) backend = backends[backend_key] + if executor.backend == "docker": + if docker_available is None: + docker_available = validate_docker_available() + if not docker_available: + logger.warning( + "Skipping job executor profile %s/%s using backend 'docker' because Docker is unavailable.", + executor.provider, + executor.profile, + ) + continue + # The config from the execution profile hasn't been validated # yet. Calling the backend constructor will serialize the raw # config into the backend's expected format and validate it - registry[registry_key] = backend(nmp_sdk, executor.config, executor.profile) + try: + registry[registry_key] = backend(nmp_sdk, executor.config, executor.profile) + except _DOCKER_BACKEND_INIT_SOFT_SKIP_ERRORS as exc: + if executor.backend != "docker": + raise + logger.warning( + "Skipping job executor profile %s/%s using backend 'docker' " + "because Docker backend initialization failed (%s).", + executor.provider, + executor.profile, + exc, + ) return cls(registry) + def registered_profile_keys(self) -> frozenset[tuple[str, str]]: + """Return (provider, profile) keys for backends that successfully registered.""" + return frozenset((key.provider, key.profile) for key in self._registry) + def get_backend(self, *, provider: str | None = None, profile: str | None = None) -> JobBackend: """Retrieve a configured backend for the specified provider and profile. diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/main.py b/services/core/jobs/src/nmp/core/jobs/controllers/main.py index 15343d1eae..cfeb9d71f9 100644 --- a/services/core/jobs/src/nmp/core/jobs/controllers/main.py +++ b/services/core/jobs/src/nmp/core/jobs/controllers/main.py @@ -17,6 +17,26 @@ from nmp.core.jobs.controllers.scheduler import JobScheduler stop_signal = threading.Event() +logger = logging.getLogger(__name__) + + +def _sync_advertised_profiles_to_registry(backend_registry: BackendRegistry) -> None: + """Keep GET /v2/execution-profiles aligned with backends that actually registered. + + ``profiles`` is built at import time; ``BackendRegistry.from_config`` may later + soft-skip Docker backends. Mutate the shared list in place so the API process + (same process in local standalone) stops advertising unusable executors. + """ + registered = backend_registry.registered_profile_keys() + kept = [profile for profile in profiles if (profile.provider, profile.profile) in registered] + skipped = len(profiles) - len(kept) + if skipped: + logger.warning( + "Removed %s execution profile(s) that failed backend registration so advertised " + "profiles match the jobs controller registry.", + skipped, + ) + profiles[:] = kept def handle_sighup(signum, frame): @@ -25,7 +45,6 @@ def handle_sighup(signum, frame): def run(parent_stop_signal: threading.Event | None = None): # Create logger after configuration is set up - logger = logging.getLogger(__name__) logger.info("Starting jobs controller") # Use provided stop signal or create our own @@ -43,6 +62,7 @@ def run(parent_stop_signal: threading.Event | None = None): logger.debug("Platform SDK initialized successfully.") backend_registry = BackendRegistry.from_config(nmp_sdk=nmp_sdk, profiles=profiles) + _sync_advertised_profiles_to_registry(backend_registry) logger.info("Executor backends registry initialized successfully.") # Wait for the jobs service to be ready before starting control loops (polls /status so we can start once jobs is ready) diff --git a/services/core/jobs/tests/test_config.py b/services/core/jobs/tests/test_config.py index 320d422b9b..f2991d5c5c 100644 --- a/services/core/jobs/tests/test_config.py +++ b/services/core/jobs/tests/test_config.py @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import logging import pathlib from unittest.mock import MagicMock, patch @@ -30,6 +31,7 @@ KubernetesJobExecutionProfile, KubernetesJobExecutionProfileConfig, KubernetesJobStorageConfig, + VolcanoJobExecutionProfile, VolcanoJobExecutionProfileConfig, ) from nmp.core.jobs.controllers.backends.registry import BackendKey, BackendRegistry, backend_registry @@ -387,6 +389,187 @@ def __init__(self, nmp_sdk, execution_profile_config, profile_name): assert registry.get_backend(provider="subprocess", profile="default") is not None +def test_backend_registry_skips_docker_when_unavailable(mock_nmp_client, caplog): + class DummyBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + self.nmp_sdk = nmp_sdk + self.execution_profile_config = execution_profile_config + self.profile_name = profile_name + + class ExplodingDockerBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + raise AssertionError("docker backend should not be constructed when unavailable") + + profiles = [ + DockerJobExecutionProfile( + provider="cpu", + profile="default", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + SubprocessJobExecutionProfile( + profile="default", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(), + ), + ] + + caplog.set_level(logging.WARNING) + with patch("nmp.core.jobs.controllers.backends.registry.validate_docker_available", return_value=False): + registry = BackendRegistry.from_config( + nmp_sdk=mock_nmp_client, + profiles=profiles, + backends={ + BackendKey("cpu", "docker"): ExplodingDockerBackend, + BackendKey("subprocess", "subprocess"): DummyBackend, + }, + ) + + assert registry.get_backend(provider="subprocess", profile="default") is not None + with pytest.raises(KeyError): + registry.get_backend(provider="cpu", profile="default") + assert "Skipping job executor profile cpu/default" in caplog.text + + +def test_backend_registry_registered_profile_keys_match_constructed_backends(mock_nmp_client): + class DummyBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + self.nmp_sdk = nmp_sdk + self.execution_profile_config = execution_profile_config + self.profile_name = profile_name + + profiles = [ + DockerJobExecutionProfile( + provider="cpu", + profile="default", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + SubprocessJobExecutionProfile( + profile="default", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(), + ), + ] + + with patch("nmp.core.jobs.controllers.backends.registry.validate_docker_available", return_value=False): + registry = BackendRegistry.from_config( + nmp_sdk=mock_nmp_client, + profiles=profiles, + backends={ + BackendKey("cpu", "docker"): DummyBackend, + BackendKey("subprocess", "subprocess"): DummyBackend, + }, + ) + + assert registry.registered_profile_keys() == frozenset({("subprocess", "default")}) + + +def test_backend_registry_soft_skips_docker_init_connection_errors(mock_nmp_client, caplog): + from docker.errors import DockerException + + class DummyBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + self.nmp_sdk = nmp_sdk + + class FailingDockerBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + raise DockerException("Error while fetching server API version") + + profiles = [ + DockerJobExecutionProfile( + provider="cpu", + profile="default", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + SubprocessJobExecutionProfile( + profile="default", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(), + ), + ] + + caplog.set_level(logging.WARNING) + with patch("nmp.core.jobs.controllers.backends.registry.validate_docker_available", return_value=True): + registry = BackendRegistry.from_config( + nmp_sdk=mock_nmp_client, + profiles=profiles, + backends={ + BackendKey("cpu", "docker"): FailingDockerBackend, + BackendKey("subprocess", "subprocess"): DummyBackend, + }, + ) + + assert registry.registered_profile_keys() == frozenset({("subprocess", "default")}) + assert "Docker backend initialization failed" in caplog.text + + +def test_backend_registry_propagates_non_connection_errors_for_docker(mock_nmp_client): + class BrokenConfigDockerBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + raise ValueError("bad executor config") + + profiles = [ + DockerJobExecutionProfile( + provider="cpu", + profile="default", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + ] + + with ( + patch("nmp.core.jobs.controllers.backends.registry.validate_docker_available", return_value=True), + pytest.raises(ValueError, match="bad executor config"), + ): + BackendRegistry.from_config( + nmp_sdk=mock_nmp_client, + profiles=profiles, + backends={BackendKey("cpu", "docker"): BrokenConfigDockerBackend}, + ) + + +def test_sync_advertised_profiles_to_registry_prunes_skipped_docker(mock_nmp_client, caplog): + from nmp.core.jobs.controllers import main as jobs_main + + class DummyBackend: + def __init__(self, nmp_sdk, execution_profile_config, profile_name): + self.nmp_sdk = nmp_sdk + + advertised = [ + DockerJobExecutionProfile( + provider="cpu", + profile="default", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + SubprocessJobExecutionProfile( + profile="default", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(), + ), + ] + + with patch("nmp.core.jobs.controllers.backends.registry.validate_docker_available", return_value=False): + registry = BackendRegistry.from_config( + nmp_sdk=mock_nmp_client, + profiles=advertised, + backends={ + BackendKey("cpu", "docker"): DummyBackend, + BackendKey("subprocess", "subprocess"): DummyBackend, + }, + ) + + with patch.object(jobs_main, "profiles", advertised): + caplog.set_level(logging.WARNING) + jobs_main._sync_advertised_profiles_to_registry(registry) + assert [(p.provider, p.profile, p.backend) for p in advertised] == [ + ("subprocess", "default", "subprocess"), + ] + assert "Removed 1 execution profile" in caplog.text + + def test_subprocess_execution_profile_defaults_provider_to_subprocess(): profile = SubprocessJobExecutionProfile(profile="default") @@ -426,6 +609,97 @@ def test_merge_executor_profiles_can_override_default_volcano_with_kubernetes_jo assert profile.config.namespace == "authentik" +def test_merge_executor_profiles_skips_container_backends_for_none_runtime(caplog): + defaults = get_default_executor_profiles_for_runtime(Runtime.NONE, DefaultExecutionProfileConfig()) + custom = [ + DockerJobExecutionProfile( + provider="cpu", + profile="custom-docker", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + KubernetesJobExecutionProfile( + provider="gpu", + profile="custom-k8s", + backend="kubernetes_job", + config=KubernetesJobExecutionProfileConfig(), + ), + VolcanoJobExecutionProfile( + provider="gpu_distributed", + profile="custom-volcano", + backend="volcano_job", + config=VolcanoJobExecutionProfileConfig(), + ), + SubprocessJobExecutionProfile( + profile="custom-subprocess", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(), + ), + ] + + caplog.set_level(logging.WARNING) + with patch("nmp.core.jobs.controllers.backends.config.validate_docker_available", return_value=False) as validate: + merged = merge_executor_profiles(custom, defaults, runtime=Runtime.NONE) + + assert [(p.provider, p.profile, p.backend) for p in merged] == [ + ("subprocess", "default", "subprocess"), + ("subprocess", "custom-subprocess", "subprocess"), + ] + validate.assert_called_once() + assert "Skipping executor profile cpu/custom-docker" in caplog.text + assert "Skipping executor profile gpu/custom-k8s" in caplog.text + assert "Skipping executor profile gpu_distributed/custom-volcano" in caplog.text + + +def test_merge_executor_profiles_skips_docker_when_unavailable_under_docker_runtime(caplog): + defaults = get_default_executor_profiles_for_runtime(Runtime.DOCKER, DefaultExecutionProfileConfig()) + custom = [ + DockerJobExecutionProfile( + provider="cpu", + profile="auditor", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + SubprocessJobExecutionProfile( + profile="custom-subprocess", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(), + ), + ] + + caplog.set_level(logging.WARNING) + with patch("nmp.core.jobs.controllers.backends.config.validate_docker_available", return_value=False) as validate: + merged = merge_executor_profiles(custom, defaults, runtime=Runtime.DOCKER) + + assert [(p.provider, p.profile, p.backend) for p in merged] == [ + ("subprocess", "default", "subprocess"), + ("subprocess", "custom-subprocess", "subprocess"), + ] + validate.assert_called_once() + assert "because Docker is unavailable" in caplog.text + + +def test_merge_executor_profiles_keeps_docker_executor_for_none_runtime_when_docker_is_available(): + defaults = get_default_executor_profiles_for_runtime(Runtime.NONE, DefaultExecutionProfileConfig()) + custom = [ + DockerJobExecutionProfile( + provider="cpu", + profile="workload", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ) + ] + + with patch("nmp.core.jobs.controllers.backends.config.validate_docker_available", return_value=True) as validate: + merged = merge_executor_profiles(custom, defaults, runtime=Runtime.NONE) + + assert [(p.provider, p.profile, p.backend) for p in merged] == [ + ("subprocess", "default", "subprocess"), + ("cpu", "workload", "docker"), + ] + validate.assert_called_once() + + def test_backend_registry_supports_gpu_distributed_kubernetes_job_override(): assert BackendKey("gpu_distributed", "kubernetes_job") in backend_registry @@ -563,6 +837,39 @@ def test_merge_executor_profiles_replaces_default_with_subprocess_profile(): assert merged[0].config.ttl_seconds_active == 123 +def test_merge_executor_profiles_keeps_subprocess_override_for_none_runtime(caplog): + caplog.set_level(logging.WARNING) + default_executors = get_default_executor_profiles_for_runtime(Runtime.NONE, DefaultExecutionProfileConfig()) + custom_executors = [ + DockerJobExecutionProfile( + provider="cpu", + profile="default", + backend="docker", + config=DockerJobExecutionProfileConfig(), + ), + KubernetesJobExecutionProfile( + provider="gpu", + profile="default", + backend="kubernetes_job", + config=KubernetesJobExecutionProfileConfig(namespace="custom-namespace"), + ), + SubprocessJobExecutionProfile( + profile="default", + backend="subprocess", + config=SubprocessJobExecutionProfileConfig(working_directory="/tmp/custom-subprocess-jobs"), + ), + ] + + with patch("nmp.core.jobs.controllers.backends.config.validate_docker_available", return_value=False): + merged = merge_executor_profiles(custom_executors, default_executors, runtime=Runtime.NONE) + + assert [(p.provider, p.profile, p.backend) for p in merged] == [("subprocess", "default", "subprocess")] + assert type(merged[0].config) is SubprocessJobExecutionProfileConfig + assert merged[0].config.working_directory == "/tmp/custom-subprocess-jobs" + assert "Skipping executor profile cpu/default using backend 'docker'" in caplog.text + assert "Skipping executor profile gpu/default using backend 'kubernetes_job'" in caplog.text + + def test_subprocess_execution_provider_requires_command(): assert SubprocessExecutionProvider(command=["python", "-m", "task"]).command == ["python", "-m", "task"]