diff --git a/tests/integration/test_lifecycle.py b/tests/integration/test_lifecycle.py new file mode 100644 index 0000000..7198671 --- /dev/null +++ b/tests/integration/test_lifecycle.py @@ -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