tech
6124abbdb2
check_connectivity now runs a two-layer short-circuit probe:
Layer 1: TCP connect to (local_host, local_port) to confirm the
backend service (e.g. Gitea, Next.js) is actually listening.
Layer 2: original SSH auth probe (unchanged).
This fixes the common false-positive where check returned OK while the
tunneled backend was down. Unlike the previously removed _is_port_in_use
bind check (commit 4c9cd00), this uses connect() with correct semantics:
the forward target SHOULD be listening, not free.
82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from autossh_mgr.check import (
|
|
check_connectivity,
|
|
check_local_backend,
|
|
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_local_backend_success(tunnel):
|
|
mock_sock = MagicMock()
|
|
with patch("autossh_mgr.check.socket.socket", return_value=mock_sock):
|
|
ok, msg = check_local_backend(tunnel)
|
|
assert ok is True
|
|
assert msg == ""
|
|
mock_sock.connect.assert_called_once_with(("127.0.0.1", 8080))
|
|
mock_sock.settimeout.assert_called_once_with(2)
|
|
mock_sock.close.assert_called_once()
|
|
|
|
|
|
def test_check_local_backend_failure(tunnel):
|
|
mock_sock = MagicMock()
|
|
mock_sock.connect.side_effect = ConnectionRefusedError("refused")
|
|
with patch("autossh_mgr.check.socket.socket", return_value=mock_sock):
|
|
ok, msg = check_local_backend(tunnel)
|
|
assert ok is False
|
|
assert "127.0.0.1:8080" in msg
|
|
assert "not listening" in msg
|
|
mock_sock.close.assert_called_once()
|
|
|
|
|
|
def test_check_connectivity_layer1_short_circuits(tunnel):
|
|
"""When Layer 1 fails, SSH (Layer 2) must not be invoked."""
|
|
mock_sock = MagicMock()
|
|
mock_sock.connect.side_effect = ConnectionRefusedError("refused")
|
|
with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
|
|
patch("autossh_mgr.check.subprocess.run") as mock_run:
|
|
success, msg = check_connectivity(tunnel)
|
|
assert success is False
|
|
assert "not listening" in msg
|
|
mock_run.assert_not_called()
|
|
|
|
|
|
def test_check_connectivity_success(tunnel):
|
|
mock_sock = MagicMock()
|
|
with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
|
|
patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=0, stderr="")) as mock_run:
|
|
success, msg = check_connectivity(tunnel)
|
|
assert success is True
|
|
assert msg == "reachable + backend up"
|
|
called_cmd = mock_run.call_args[0][0]
|
|
assert called_cmd == build_ssh_check_cmd(tunnel)
|
|
|
|
|
|
def test_check_connectivity_ssh_failure(tunnel):
|
|
mock_sock = MagicMock()
|
|
with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
|
|
patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=255, stderr="Connection refused")):
|
|
success, msg = check_connectivity(tunnel)
|
|
assert success is False
|
|
assert "SSH auth failed" in msg
|
|
assert "Connection refused" in msg
|