test_ssh_environment.py
1 """Tests for the SSH remote execution environment backend.""" 2 3 import json 4 import os 5 import subprocess 6 from unittest.mock import MagicMock 7 8 import pytest 9 10 from tools.environments.ssh import SSHEnvironment 11 from tools.environments import ssh as ssh_env 12 13 _SSH_HOST = os.getenv("TERMINAL_SSH_HOST", "") 14 _SSH_USER = os.getenv("TERMINAL_SSH_USER", "") 15 _SSH_PORT = int(os.getenv("TERMINAL_SSH_PORT", "22")) 16 _SSH_KEY = os.getenv("TERMINAL_SSH_KEY", "") 17 18 _has_ssh = bool(_SSH_HOST and _SSH_USER) 19 20 requires_ssh = pytest.mark.skipif( 21 not _has_ssh, 22 reason="TERMINAL_SSH_HOST / TERMINAL_SSH_USER not set", 23 ) 24 25 26 def _run(command, task_id="ssh_test", **kwargs): 27 from tools.terminal_tool import terminal_tool 28 return json.loads(terminal_tool(command, task_id=task_id, **kwargs)) 29 30 31 def _cleanup(task_id="ssh_test"): 32 from tools.terminal_tool import cleanup_vm 33 cleanup_vm(task_id) 34 35 36 class TestBuildSSHCommand: 37 38 @pytest.fixture(autouse=True) 39 def _mock_connection(self, monkeypatch): 40 monkeypatch.setattr("tools.environments.ssh.subprocess.run", 41 lambda *a, **k: subprocess.CompletedProcess([], 0)) 42 monkeypatch.setattr("tools.environments.ssh.subprocess.Popen", 43 lambda *a, **k: MagicMock(stdout=iter([]), 44 stderr=iter([]), 45 stdin=MagicMock())) 46 monkeypatch.setattr("tools.environments.base.time.sleep", lambda _: None) 47 48 def test_base_flags(self): 49 env = SSHEnvironment(host="h", user="u") 50 cmd = " ".join(env._build_ssh_command()) 51 for flag in ("ControlMaster=auto", "ControlPersist=300", 52 "BatchMode=yes", "StrictHostKeyChecking=accept-new"): 53 assert flag in cmd 54 55 def test_custom_port(self): 56 env = SSHEnvironment(host="h", user="u", port=2222) 57 cmd = env._build_ssh_command() 58 assert "-p" in cmd and "2222" in cmd 59 60 def test_key_path(self): 61 env = SSHEnvironment(host="h", user="u", key_path="/k") 62 cmd = env._build_ssh_command() 63 assert "-i" in cmd and "/k" in cmd 64 65 def test_user_host_suffix(self): 66 env = SSHEnvironment(host="h", user="u") 67 assert env._build_ssh_command()[-1] == "u@h" 68 69 70 class TestControlSocketPath: 71 """Regression tests for issue #11840. 72 73 macOS caps Unix domain socket paths at 104 bytes (sun_path). SSH 74 appends a 16-byte random suffix to the control socket path when 75 operating in ControlMaster mode. An IPv6 host embedded in the 76 filename plus the deeply-nested macOS $TMPDIR easily blows past 77 the limit, causing every tool call to fail immediately. 78 """ 79 80 @pytest.fixture(autouse=True) 81 def _mock_connection(self, monkeypatch): 82 monkeypatch.setattr("tools.environments.ssh.subprocess.run", 83 lambda *a, **k: subprocess.CompletedProcess([], 0)) 84 monkeypatch.setattr("tools.environments.ssh.subprocess.Popen", 85 lambda *a, **k: MagicMock(stdout=iter([]), 86 stderr=iter([]), 87 stdin=MagicMock())) 88 monkeypatch.setattr("tools.environments.base.time.sleep", lambda _: None) 89 90 # SSH appends ``.XXXXXXXXXXXXXXXX`` (17 bytes) to the ControlPath in 91 # ControlMaster mode; the macOS sun_path field is 104 bytes including 92 # the NUL terminator, so the usable path length is 103 bytes. 93 _SSH_CONTROLMASTER_SUFFIX = 17 94 _MAX_SUN_PATH = 103 95 96 def test_fits_under_macos_socket_limit_with_ipv6_host(self, monkeypatch): 97 """A realistic macOS $TMPDIR + IPv6 host must still produce a 98 control socket path that fits once SSH appends its ControlMaster 99 suffix (see issue #11840).""" 100 # Simulate the macOS $TMPDIR shape from the issue traceback — 101 # 48 bytes, the typical length of ``/var/folders/XX/YYYYYYYYY/T``. 102 fake_tmp = "/var/folders/2t/wbkw5yb158jc3zhswgl7tz9c0000gn/T" 103 monkeypatch.setattr("tools.environments.ssh.tempfile.gettempdir", 104 lambda: fake_tmp) 105 # The simulated path doesn't exist on the test host — skip the 106 # real mkdir so __init__ can proceed. 107 from pathlib import Path as _Path 108 monkeypatch.setattr(_Path, "mkdir", lambda *a, **k: None) 109 110 env = SSHEnvironment( 111 host="9373:9b91:4480:558d:708e:e601:24e8:d8d0", 112 user="hermes", 113 port=22, 114 ) 115 116 total_len = len(str(env.control_socket)) + self._SSH_CONTROLMASTER_SUFFIX 117 assert total_len <= self._MAX_SUN_PATH, ( 118 f"control socket path would exceed the {self._MAX_SUN_PATH}-byte " 119 f"Unix domain socket limit once SSH appends its 16-byte suffix: " 120 f"{env.control_socket} (+{self._SSH_CONTROLMASTER_SUFFIX} = {total_len})" 121 ) 122 123 def test_path_is_deterministic_across_instances(self): 124 """Same (user, host, port) must yield the same control socket so 125 ControlMaster reuse works across reconnects.""" 126 first = SSHEnvironment(host="example.com", user="alice", port=2222) 127 second = SSHEnvironment(host="example.com", user="alice", port=2222) 128 assert first.control_socket == second.control_socket 129 130 def test_path_differs_for_different_targets(self): 131 """Different (user, host, port) triples must produce different paths.""" 132 base = SSHEnvironment(host="h", user="u", port=22).control_socket 133 assert SSHEnvironment(host="h", user="u", port=23).control_socket != base 134 assert SSHEnvironment(host="h", user="v", port=22).control_socket != base 135 assert SSHEnvironment(host="g", user="u", port=22).control_socket != base 136 137 138 class TestTerminalToolConfig: 139 def test_ssh_persistent_default_true(self, monkeypatch): 140 """SSH persistent defaults to True (via TERMINAL_PERSISTENT_SHELL).""" 141 monkeypatch.delenv("TERMINAL_SSH_PERSISTENT", raising=False) 142 monkeypatch.delenv("TERMINAL_PERSISTENT_SHELL", raising=False) 143 from tools.terminal_tool import _get_env_config 144 assert _get_env_config()["ssh_persistent"] is True 145 146 def test_ssh_persistent_explicit_false(self, monkeypatch): 147 """Per-backend env var overrides the global default.""" 148 monkeypatch.setenv("TERMINAL_SSH_PERSISTENT", "false") 149 from tools.terminal_tool import _get_env_config 150 assert _get_env_config()["ssh_persistent"] is False 151 152 def test_ssh_persistent_explicit_true(self, monkeypatch): 153 monkeypatch.setenv("TERMINAL_SSH_PERSISTENT", "true") 154 from tools.terminal_tool import _get_env_config 155 assert _get_env_config()["ssh_persistent"] is True 156 157 def test_ssh_persistent_respects_config(self, monkeypatch): 158 """TERMINAL_PERSISTENT_SHELL=false disables SSH persistent by default.""" 159 monkeypatch.delenv("TERMINAL_SSH_PERSISTENT", raising=False) 160 monkeypatch.setenv("TERMINAL_PERSISTENT_SHELL", "false") 161 from tools.terminal_tool import _get_env_config 162 assert _get_env_config()["ssh_persistent"] is False 163 164 165 class TestSSHPreflight: 166 def test_ensure_ssh_available_raises_clear_error_when_missing(self, monkeypatch): 167 monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: None) 168 169 with pytest.raises(RuntimeError, match="SSH is not installed or not in PATH"): 170 ssh_env._ensure_ssh_available() 171 172 def test_ssh_environment_checks_availability_before_connect(self, monkeypatch): 173 monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: None) 174 monkeypatch.setattr( 175 ssh_env.SSHEnvironment, 176 "_establish_connection", 177 lambda self: pytest.fail("_establish_connection should not run when ssh is missing"), 178 ) 179 180 with pytest.raises(RuntimeError, match="openssh-client"): 181 ssh_env.SSHEnvironment(host="example.com", user="alice") 182 183 def test_ssh_environment_connects_when_ssh_exists(self, monkeypatch): 184 called = {"count": 0} 185 186 monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: "/usr/bin/ssh") 187 188 def _fake_establish(self): 189 called["count"] += 1 190 191 monkeypatch.setattr(ssh_env.SSHEnvironment, "_establish_connection", _fake_establish) 192 monkeypatch.setattr(ssh_env.SSHEnvironment, "_detect_remote_home", lambda self: "/home/alice") 193 monkeypatch.setattr(ssh_env.SSHEnvironment, "_ensure_remote_dirs", lambda self: None) 194 monkeypatch.setattr(ssh_env.SSHEnvironment, "init_session", lambda self: None) 195 monkeypatch.setattr(ssh_env, "FileSyncManager", lambda **kw: type("M", (), {"sync": lambda self, **k: None})()) 196 197 env = ssh_env.SSHEnvironment(host="example.com", user="alice") 198 199 assert called["count"] == 1 200 assert env.host == "example.com" 201 assert env.user == "alice" 202 203 204 def _setup_ssh_env(monkeypatch, persistent: bool): 205 monkeypatch.setenv("TERMINAL_ENV", "ssh") 206 monkeypatch.setenv("TERMINAL_SSH_HOST", _SSH_HOST) 207 monkeypatch.setenv("TERMINAL_SSH_USER", _SSH_USER) 208 monkeypatch.setenv("TERMINAL_SSH_PERSISTENT", "true" if persistent else "false") 209 if _SSH_PORT != 22: 210 monkeypatch.setenv("TERMINAL_SSH_PORT", str(_SSH_PORT)) 211 if _SSH_KEY: 212 monkeypatch.setenv("TERMINAL_SSH_KEY", _SSH_KEY) 213 214 215 @requires_ssh 216 class TestOneShotSSH: 217 218 @pytest.fixture(autouse=True) 219 def _setup(self, monkeypatch): 220 _setup_ssh_env(monkeypatch, persistent=False) 221 yield 222 _cleanup() 223 224 def test_echo(self): 225 r = _run("echo hello") 226 assert r["exit_code"] == 0 227 assert "hello" in r["output"] 228 229 def test_exit_code(self): 230 r = _run("exit 42") 231 assert r["exit_code"] == 42 232 233 def test_state_does_not_persist(self): 234 _run("export HERMES_ONESHOT_TEST=yes") 235 r = _run("echo $HERMES_ONESHOT_TEST") 236 assert r["output"].strip() == "" 237 238 239 @requires_ssh 240 class TestPersistentSSH: 241 242 @pytest.fixture(autouse=True) 243 def _setup(self, monkeypatch): 244 _setup_ssh_env(monkeypatch, persistent=True) 245 yield 246 _cleanup() 247 248 def test_echo(self): 249 r = _run("echo hello-persistent") 250 assert r["exit_code"] == 0 251 assert "hello-persistent" in r["output"] 252 253 def test_env_var_persists(self): 254 _run("export HERMES_PERSIST_TEST=works") 255 r = _run("echo $HERMES_PERSIST_TEST") 256 assert r["output"].strip() == "works" 257 258 def test_cwd_persists(self): 259 _run("cd /tmp") 260 r = _run("pwd") 261 assert r["output"].strip() == "/tmp" 262 263 def test_exit_code(self): 264 r = _run("(exit 42)") 265 assert r["exit_code"] == 42 266 267 def test_stderr(self): 268 r = _run("echo oops >&2") 269 assert r["exit_code"] == 0 270 assert "oops" in r["output"] 271 272 def test_multiline_output(self): 273 r = _run("echo a; echo b; echo c") 274 lines = r["output"].strip().splitlines() 275 assert lines == ["a", "b", "c"] 276 277 def test_timeout_then_recovery(self): 278 r = _run("sleep 999", timeout=2) 279 assert r["exit_code"] == 124 280 r = _run("echo alive") 281 assert r["exit_code"] == 0 282 assert "alive" in r["output"] 283 284 def test_large_output(self): 285 r = _run("seq 1 1000") 286 assert r["exit_code"] == 0 287 lines = r["output"].strip().splitlines() 288 assert len(lines) == 1000 289 assert lines[0] == "1" 290 assert lines[-1] == "1000"