generated from coulomb/repo-seed
tegwick
10c6fdaec9
bridge restart now means blank-slate recovery: reverse tunnels run should_cleanup_tunnel and clear orphan remote listeners before reconnecting; healthy forwards are left running. Local-direction tunnels keep stop/start only. CLI and MCP report per-tunnel actions (healthy, cleaned_and_restarted, restarted, error) and exit non-zero on cleanup failure. Closes BRIDGE-WP-0005.
412 lines
14 KiB
Python
412 lines
14 KiB
Python
"""Tests for CLI commands."""
|
|
import json
|
|
import textwrap
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from bridge.cli import app
|
|
|
|
|
|
VALID_CONFIG = textwrap.dedent("""\
|
|
tunnels:
|
|
test-tunnel:
|
|
host: host.local
|
|
remote_port: 18000
|
|
local_port: 8000
|
|
ssh_user: ubuntu
|
|
ssh_key: ~/.ssh/id_ops
|
|
actor: adm-bernd
|
|
actors:
|
|
adm-bernd:
|
|
class: adm
|
|
description: Bernd
|
|
""")
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def config_file(tmp_path):
|
|
f = tmp_path / "tunnels.yaml"
|
|
f.write_text(VALID_CONFIG)
|
|
return f
|
|
|
|
|
|
@pytest.fixture
|
|
def state_dir(tmp_path):
|
|
return tmp_path / "state"
|
|
|
|
|
|
@pytest.fixture
|
|
def env(config_file, state_dir):
|
|
return {"BRIDGE_CONFIG": str(config_file), "BRIDGE_STATE_DIR": str(state_dir)}
|
|
|
|
|
|
class TestHelpCommand:
|
|
def test_app_help(self):
|
|
result = runner.invoke(app, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "bridge" in result.output.lower() or "Usage" in result.output
|
|
|
|
def test_up_help(self):
|
|
result = runner.invoke(app, ["up", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_down_help(self):
|
|
result = runner.invoke(app, ["down", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_status_help(self):
|
|
result = runner.invoke(app, ["status", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_logs_help(self):
|
|
result = runner.invoke(app, ["logs", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_restart_help(self):
|
|
result = runner.invoke(app, ["restart", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestStatusCommand:
|
|
@pytest.mark.capability("bridge_status")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_status_shows_tunnels(self, env, state_dir):
|
|
result = runner.invoke(app, ["status"], env=env)
|
|
assert result.exit_code == 0
|
|
assert "test-tunnel" in result.output
|
|
|
|
def test_status_json_flag(self, env, state_dir):
|
|
result = runner.invoke(app, ["status", "--json"], env=env)
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
assert data[0]["tunnel"] == "test-tunnel"
|
|
assert "state" in data[0]
|
|
assert "actor" in data[0]
|
|
assert "host" in data[0]
|
|
|
|
def test_status_shows_state(self, env, state_dir):
|
|
result = runner.invoke(app, ["status"], env=env)
|
|
assert result.exit_code == 0
|
|
assert "stopped" in result.output.lower()
|
|
|
|
def test_status_unknown_config_exit_1(self, tmp_path):
|
|
result = runner.invoke(app, ["status"], env={"BRIDGE_CONFIG": str(tmp_path / "no.yaml")})
|
|
assert result.exit_code == 1
|
|
|
|
|
|
class TestUpCommand:
|
|
def test_up_unknown_tunnel_exit_1(self, env):
|
|
result = runner.invoke(app, ["up", "nonexistent"], env=env)
|
|
assert result.exit_code == 1
|
|
assert "nonexistent" in result.output
|
|
|
|
@pytest.mark.capability("bridge_up")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_up_calls_manager_start(self, env, state_dir):
|
|
with patch("bridge.cli.TunnelManager") as mock_mgr_cls:
|
|
mock_mgr = MagicMock()
|
|
mock_mgr.is_running.return_value = False
|
|
mock_mgr_cls.return_value = mock_mgr
|
|
|
|
result = runner.invoke(app, ["up", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 0
|
|
mock_mgr.start.assert_called_once()
|
|
|
|
def test_up_already_running_exit_2(self, env, state_dir):
|
|
with patch("bridge.cli.TunnelManager") as mock_mgr_cls:
|
|
mock_mgr = MagicMock()
|
|
mock_mgr.is_running.return_value = True
|
|
mock_mgr_cls.return_value = mock_mgr
|
|
|
|
result = runner.invoke(app, ["up", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 2
|
|
|
|
|
|
class TestDownCommand:
|
|
def test_down_unknown_tunnel_exit_1(self, env):
|
|
result = runner.invoke(app, ["down", "nonexistent"], env=env)
|
|
assert result.exit_code == 1
|
|
|
|
@pytest.mark.capability("bridge_down")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_down_calls_manager_stop(self, env, state_dir):
|
|
with patch("bridge.cli.TunnelManager") as mock_mgr_cls:
|
|
mock_mgr = MagicMock()
|
|
mock_mgr.is_running.return_value = True
|
|
mock_mgr_cls.return_value = mock_mgr
|
|
|
|
result = runner.invoke(app, ["down", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 0
|
|
mock_mgr.stop.assert_called_once()
|
|
|
|
def test_down_not_running_exit_2(self, env, state_dir):
|
|
with patch("bridge.cli.TunnelManager") as mock_mgr_cls:
|
|
mock_mgr = MagicMock()
|
|
mock_mgr.is_running.return_value = False
|
|
mock_mgr_cls.return_value = mock_mgr
|
|
|
|
result = runner.invoke(app, ["down", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 2
|
|
|
|
|
|
class TestLogsCommand:
|
|
def test_logs_unknown_tunnel_exit_1(self, env):
|
|
result = runner.invoke(app, ["logs", "nonexistent"], env=env)
|
|
assert result.exit_code == 1
|
|
|
|
def test_logs_no_log_file_shows_empty(self, env, state_dir):
|
|
result = runner.invoke(app, ["logs", "test-tunnel"], env=env)
|
|
assert result.exit_code == 0
|
|
|
|
@pytest.mark.capability("bridge_logs")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_logs_shows_events(self, env, state_dir):
|
|
import json as _json
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = state_dir / "test-tunnel.log"
|
|
log_file.write_text(
|
|
_json.dumps({
|
|
"timestamp": "2026-01-01T00:00:00+00:00",
|
|
"tunnel": "test-tunnel",
|
|
"actor": "operator.bernd",
|
|
"actor_class": "human",
|
|
"event": "bridge_started",
|
|
}) + "\n"
|
|
)
|
|
result = runner.invoke(app, ["logs", "test-tunnel"], env=env)
|
|
assert result.exit_code == 0
|
|
assert "bridge_started" in result.output
|
|
|
|
|
|
class TestCheckCommand:
|
|
def test_check_help(self):
|
|
result = runner.invoke(app, ["check", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
@pytest.mark.capability("bridge_check")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_check_all_pass(self, env):
|
|
from bridge.diagnostics import TunnelCheckResult
|
|
ok_result = TunnelCheckResult(
|
|
tunnel="test-tunnel",
|
|
ssh_process="ok",
|
|
pid=12345,
|
|
remote_port="listening",
|
|
local_api=None,
|
|
latency_ms=None,
|
|
stale_state=False,
|
|
)
|
|
with patch("bridge.cli.check_all_tunnels", return_value=[ok_result]):
|
|
result = runner.invoke(app, ["check"], env=env)
|
|
assert result.exit_code == 0
|
|
|
|
def test_check_any_fail(self, env):
|
|
from bridge.diagnostics import TunnelCheckResult
|
|
fail_result = TunnelCheckResult(
|
|
tunnel="test-tunnel",
|
|
ssh_process="dead",
|
|
pid=None,
|
|
remote_port="closed",
|
|
local_api=None,
|
|
latency_ms=None,
|
|
stale_state=True,
|
|
)
|
|
with patch("bridge.cli.check_all_tunnels", return_value=[fail_result]):
|
|
result = runner.invoke(app, ["check"], env=env)
|
|
assert result.exit_code == 1
|
|
|
|
def test_check_json_flag(self, env):
|
|
from bridge.diagnostics import TunnelCheckResult
|
|
ok_result = TunnelCheckResult(
|
|
tunnel="test-tunnel",
|
|
ssh_process="ok",
|
|
pid=12345,
|
|
remote_port="listening",
|
|
local_api=None,
|
|
latency_ms=None,
|
|
stale_state=False,
|
|
)
|
|
with patch("bridge.cli.check_all_tunnels", return_value=[ok_result]):
|
|
result = runner.invoke(app, ["check", "--json"], env=env)
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
assert data[0]["ok"] is True
|
|
assert data[0]["tunnel"] == "test-tunnel"
|
|
assert data[0]["ssh_process"] == "ok"
|
|
|
|
def test_check_specific_tunnel(self, env):
|
|
from bridge.diagnostics import TunnelCheckResult
|
|
ok_result = TunnelCheckResult(
|
|
tunnel="test-tunnel",
|
|
ssh_process="ok",
|
|
pid=12345,
|
|
remote_port="listening",
|
|
local_api=None,
|
|
latency_ms=None,
|
|
stale_state=False,
|
|
)
|
|
with patch("bridge.cli.check_tunnel", return_value=ok_result):
|
|
result = runner.invoke(app, ["check", "test-tunnel"], env=env)
|
|
assert result.exit_code == 0
|
|
|
|
def test_check_unknown_tunnel_exit_1(self, env):
|
|
result = runner.invoke(app, ["check", "nonexistent"], env=env)
|
|
assert result.exit_code == 1
|
|
|
|
|
|
REVERSE_CONFIG = VALID_CONFIG
|
|
|
|
LOCAL_TUNNEL_CONFIG = textwrap.dedent("""\
|
|
tunnels:
|
|
k3s-api:
|
|
host: host.local
|
|
remote_port: 6443
|
|
local_port: 6443
|
|
ssh_user: ubuntu
|
|
ssh_key: ~/.ssh/id_ops
|
|
actor: adm-bernd
|
|
direction: local
|
|
actors:
|
|
adm-bernd:
|
|
class: adm
|
|
description: Bernd
|
|
""")
|
|
|
|
|
|
class TestRestartCommand:
|
|
def test_restart_unknown_tunnel_exit_1(self, env):
|
|
result = runner.invoke(app, ["restart", "nonexistent"], env=env)
|
|
assert result.exit_code == 1
|
|
|
|
def test_restart_help_mentions_remote_cleanup(self):
|
|
result = runner.invoke(app, ["restart", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "stale-forward" in result.output.lower() or "remote" in result.output.lower()
|
|
|
|
@pytest.mark.capability("bridge_restart")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_restart_reverse_tunnel_delegates_to_cleanup(self, env):
|
|
from bridge.cleanup import CleanupAction
|
|
|
|
with patch("bridge.cli.restart_tunnel") as mock_restart:
|
|
mock_restart.return_value = CleanupAction(
|
|
"test-tunnel", "healthy", "remote forward healthy"
|
|
)
|
|
result = runner.invoke(app, ["restart", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 0
|
|
mock_restart.assert_called_once()
|
|
assert "test-tunnel: healthy" in result.output
|
|
|
|
def test_restart_reverse_tunnel_reports_cleaned_and_restarted(self, env):
|
|
from bridge.cleanup import CleanupAction
|
|
|
|
with patch("bridge.cli.restart_tunnel") as mock_restart:
|
|
mock_restart.return_value = CleanupAction(
|
|
"test-tunnel",
|
|
"cleaned_and_restarted",
|
|
"stale forward; restarted tunnel; cleared",
|
|
)
|
|
result = runner.invoke(app, ["restart", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 0
|
|
assert "cleaned_and_restarted" in result.output
|
|
|
|
def test_restart_reverse_tunnel_error_exit_1(self, env):
|
|
from bridge.cleanup import CleanupAction
|
|
|
|
with patch("bridge.cli.restart_tunnel") as mock_restart:
|
|
mock_restart.return_value = CleanupAction(
|
|
"test-tunnel", "error", "cleanup failed: still_listening"
|
|
)
|
|
result = runner.invoke(app, ["restart", "test-tunnel"], env=env)
|
|
|
|
assert result.exit_code == 1
|
|
assert "error" in result.output
|
|
|
|
def test_restart_local_tunnel_uses_stop_start(self, tmp_path, state_dir):
|
|
config_file = tmp_path / "tunnels.yaml"
|
|
config_file.write_text(LOCAL_TUNNEL_CONFIG)
|
|
env = {
|
|
"BRIDGE_CONFIG": str(config_file),
|
|
"BRIDGE_STATE_DIR": str(state_dir),
|
|
}
|
|
|
|
with patch("bridge.cleanup.TunnelManager") as mock_mgr_cls:
|
|
mock_mgr = MagicMock()
|
|
mock_mgr_cls.return_value = mock_mgr
|
|
call_order = []
|
|
mock_mgr.stop.side_effect = lambda: call_order.append("stop")
|
|
mock_mgr.start.side_effect = lambda: call_order.append("start")
|
|
|
|
result = runner.invoke(app, ["restart", "k3s-api"], env=env)
|
|
|
|
assert result.exit_code == 0
|
|
assert call_order == ["stop", "start"]
|
|
assert "k3s-api: restarted" in result.output
|
|
|
|
|
|
class TestCertStatusCommand:
|
|
@pytest.mark.capability("bridge_cert_status")
|
|
@pytest.mark.access_mode("cli")
|
|
def test_cert_status_no_cert_shows_static_key(self, env, state_dir):
|
|
result = runner.invoke(app, ["cert-status"], env=env)
|
|
assert result.exit_code == 0
|
|
assert "static-key" in result.output
|
|
|
|
def test_cert_status_json_no_cert(self, env, state_dir):
|
|
result = runner.invoke(app, ["cert-status", "--json"], env=env)
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
assert data[0]["mode"] == "static-key"
|
|
|
|
def test_cert_status_exit_1_on_expired(self, env, state_dir, tmp_path):
|
|
# Write a fake cert file in state dir; mock ssh-keygen to report expired
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
cert_file = state_dir / "test-tunnel-cert.pub"
|
|
cert_file.write_text("fake cert")
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(
|
|
stdout=(
|
|
"test-tunnel-cert.pub:\n"
|
|
" Key ID: \"agt-test\"\n"
|
|
" Valid: from 2026-01-01T00:00:00 to 2026-01-02T00:00:00\n"
|
|
),
|
|
returncode=0,
|
|
)
|
|
result = runner.invoke(app, ["cert-status"], env=env)
|
|
assert result.exit_code == 1
|
|
assert "EXPIRED" in result.output
|
|
|
|
def test_cert_status_json_with_cert(self, env, state_dir):
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
cert_file = state_dir / "test-tunnel-cert.pub"
|
|
cert_file.write_text("fake cert")
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(
|
|
stdout=(
|
|
"test-tunnel-cert.pub:\n"
|
|
" Key ID: \"agt-test\"\n"
|
|
" Valid: from 2030-01-01T00:00:00 to 2030-01-02T00:00:00\n"
|
|
),
|
|
returncode=0,
|
|
)
|
|
result = runner.invoke(app, ["cert-status", "--json"], env=env)
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
assert data[0]["mode"] == "cert"
|
|
assert data[0]["key_id"] == "agt-test"
|
|
assert data[0]["expired"] is False
|