112 lines
2.9 KiB
Python
112 lines
2.9 KiB
Python
from utils.ssh_helper import SshExecResult, build_ssh_command
|
|
|
|
|
|
def test_build_ssh_command_supports_jump_host_and_argv_command():
|
|
argv = build_ssh_command(
|
|
host="192.168.0.188",
|
|
user="ollama",
|
|
command=["docker", "ps"],
|
|
port=22,
|
|
key_path="/tmp/autoheal.key",
|
|
connect_timeout=10,
|
|
jump_host="192.168.0.110",
|
|
jump_user="wooo",
|
|
batch_mode=True,
|
|
server_alive_interval=15,
|
|
server_alive_count_max=3,
|
|
)
|
|
|
|
assert argv == [
|
|
"ssh",
|
|
"-p",
|
|
"22",
|
|
"-i",
|
|
"/tmp/autoheal.key",
|
|
"-o",
|
|
"StrictHostKeyChecking=no",
|
|
"-o",
|
|
"BatchMode=yes",
|
|
"-o",
|
|
"ConnectTimeout=10",
|
|
"-o",
|
|
"ServerAliveInterval=15",
|
|
"-o",
|
|
"ServerAliveCountMax=3",
|
|
"-J",
|
|
"wooo@192.168.0.110",
|
|
"ollama@192.168.0.188",
|
|
"--",
|
|
"docker",
|
|
"ps",
|
|
]
|
|
|
|
|
|
def test_build_ssh_command_preserves_direct_shell_command():
|
|
argv = build_ssh_command(
|
|
host="192.168.0.110",
|
|
user="wooo",
|
|
command="cd /home/wooo/ewoooc && git status",
|
|
port=2222,
|
|
key_path="/tmp/heal.key",
|
|
connect_timeout=5,
|
|
)
|
|
|
|
assert "--" not in argv
|
|
assert argv[-2:] == [
|
|
"wooo@192.168.0.110",
|
|
"cd /home/wooo/ewoooc && git status",
|
|
]
|
|
|
|
|
|
def test_auto_heal_ssh_wrapper_keeps_legacy_dict_contract(monkeypatch):
|
|
from services import auto_heal_service
|
|
|
|
captured = {}
|
|
|
|
def fake_run_ssh_command(**kwargs):
|
|
captured.update(kwargs)
|
|
return SshExecResult(0, "ok", "", ["ssh"])
|
|
|
|
monkeypatch.setattr(auto_heal_service, "run_ssh_command", fake_run_ssh_command)
|
|
|
|
result = auto_heal_service._ssh_exec(
|
|
jump_host="192.168.0.110",
|
|
jump_user="wooo",
|
|
target_host="192.168.0.188",
|
|
target_user="ollama",
|
|
command=["df", "-h"],
|
|
key_path="/tmp/autoheal.key",
|
|
)
|
|
|
|
assert result == {
|
|
"success": True,
|
|
"exit_code": 0,
|
|
"stdout": "ok",
|
|
"stderr": "",
|
|
"command": ["df", "-h"],
|
|
}
|
|
assert captured["jump_host"] == "192.168.0.110"
|
|
assert captured["jump_user"] == "wooo"
|
|
assert captured["batch_mode"] is True
|
|
assert captured["server_alive_interval"] == 15
|
|
|
|
|
|
def test_aider_heal_ssh_wrapper_keeps_tuple_contract(monkeypatch):
|
|
from services import aider_heal_executor
|
|
|
|
captured = {}
|
|
|
|
def fake_run_ssh_command(**kwargs):
|
|
captured.update(kwargs)
|
|
return SshExecResult(0, "ready", "", ["ssh"])
|
|
|
|
monkeypatch.setattr(aider_heal_executor, "run_ssh_command", fake_run_ssh_command)
|
|
|
|
rc, out, err = aider_heal_executor._ssh_exec("echo ready", timeout=5)
|
|
|
|
assert (rc, out, err) == (0, "ready", "")
|
|
assert captured["host"] == aider_heal_executor.HEAL_SSH_HOST
|
|
assert captured["user"] == aider_heal_executor.HEAL_SSH_USER
|
|
assert captured["command"] == "echo ready"
|
|
assert captured["command_timeout"] == 5
|