162 lines
5.6 KiB
Python
162 lines
5.6 KiB
Python
import os
|
|
import signal
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
from unittest.mock import patch, MagicMock
|
|
from autossh_mgr.cli import cli
|
|
from autossh_mgr.config import TunnelConfig, save_tunnels
|
|
from autossh_mgr.process import write_pid_file, read_pid_file
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def set_config_dir(config_dir, monkeypatch):
|
|
monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
|
|
|
|
|
|
@pytest.fixture
|
|
def runner():
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def with_tunnel(config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
return sample_tunnel
|
|
|
|
|
|
def test_start_already_running(runner, config_dir, with_tunnel):
|
|
write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc))
|
|
result = runner.invoke(cli, ["start", "web-service"])
|
|
assert result.exit_code == 0
|
|
assert "already running" in result.output
|
|
|
|
|
|
def test_start_stale_pid_cleaned(runner, config_dir, with_tunnel, tmp_path, monkeypatch):
|
|
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']}")
|
|
|
|
write_pid_file(config_dir, "web-service", 9999999, datetime.now(timezone.utc))
|
|
result = runner.invoke(cli, ["start", "web-service"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "stale" in result.output.lower()
|
|
# Clean up launched process
|
|
pid_data = read_pid_file(config_dir, "web-service")
|
|
if pid_data:
|
|
os.kill(pid_data[0], signal.SIGKILL)
|
|
|
|
|
|
def test_start_unknown_tunnel(runner):
|
|
result = runner.invoke(cli, ["start", "missing"])
|
|
assert result.exit_code != 0
|
|
assert "No tunnel named" in result.output
|
|
|
|
|
|
def test_stop_running_tunnel(runner, config_dir, with_tunnel, tmp_path, monkeypatch):
|
|
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']}")
|
|
|
|
runner.invoke(cli, ["start", "web-service"])
|
|
result = runner.invoke(cli, ["stop", "web-service"])
|
|
assert result.exit_code == 0
|
|
assert read_pid_file(config_dir, "web-service") is None
|
|
|
|
|
|
def test_stop_already_stopped(runner, with_tunnel):
|
|
result = runner.invoke(cli, ["stop", "web-service"])
|
|
assert result.exit_code == 0
|
|
assert "not running" in result.output
|
|
|
|
|
|
def test_status_all_stopped(runner, config_dir, with_tunnel):
|
|
result = runner.invoke(cli, ["status"])
|
|
assert result.exit_code == 0
|
|
assert "web-service" in result.output
|
|
assert "stopped" in result.output
|
|
|
|
|
|
def test_status_single(runner, config_dir, with_tunnel):
|
|
result = runner.invoke(cli, ["status", "web-service"])
|
|
assert result.exit_code == 0
|
|
assert "stopped" in result.output
|
|
assert "relay.example.com" in result.output
|
|
|
|
|
|
def test_check_success(runner, with_tunnel):
|
|
with patch("autossh_mgr.check.subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
|
result = runner.invoke(cli, ["check", "web-service"])
|
|
assert result.exit_code == 0
|
|
assert "OK" in result.output or "success" in result.output.lower()
|
|
|
|
|
|
def test_check_failure(runner, with_tunnel):
|
|
with patch("autossh_mgr.check.subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=255, stderr="Connection refused")
|
|
result = runner.invoke(cli, ["check", "web-service"])
|
|
assert result.exit_code != 0
|
|
assert "Connection refused" in result.output
|
|
|
|
|
|
def test_restart_running_tunnel(runner, config_dir, with_tunnel, tmp_path, monkeypatch):
|
|
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']}")
|
|
|
|
runner.invoke(cli, ["start", "web-service"])
|
|
first_pid = read_pid_file(config_dir, "web-service")[0]
|
|
|
|
result = runner.invoke(cli, ["restart", "web-service"])
|
|
assert result.exit_code == 0, result.output
|
|
|
|
second_pid_data = read_pid_file(config_dir, "web-service")
|
|
assert second_pid_data is not None
|
|
assert second_pid_data[0] != first_pid
|
|
|
|
os.kill(second_pid_data[0], signal.SIGKILL)
|
|
|
|
|
|
def test_start_all(runner, config_dir, tmp_path, monkeypatch):
|
|
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']}")
|
|
|
|
t1 = TunnelConfig(name="alpha", host="h.com", user="u", local_port=8001, remote_port=18001)
|
|
t2 = TunnelConfig(name="beta", host="h.com", user="u", local_port=8002, remote_port=18002)
|
|
save_tunnels(config_dir, [t1, t2])
|
|
|
|
result = runner.invoke(cli, ["start"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "alpha" in result.output
|
|
assert "beta" in result.output
|
|
|
|
for name in ["alpha", "beta"]:
|
|
pid_data = read_pid_file(config_dir, name)
|
|
if pid_data:
|
|
os.kill(pid_data[0], signal.SIGKILL)
|
|
|
|
|
|
def test_stop_all(runner, config_dir, tmp_path, monkeypatch):
|
|
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']}")
|
|
|
|
t1 = TunnelConfig(name="alpha", host="h.com", user="u", local_port=8001, remote_port=18001)
|
|
t2 = TunnelConfig(name="beta", host="h.com", user="u", local_port=8002, remote_port=18002)
|
|
save_tunnels(config_dir, [t1, t2])
|
|
|
|
runner.invoke(cli, ["start"])
|
|
result = runner.invoke(cli, ["stop"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "alpha" in result.output
|
|
assert "beta" in result.output
|
|
assert read_pid_file(config_dir, "alpha") is None
|
|
assert read_pid_file(config_dir, "beta") is None
|