feat: add SSH connectivity check
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
24
src/autossh_mgr/check.py
Normal file
24
src/autossh_mgr/check.py
Normal file
@@ -0,0 +1,24 @@
|
||||
import os
|
||||
import subprocess
|
||||
from autossh_mgr.config import TunnelConfig
|
||||
|
||||
|
||||
def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
|
||||
return [
|
||||
"ssh",
|
||||
"-o", "BatchMode=yes",
|
||||
"-o", "ConnectTimeout=5",
|
||||
"-i", os.path.expanduser(tunnel.identity_file),
|
||||
"-p", str(tunnel.port),
|
||||
f"{tunnel.user}@{tunnel.host}",
|
||||
"true",
|
||||
]
|
||||
|
||||
|
||||
def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
|
||||
result = subprocess.run(
|
||||
build_ssh_check_cmd(tunnel),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
return result.returncode == 0, result.stderr.strip()
|
||||
45
tests/unit/test_check.py
Normal file
45
tests/unit/test_check.py
Normal file
@@ -0,0 +1,45 @@
|
||||
import pytest
|
||||
import subprocess
|
||||
from unittest.mock import patch, MagicMock
|
||||
from autossh_mgr.check import check_connectivity, build_ssh_check_cmd
|
||||
from autossh_mgr.config import TunnelConfig
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tunnel():
|
||||
return TunnelConfig(
|
||||
name="web", host="relay.example.com", user="deploy",
|
||||
local_port=8080, remote_port=18080,
|
||||
)
|
||||
|
||||
|
||||
def test_build_ssh_check_cmd(tunnel):
|
||||
cmd = build_ssh_check_cmd(tunnel)
|
||||
assert cmd[0] == "ssh"
|
||||
assert "-o" in cmd and "BatchMode=yes" in cmd
|
||||
assert "ConnectTimeout=5" in " ".join(cmd)
|
||||
assert f"{tunnel.user}@{tunnel.host}" in cmd
|
||||
assert "true" in cmd
|
||||
assert "-p" in cmd and str(tunnel.port) in cmd
|
||||
|
||||
|
||||
def test_check_connectivity_success(tunnel):
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
mock_result.stderr = ""
|
||||
with patch("autossh_mgr.check.subprocess.run", return_value=mock_result) as mock_run:
|
||||
success, msg = check_connectivity(tunnel)
|
||||
assert success is True
|
||||
assert msg == ""
|
||||
called_cmd = mock_run.call_args[0][0]
|
||||
assert called_cmd == build_ssh_check_cmd(tunnel)
|
||||
|
||||
|
||||
def test_check_connectivity_failure(tunnel):
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 255
|
||||
mock_result.stderr = "Connection refused"
|
||||
with patch("autossh_mgr.check.subprocess.run", return_value=mock_result):
|
||||
success, msg = check_connectivity(tunnel)
|
||||
assert success is False
|
||||
assert "Connection refused" in msg
|
||||
Reference in New Issue
Block a user