224 lines
9.3 KiB
Python
224 lines
9.3 KiB
Python
"""Remote terminal backend graceful degradation (terminal.degraded_mode).
|
|
|
|
Connection-class infrastructure failures (SSH unreachable, Docker daemon
|
|
down) must come back to the model as a structured ``status: "degraded"``
|
|
tool result with a reason and a retry hint — not as a raised traceback
|
|
blob. Command failures (nonzero exit codes) are NOT infrastructure
|
|
failures and must stay untouched. ``terminal.degraded_mode: fail``
|
|
preserves the historical raise/traceback behavior for anyone relying
|
|
on it.
|
|
|
|
Inspired by: Claude Cowork degraded-backend behavior (idea-level,
|
|
docs-only evidence).
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from tools.environments.base import EnvironmentConnectionError
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_env(tmp_path, monkeypatch):
|
|
"""Isolated HERMES_HOME + a clean environment cache for terminal_tool."""
|
|
import tools.terminal_tool as tt
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
# The one-shot config bridge would overwrite our TERMINAL_* test vars
|
|
# from the developer's real config.yaml; mark it as already attempted.
|
|
monkeypatch.setattr(tt, "_terminal_config_bridge_attempted", True)
|
|
|
|
def _clear():
|
|
with tt._env_lock:
|
|
tt._active_environments.clear()
|
|
tt._last_activity.clear()
|
|
|
|
_clear()
|
|
yield tt
|
|
_clear()
|
|
|
|
|
|
def _mock_ssh_unreachable(monkeypatch, stderr="ssh: connect to host unreachable.invalid port 22: Connection refused"):
|
|
"""Make every ssh subprocess in the ssh backend fail like a dead host."""
|
|
monkeypatch.setattr("tools.environments.ssh.shutil.which", lambda _x: "/usr/bin/ssh")
|
|
monkeypatch.setattr(
|
|
"tools.environments.ssh.subprocess.run",
|
|
lambda *a, **k: subprocess.CompletedProcess([], 255, stdout="", stderr=stderr),
|
|
)
|
|
|
|
|
|
def _ssh_backend_env(monkeypatch):
|
|
monkeypatch.setenv("TERMINAL_ENV", "ssh")
|
|
monkeypatch.setenv("TERMINAL_SSH_HOST", "unreachable.invalid")
|
|
monkeypatch.setenv("TERMINAL_SSH_USER", "nobody")
|
|
monkeypatch.delenv("TERMINAL_SSH_PORT", raising=False)
|
|
monkeypatch.delenv("TERMINAL_SSH_KEY", raising=False)
|
|
|
|
|
|
class TestExceptionClassification:
|
|
"""Backends raise EnvironmentConnectionError for connection-class failures."""
|
|
|
|
def test_ssh_connect_refused_raises_connection_error(self, monkeypatch):
|
|
from tools.environments.ssh import SSHEnvironment
|
|
|
|
_mock_ssh_unreachable(monkeypatch)
|
|
with pytest.raises(EnvironmentConnectionError):
|
|
SSHEnvironment(host="unreachable.invalid", user="nobody")
|
|
|
|
def test_ssh_connect_timeout_raises_connection_error(self, monkeypatch):
|
|
from tools.environments.ssh import SSHEnvironment
|
|
|
|
monkeypatch.setattr("tools.environments.ssh.shutil.which", lambda _x: "/usr/bin/ssh")
|
|
|
|
def _timeout(*a, **k):
|
|
raise subprocess.TimeoutExpired(cmd="ssh", timeout=15)
|
|
|
|
monkeypatch.setattr("tools.environments.ssh.subprocess.run", _timeout)
|
|
with pytest.raises(EnvironmentConnectionError):
|
|
SSHEnvironment(host="unreachable.invalid", user="nobody")
|
|
|
|
def test_docker_missing_executable_raises_connection_error(self, monkeypatch):
|
|
from tools.environments import docker as docker_env
|
|
|
|
monkeypatch.setattr(docker_env, "find_docker", lambda: None)
|
|
with pytest.raises(EnvironmentConnectionError):
|
|
docker_env._ensure_docker_available()
|
|
|
|
def test_docker_daemon_timeout_raises_connection_error(self, monkeypatch):
|
|
from tools.environments import docker as docker_env
|
|
|
|
monkeypatch.setattr(docker_env, "find_docker", lambda: "/usr/bin/docker")
|
|
|
|
def _timeout(*a, **k):
|
|
raise subprocess.TimeoutExpired(cmd="docker version", timeout=5)
|
|
|
|
monkeypatch.setattr(docker_env.subprocess, "run", _timeout)
|
|
with pytest.raises(EnvironmentConnectionError):
|
|
docker_env._ensure_docker_available()
|
|
|
|
def test_connection_error_is_a_runtime_error(self):
|
|
# Existing catchers of RuntimeError must keep working unchanged.
|
|
assert issubclass(EnvironmentConnectionError, RuntimeError)
|
|
err = EnvironmentConnectionError("boom")
|
|
assert err.reason == "boom"
|
|
assert err.retry_hint # non-empty default hint
|
|
|
|
|
|
class TestDegradedToolResult:
|
|
"""terminal_tool returns structured degraded results in warn mode."""
|
|
|
|
def test_ssh_unreachable_returns_degraded_result(self, isolated_env, monkeypatch):
|
|
_ssh_backend_env(monkeypatch)
|
|
_mock_ssh_unreachable(monkeypatch)
|
|
monkeypatch.delenv("TERMINAL_DEGRADED_MODE", raising=False)
|
|
|
|
r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-ssh"))
|
|
assert r["status"] == "degraded"
|
|
assert r["exit_code"] == -1
|
|
assert "reason" in r and r["reason"]
|
|
assert "retry_hint" in r and r["retry_hint"]
|
|
assert "traceback" not in r
|
|
|
|
def test_docker_daemon_down_returns_degraded_result(self, isolated_env, monkeypatch):
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
monkeypatch.delenv("TERMINAL_DEGRADED_MODE", raising=False)
|
|
monkeypatch.setattr(isolated_env, "_maybe_reap_docker_orphans", lambda _cc: None)
|
|
monkeypatch.setattr("tools.environments.docker.find_docker", lambda: None)
|
|
|
|
r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-docker"))
|
|
assert r["status"] == "degraded"
|
|
assert "reason" in r and r["reason"]
|
|
assert "retry_hint" in r and r["retry_hint"]
|
|
assert "traceback" not in r
|
|
|
|
def test_degraded_env_is_not_cached(self, isolated_env, monkeypatch):
|
|
"""A degraded backend must not be cached — a later call must retry."""
|
|
_ssh_backend_env(monkeypatch)
|
|
_mock_ssh_unreachable(monkeypatch)
|
|
|
|
r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-cache"))
|
|
assert r["status"] == "degraded"
|
|
with isolated_env._env_lock:
|
|
assert not isolated_env._active_environments
|
|
|
|
def test_recovery_after_degraded(self, isolated_env, monkeypatch):
|
|
"""When the backend comes back, the next call just works."""
|
|
import shutil as real_shutil
|
|
|
|
real_run = subprocess.run
|
|
real_which = real_shutil.which
|
|
_ssh_backend_env(monkeypatch)
|
|
_mock_ssh_unreachable(monkeypatch)
|
|
r1 = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-recover"))
|
|
assert r1["status"] == "degraded"
|
|
|
|
# Backend "recovers" — restore the real subprocess machinery (the
|
|
# ssh-module patch hits the shared subprocess/shutil modules) and
|
|
# switch to a reachable backend; the tool path must not be poisoned.
|
|
monkeypatch.setattr(subprocess, "run", real_run)
|
|
monkeypatch.setattr(real_shutil, "which", real_which)
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
r2 = json.loads(isolated_env.terminal_tool("echo back", task_id="t-degraded-recover"))
|
|
assert r2["exit_code"] == 0
|
|
assert "back" in r2["output"]
|
|
|
|
|
|
class TestNonInfrastructureFailuresUntouched:
|
|
def test_nonzero_exit_is_not_degraded(self, isolated_env, monkeypatch):
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
r = json.loads(isolated_env.terminal_tool("exit 3", task_id="t-degraded-exit3"))
|
|
assert r["exit_code"] == 3
|
|
assert r.get("status") != "degraded"
|
|
|
|
def test_command_not_found_is_not_degraded(self, isolated_env, monkeypatch):
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
r = json.loads(isolated_env.terminal_tool(
|
|
"definitely_not_a_real_command_zzz_42", task_id="t-degraded-notfound"))
|
|
assert r["exit_code"] != 0
|
|
assert r.get("status") != "degraded"
|
|
|
|
|
|
class TestFailModePreservesRaiseBehavior:
|
|
def test_fail_mode_returns_error_with_traceback(self, isolated_env, monkeypatch):
|
|
_ssh_backend_env(monkeypatch)
|
|
_mock_ssh_unreachable(monkeypatch)
|
|
monkeypatch.setenv("TERMINAL_DEGRADED_MODE", "fail")
|
|
|
|
r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-fail"))
|
|
assert r["status"] == "error"
|
|
assert "traceback" in r
|
|
assert "SSH connection failed" in r["error"]
|
|
|
|
def test_invalid_mode_falls_back_to_warn(self, isolated_env, monkeypatch):
|
|
_ssh_backend_env(monkeypatch)
|
|
_mock_ssh_unreachable(monkeypatch)
|
|
monkeypatch.setenv("TERMINAL_DEGRADED_MODE", "bogus-value")
|
|
|
|
r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-bogus"))
|
|
assert r["status"] == "degraded"
|
|
|
|
|
|
class TestConfigBridging:
|
|
def test_degraded_mode_is_bridged_everywhere(self):
|
|
"""terminal.degraded_mode must ride every config->env bridge path,
|
|
same four-site invariant as the docker_* keys."""
|
|
from tests.tools.test_terminal_config_env_sync import (
|
|
_cli_env_map_keys,
|
|
_gateway_env_map_keys,
|
|
_save_config_env_sync_keys,
|
|
_terminal_tool_env_var_names,
|
|
)
|
|
|
|
assert "degraded_mode" in _cli_env_map_keys()
|
|
assert "degraded_mode" in _gateway_env_map_keys()
|
|
assert "degraded_mode" in _save_config_env_sync_keys()
|
|
assert "TERMINAL_DEGRADED_MODE" in _terminal_tool_env_var_names()
|
|
|
|
def test_default_config_carries_degraded_mode(self):
|
|
from hermes_cli.config_defaults import DEFAULT_CONFIG
|
|
|
|
assert DEFAULT_CONFIG["terminal"].get("degraded_mode") == "warn"
|