129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
import pytest
|
|
from click.testing import CliRunner
|
|
from autossh_mgr.cli import cli
|
|
from autossh_mgr.config import load_tunnels, save_tunnels
|
|
|
|
|
|
@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()
|
|
|
|
|
|
def test_init_creates_structure(runner, tmp_path, monkeypatch):
|
|
new_dir = tmp_path / "new-config"
|
|
monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(new_dir))
|
|
result = runner.invoke(cli, ["init"])
|
|
assert result.exit_code == 0
|
|
assert (new_dir / "tunnels.yaml").exists()
|
|
assert (new_dir / "pids").is_dir()
|
|
assert (new_dir / "logs").is_dir()
|
|
|
|
|
|
def test_init_idempotent(runner):
|
|
result = runner.invoke(cli, ["init"])
|
|
assert result.exit_code == 0
|
|
result = runner.invoke(cli, ["init"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_add_non_interactive(runner, config_dir):
|
|
result = runner.invoke(cli, [
|
|
"add", "web-service",
|
|
"--host", "relay.example.com",
|
|
"--user", "deploy",
|
|
"--local-port", "8080",
|
|
"--remote-port", "18080",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
tunnels = load_tunnels(config_dir)
|
|
assert len(tunnels) == 1
|
|
assert tunnels[0].name == "web-service"
|
|
assert tunnels[0].host == "relay.example.com"
|
|
|
|
|
|
def test_add_duplicate_name_fails(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, [
|
|
"add", "web-service",
|
|
"--host", "other.com", "--user", "u",
|
|
"--local-port", "9090", "--remote-port", "19090",
|
|
])
|
|
assert result.exit_code != 0
|
|
assert "already exists" in result.output
|
|
|
|
|
|
def test_remove_non_running(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["remove", "web-service"], input="y\n")
|
|
assert result.exit_code == 0
|
|
assert load_tunnels(config_dir) == []
|
|
|
|
|
|
def test_remove_unknown_name(runner):
|
|
result = runner.invoke(cli, ["remove", "missing"], input="y\n")
|
|
assert result.exit_code != 0
|
|
assert "No tunnel named" in result.output
|
|
|
|
|
|
def test_remove_aborted(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["remove", "web-service"], input="n\n")
|
|
assert result.exit_code == 0
|
|
assert len(load_tunnels(config_dir)) == 1
|
|
|
|
|
|
def test_list_empty(runner):
|
|
result = runner.invoke(cli, ["list"])
|
|
assert result.exit_code == 0
|
|
assert "No tunnels" in result.output
|
|
|
|
|
|
def test_list_shows_tunnels(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["list"])
|
|
assert result.exit_code == 0
|
|
assert "web-service" in result.output
|
|
assert "relay.example.com" in result.output
|
|
|
|
|
|
def test_show_tunnel(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["show", "web-service"])
|
|
assert result.exit_code == 0
|
|
assert "relay.example.com" in result.output
|
|
assert "8080" in result.output
|
|
assert "18080" in result.output
|
|
|
|
|
|
def test_config_update_field(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["config", "web-service", "remote_port", "19000"])
|
|
assert result.exit_code == 0
|
|
t = load_tunnels(config_dir)[0]
|
|
assert t.remote_port == 19000
|
|
|
|
|
|
def test_config_unknown_key(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["config", "web-service", "bad_key", "val"])
|
|
assert result.exit_code != 0
|
|
assert "Unknown field" in result.output
|
|
|
|
|
|
def test_config_rename_disallowed(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["config", "web-service", "name", "new-name"])
|
|
assert result.exit_code != 0
|
|
assert "Cannot rename" in result.output
|
|
|
|
|
|
def test_config_invalid_type(runner, config_dir, sample_tunnel):
|
|
save_tunnels(config_dir, [sample_tunnel])
|
|
result = runner.invoke(cli, ["config", "web-service", "port", "notanumber"])
|
|
assert result.exit_code != 0
|