generated from coulomb/repo-seed
214 lines
8.0 KiB
Python
214 lines
8.0 KiB
Python
"""Tests for bridge.diagnostics — check_tunnel() logic."""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from bridge.diagnostics import (
|
|
_remote_port_probe_command,
|
|
check_all_tunnels,
|
|
check_tunnel,
|
|
)
|
|
from bridge.models import BridgeState, TunnelConfig
|
|
from bridge.state import StateManager
|
|
|
|
|
|
@pytest.fixture
|
|
def tcfg():
|
|
return TunnelConfig(
|
|
name="test-tunnel",
|
|
host="coulombcore.local",
|
|
remote_port=18000,
|
|
local_port=8000,
|
|
ssh_user="ubuntu",
|
|
ssh_key="~/.ssh/id_ops",
|
|
actor="adm-bernd",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def state_mgr(tmp_path):
|
|
d = tmp_path / "state"
|
|
d.mkdir()
|
|
return StateManager(state_dir=d)
|
|
|
|
|
|
class TestCheckTunnel:
|
|
def test_remote_port_probe_has_minimal_host_fallback(self):
|
|
"""Remote probe supports minimal hosts without ss/netstat."""
|
|
command = _remote_port_probe_command(18000)
|
|
assert "command -v ss" in command
|
|
assert "command -v netstat" in command
|
|
assert "/proc/net/tcp" in command
|
|
assert "/proc/net/tcp6" in command
|
|
|
|
def test_no_pid(self, tcfg, state_mgr):
|
|
"""No PID file → ssh_process='no_pid', ok=False."""
|
|
with patch("bridge.diagnostics.subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(stdout="closed\n", stderr="", returncode=1)
|
|
result = check_tunnel(tcfg, state_mgr)
|
|
assert result.ssh_process == "no_pid"
|
|
assert result.pid is None
|
|
assert result.stale_state is False
|
|
assert result.ok is False
|
|
|
|
def test_pid_dead(self, tcfg, state_mgr):
|
|
"""Dead PID + connected state → ssh_process='dead', stale_state=True."""
|
|
state_mgr.write_pid("test-tunnel", 99999)
|
|
state_mgr.write_state("test-tunnel", BridgeState.CONNECTED)
|
|
with (
|
|
patch("bridge.diagnostics._pid_alive", return_value=False),
|
|
patch("bridge.diagnostics.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(stdout="closed\n", stderr="", returncode=1)
|
|
result = check_tunnel(tcfg, state_mgr)
|
|
assert result.ssh_process == "dead"
|
|
assert result.stale_state is True
|
|
assert result.ok is False
|
|
|
|
def test_pid_alive_port_listening(self, tcfg, state_mgr):
|
|
"""Alive PID + SSH reports port listening → remote_port='listening', ok=True."""
|
|
state_mgr.write_pid("test-tunnel", 12345)
|
|
with (
|
|
patch("bridge.diagnostics._pid_alive", return_value=True),
|
|
patch("bridge.diagnostics.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(stdout="ok\n", stderr="", returncode=0)
|
|
result = check_tunnel(tcfg, state_mgr)
|
|
assert result.ssh_process == "ok"
|
|
assert result.pid == 12345
|
|
assert result.remote_port == "listening"
|
|
assert result.ok is True
|
|
|
|
def test_pid_alive_port_closed(self, tcfg, state_mgr):
|
|
"""Alive PID + SSH reports port closed → remote_port='closed', ok=False."""
|
|
state_mgr.write_pid("test-tunnel", 12345)
|
|
with (
|
|
patch("bridge.diagnostics._pid_alive", return_value=True),
|
|
patch("bridge.diagnostics.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(stdout="closed\n", stderr="", returncode=1)
|
|
result = check_tunnel(tcfg, state_mgr)
|
|
assert result.ssh_process == "ok"
|
|
assert result.remote_port == "closed"
|
|
assert result.ok is False
|
|
|
|
def test_local_direction_checks_local_port(self, tcfg, state_mgr):
|
|
"""Local tunnels verify the local listener instead of a remote -R port."""
|
|
local_cfg = TunnelConfig(
|
|
name="local-tunnel",
|
|
host="haskelseed.local",
|
|
remote_port=1234,
|
|
local_port=11234,
|
|
ssh_user="root",
|
|
ssh_key="~/.ssh/id_ops",
|
|
actor="adm-bernd",
|
|
direction="local",
|
|
)
|
|
state_mgr.write_pid("local-tunnel", 12345)
|
|
with (
|
|
patch("bridge.diagnostics._pid_alive", return_value=True),
|
|
patch("bridge.diagnostics._probe_local_port", return_value="listening"),
|
|
patch("bridge.diagnostics.subprocess.run") as mock_run,
|
|
):
|
|
result = check_tunnel(local_cfg, state_mgr)
|
|
mock_run.assert_not_called()
|
|
assert result.remote_port == "listening"
|
|
assert result.ok is True
|
|
|
|
def test_ssh_timeout(self, tcfg, state_mgr):
|
|
"""SSH probe timeout → remote_port='error:timeout'."""
|
|
state_mgr.write_pid("test-tunnel", 12345)
|
|
with (
|
|
patch("bridge.diagnostics._pid_alive", return_value=True),
|
|
patch(
|
|
"bridge.diagnostics.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd=["ssh"], timeout=10),
|
|
),
|
|
):
|
|
result = check_tunnel(tcfg, state_mgr)
|
|
assert result.remote_port == "error:timeout"
|
|
assert result.ok is False
|
|
|
|
def test_stale_state_not_flagged_when_stopped(self, tcfg, state_mgr):
|
|
"""State=stopped + no PID → stale_state is False (not connected/degraded)."""
|
|
with patch("bridge.diagnostics.subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(stdout="closed\n", stderr="", returncode=1)
|
|
result = check_tunnel(tcfg, state_mgr)
|
|
assert result.stale_state is False
|
|
|
|
def test_local_api_ok(self, tcfg, state_mgr, tmp_path):
|
|
"""With health_check configured, ok response sets local_api='ok'."""
|
|
from bridge.models import HealthCheckConfig
|
|
tcfg_with_health = TunnelConfig(
|
|
name="test-tunnel",
|
|
host="coulombcore.local",
|
|
remote_port=18000,
|
|
local_port=8000,
|
|
ssh_user="ubuntu",
|
|
ssh_key="~/.ssh/id_ops",
|
|
actor="adm-bernd",
|
|
health_check=HealthCheckConfig(url="http://127.0.0.1:8000/health"),
|
|
)
|
|
state_mgr.write_pid("test-tunnel", 12345)
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
with (
|
|
patch("bridge.diagnostics._pid_alive", return_value=True),
|
|
patch("bridge.diagnostics.subprocess.run") as mock_run,
|
|
patch("bridge.diagnostics.httpx.get", return_value=mock_resp),
|
|
):
|
|
mock_run.return_value = MagicMock(stdout="ok\n", stderr="", returncode=0)
|
|
result = check_tunnel(tcfg_with_health, state_mgr)
|
|
assert result.local_api == "ok"
|
|
assert result.latency_ms is not None
|
|
|
|
|
|
class TestCheckAllTunnels:
|
|
def test_check_all_iterates_tunnels(self, tmp_path):
|
|
"""check_all_tunnels returns one result per tunnel in cfg."""
|
|
from bridge.config import load_config
|
|
import textwrap
|
|
import os
|
|
|
|
cfg_file = tmp_path / "tunnels.yaml"
|
|
cfg_file.write_text(textwrap.dedent("""\
|
|
tunnels:
|
|
t1:
|
|
host: h1.local
|
|
remote_port: 18001
|
|
local_port: 8001
|
|
ssh_user: ubuntu
|
|
ssh_key: ~/.ssh/id_ops
|
|
actor: adm-bernd
|
|
t2:
|
|
host: h2.local
|
|
remote_port: 18002
|
|
local_port: 8002
|
|
ssh_user: ubuntu
|
|
ssh_key: ~/.ssh/id_ops
|
|
actor: adm-bernd
|
|
actors:
|
|
adm-bernd:
|
|
class: adm
|
|
description: Bernd
|
|
"""))
|
|
os.environ["BRIDGE_CONFIG"] = str(cfg_file)
|
|
try:
|
|
cfg = load_config()
|
|
finally:
|
|
del os.environ["BRIDGE_CONFIG"]
|
|
|
|
state_dir = tmp_path / "state"
|
|
state_dir.mkdir()
|
|
state_mgr = StateManager(state_dir=state_dir)
|
|
|
|
with patch("bridge.diagnostics.subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(stdout="closed\n", stderr="", returncode=1)
|
|
results = check_all_tunnels(cfg, state_mgr)
|
|
|
|
assert len(results) == 2
|
|
assert {r.tunnel for r in results} == {"t1", "t2"}
|