test: add integration tests for full tunnel lifecycle

Exercises the complete PID file lifecycle end-to-end using a real subprocess
with a fake autossh (sleep 60 script), covering start, stop, status, stale PID
cleanup, restart, and idempotency scenarios.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-21 13:01:40 +08:00
parent c365e668a5
commit 4b0e81c556

View File

@@ -0,0 +1,131 @@
import os
import signal
import time
import pytest
from click.testing import CliRunner
from autossh_mgr.cli import cli
from autossh_mgr.config import save_tunnels, TunnelConfig, ensure_dirs
from autossh_mgr.process import read_pid_file, is_process_alive
@pytest.fixture
def fake_autossh(tmp_path, monkeypatch):
"""Replace autossh with a long-running dummy process."""
fake = tmp_path / "autossh"
fake.write_text("#!/bin/sh\nexec sleep 60\n")
fake.chmod(0o755)
monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
return fake
@pytest.fixture
def config_dir(tmp_path):
ensure_dirs(tmp_path)
return tmp_path
@pytest.fixture(autouse=True)
def set_config_dir(config_dir, monkeypatch):
monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
@pytest.fixture
def tunnel(config_dir):
t = TunnelConfig(
name="test-tunnel",
host="relay.example.com",
user="deploy",
local_port=19999,
remote_port=29999,
)
save_tunnels(config_dir, [t])
return t
@pytest.fixture
def runner():
return CliRunner()
def test_start_writes_pid_file(runner, config_dir, tunnel, fake_autossh):
result = runner.invoke(cli, ["start", "test-tunnel"])
assert result.exit_code == 0, result.output
pid_data = read_pid_file(config_dir, "test-tunnel")
assert pid_data is not None
pid, started_at = pid_data
assert is_process_alive(pid)
# cleanup
os.kill(pid, signal.SIGKILL)
def test_stop_kills_process(runner, config_dir, tunnel, fake_autossh):
runner.invoke(cli, ["start", "test-tunnel"])
pid_data = read_pid_file(config_dir, "test-tunnel")
assert pid_data is not None
pid = pid_data[0]
assert is_process_alive(pid)
result = runner.invoke(cli, ["stop", "test-tunnel"])
assert result.exit_code == 0
time.sleep(0.2)
assert not is_process_alive(pid)
assert read_pid_file(config_dir, "test-tunnel") is None
def test_status_shows_running(runner, config_dir, tunnel, fake_autossh):
runner.invoke(cli, ["start", "test-tunnel"])
result = runner.invoke(cli, ["status", "test-tunnel"])
assert result.exit_code == 0
assert "running" in result.output
pid_data = read_pid_file(config_dir, "test-tunnel")
if pid_data:
os.kill(pid_data[0], signal.SIGKILL)
def test_stale_pid_cleaned_on_start(runner, config_dir, tunnel, fake_autossh):
from autossh_mgr.process import write_pid_file
from datetime import datetime, timezone
write_pid_file(config_dir, "test-tunnel", 9999999, datetime.now(timezone.utc))
result = runner.invoke(cli, ["start", "test-tunnel"])
assert result.exit_code == 0, result.output
assert "stale" in result.output.lower()
pid_data = read_pid_file(config_dir, "test-tunnel")
assert pid_data is not None
assert pid_data[0] != 9999999
os.kill(pid_data[0], signal.SIGKILL)
def test_restart(runner, config_dir, tunnel, fake_autossh):
runner.invoke(cli, ["start", "test-tunnel"])
first_pid = read_pid_file(config_dir, "test-tunnel")[0]
result = runner.invoke(cli, ["restart", "test-tunnel"])
assert result.exit_code == 0, result.output
second_pid = read_pid_file(config_dir, "test-tunnel")[0]
assert second_pid != first_pid
assert is_process_alive(second_pid)
os.kill(second_pid, signal.SIGKILL)
def test_idempotent_start(runner, config_dir, tunnel, fake_autossh):
runner.invoke(cli, ["start", "test-tunnel"])
first_pid = read_pid_file(config_dir, "test-tunnel")[0]
result = runner.invoke(cli, ["start", "test-tunnel"])
assert result.exit_code == 0
assert "already running" in result.output
assert read_pid_file(config_dir, "test-tunnel")[0] == first_pid
os.kill(first_pid, signal.SIGKILL)
def test_idempotent_stop(runner, tunnel):
result = runner.invoke(cli, ["stop", "test-tunnel"])
assert result.exit_code == 0
assert "not running" in result.output