test: add integration tests for full tunnel lifecycle
This commit is contained in:
@@ -2,10 +2,11 @@ import os
|
|||||||
import signal
|
import signal
|
||||||
import time
|
import time
|
||||||
import pytest
|
import pytest
|
||||||
|
from datetime import datetime, timezone
|
||||||
from click.testing import CliRunner
|
from click.testing import CliRunner
|
||||||
from autossh_mgr.cli import cli
|
from autossh_mgr.cli import cli
|
||||||
from autossh_mgr.config import save_tunnels, TunnelConfig, ensure_dirs
|
from autossh_mgr.config import save_tunnels, TunnelConfig, ensure_dirs
|
||||||
from autossh_mgr.process import read_pid_file, is_process_alive
|
from autossh_mgr.process import read_pid_file, is_process_alive, write_pid_file
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
@@ -29,6 +30,22 @@ def set_config_dir(config_dir, monkeypatch):
|
|||||||
monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
|
monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def cleanup_processes(config_dir):
|
||||||
|
"""Kill any processes recorded in PID files after each test."""
|
||||||
|
yield
|
||||||
|
pids_dir = config_dir / "pids"
|
||||||
|
if not pids_dir.exists():
|
||||||
|
return
|
||||||
|
for pid_file in pids_dir.glob("*.pid"):
|
||||||
|
try:
|
||||||
|
lines = pid_file.read_text().splitlines()
|
||||||
|
if lines:
|
||||||
|
os.kill(int(lines[0]), signal.SIGKILL)
|
||||||
|
except (ValueError, OSError, ProcessLookupError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def tunnel(config_dir):
|
def tunnel(config_dir):
|
||||||
t = TunnelConfig(
|
t = TunnelConfig(
|
||||||
@@ -52,10 +69,8 @@ def test_start_writes_pid_file(runner, config_dir, tunnel, fake_autossh):
|
|||||||
assert result.exit_code == 0, result.output
|
assert result.exit_code == 0, result.output
|
||||||
pid_data = read_pid_file(config_dir, "test-tunnel")
|
pid_data = read_pid_file(config_dir, "test-tunnel")
|
||||||
assert pid_data is not None
|
assert pid_data is not None
|
||||||
pid, started_at = pid_data
|
pid, _ = pid_data
|
||||||
assert is_process_alive(pid)
|
assert is_process_alive(pid)
|
||||||
# cleanup
|
|
||||||
os.kill(pid, signal.SIGKILL)
|
|
||||||
|
|
||||||
|
|
||||||
def test_stop_kills_process(runner, config_dir, tunnel, fake_autossh):
|
def test_stop_kills_process(runner, config_dir, tunnel, fake_autossh):
|
||||||
@@ -78,14 +93,8 @@ def test_status_shows_running(runner, config_dir, tunnel, fake_autossh):
|
|||||||
assert result.exit_code == 0
|
assert result.exit_code == 0
|
||||||
assert "running" in result.output
|
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):
|
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))
|
write_pid_file(config_dir, "test-tunnel", 9999999, datetime.now(timezone.utc))
|
||||||
|
|
||||||
result = runner.invoke(cli, ["start", "test-tunnel"])
|
result = runner.invoke(cli, ["start", "test-tunnel"])
|
||||||
@@ -96,8 +105,6 @@ def test_stale_pid_cleaned_on_start(runner, config_dir, tunnel, fake_autossh):
|
|||||||
assert pid_data is not None
|
assert pid_data is not None
|
||||||
assert pid_data[0] != 9999999
|
assert pid_data[0] != 9999999
|
||||||
|
|
||||||
os.kill(pid_data[0], signal.SIGKILL)
|
|
||||||
|
|
||||||
|
|
||||||
def test_restart(runner, config_dir, tunnel, fake_autossh):
|
def test_restart(runner, config_dir, tunnel, fake_autossh):
|
||||||
runner.invoke(cli, ["start", "test-tunnel"])
|
runner.invoke(cli, ["start", "test-tunnel"])
|
||||||
@@ -109,8 +116,7 @@ def test_restart(runner, config_dir, tunnel, fake_autossh):
|
|||||||
second_pid = read_pid_file(config_dir, "test-tunnel")[0]
|
second_pid = read_pid_file(config_dir, "test-tunnel")[0]
|
||||||
assert second_pid != first_pid
|
assert second_pid != first_pid
|
||||||
assert is_process_alive(second_pid)
|
assert is_process_alive(second_pid)
|
||||||
|
assert not is_process_alive(first_pid)
|
||||||
os.kill(second_pid, signal.SIGKILL)
|
|
||||||
|
|
||||||
|
|
||||||
def test_idempotent_start(runner, config_dir, tunnel, fake_autossh):
|
def test_idempotent_start(runner, config_dir, tunnel, fake_autossh):
|
||||||
@@ -122,8 +128,6 @@ def test_idempotent_start(runner, config_dir, tunnel, fake_autossh):
|
|||||||
assert "already running" in result.output
|
assert "already running" in result.output
|
||||||
assert read_pid_file(config_dir, "test-tunnel")[0] == first_pid
|
assert read_pid_file(config_dir, "test-tunnel")[0] == first_pid
|
||||||
|
|
||||||
os.kill(first_pid, signal.SIGKILL)
|
|
||||||
|
|
||||||
|
|
||||||
def test_idempotent_stop(runner, tunnel):
|
def test_idempotent_stop(runner, tunnel):
|
||||||
result = runner.invoke(cli, ["stop", "test-tunnel"])
|
result = runner.invoke(cli, ["stop", "test-tunnel"])
|
||||||
|
|||||||
Reference in New Issue
Block a user