104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
import os
|
|
import signal
|
|
import time
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
from unittest.mock import patch, MagicMock
|
|
from autossh_mgr.cli import cli
|
|
from autossh_mgr.config import 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
|